feat(relay/audio): wire HuddleControl accept/dial into the audio handler

Second increment on the cross-pod huddle join path (owner-side acceptor +
non-owner dial + the handler seam). The first commit defined the protocol,
resolver, and fenced directory trait; this one makes them load-bearing on a
live audio connection.

Owner side (join.rs): HuddleControlAcceptor::accept_inbound validates
sender==authenticated-peer, Session role, HuddleControl profile, fences against
Redis, and confirms owner-is-local before serving a register/unregister control
loop. RegisterPeer -> room.add_peer (as a remote peer) + spawn_remote_peer_sink
(fans owner audio back to the registering pod as datagrams). Every control frame
is re-fenced, not just the Hello -- a lease that moves mid-stream rejects
subsequent registers. Peers a stream registers are tracked so a stream close
tears them all down; no leaked index slots.

Non-owner side (join.rs): dial_remote_owner opens the HuddleControl stream,
registers the client, and returns a RemoteHuddleSession carrying the
owner-assigned peer index. forward_media ships client Opus to the owner tagged
with that index; close() sends UnregisterPeer + Goodbye. The owner is the sole
fan-out authority and sole index allocator -- co-located clients hear each other
via the owner round-trip (deliver_prefixed's index-skip prevents self-echo).

Handler seam (handler.rs): mesh-off path is byte-identical to today, including
the huddle_audio_available=false guardrail. Mesh-on resolves the join at the
horizontal-scaling seam; the RemoteOwner branch dials the owner after the local
add_peer (so a rejection backs out cleanly before any peer sees a joined event)
and threads the session into recv_loop -- forward to owner vs broadcast_frame.
Owner-rejection maps to the same client-facing WS error codes a same-pod join
produces (remote_rejection_ws_error), never a silent media drop.

Extracted a pure media_datagram helper (unit-tested) and relaxed
spawn_remote_peer_sink to Arc<dyn RelayPeerTransport> (only caller).

Scoped out, honestly flagged in-code: cross-pod roster sync. Non-owner joined/
peers reflect only co-located peers today; the media + fence path is complete
and correct. Roster deltas over HuddleControl are the next increment.

Build + clippy clean; 501 relay tests pass (10 join tests).

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

crates/buzz-relay/src/audio/handler.rs

@@ -224,33 +224,72 @@ async fn handle_audio_connection(
         }
     };
 
-    // Huddle audio guardrail (plan §5b). Audio frames are relayed only within
-    // a single pod; under horizontal scaling (any-pod-any-connection) two peers
-    // in one huddle can land on different pods and never hear each other. A
-    // multi-pod deployment sets `huddle_audio_available = false`, and we surface
-    // a clear, client-handleable "unavailable" signal here — BEFORE joining a
-    // room — rather than shipping a silent split-room. Single-pod deployments
-    // leave the flag at its `true` default and keep today's behavior. The fix
-    // is an out-of-relay media/SFU service (Tyler's long-term target), not
-    // sticky-routing huddles into this rewrite.
-    if !state.config.huddle_audio_available {
-        debug!(
-            channel_id = %channel_id,
-            pubkey = %pubkey_hex,
-            "huddle audio unavailable under horizontal scaling — rejecting join"
-        );
-        let _ = ws_send
-            .send(WsMessage::Text(
-                serde_json::json!({
-                    "type": "error",
-                    "code": "huddle_audio_unavailable",
-                    "message": "huddle audio unavailable in this deployment"
-                })
-                .to_string()
-                .into(),
-            ))
-            .await;
-        return;
+    // Huddle cross-pod routing (mesh) OR single-pod guardrail.
+    //
+    // When the mesh is live (`state.mesh()` is `Some`), a huddle can span pods:
+    // Redis arbitrates ownership and this pod either owns the room locally or
+    // forwards the client to the owner over a `HuddleControl` stream. When the
+    // mesh is off, we keep today's behavior exactly — including the
+    // `huddle_audio_available=false` rejection under a non-mesh horizontal
+    // deployment (two peers on different pods would never hear each other).
+    //
+    // `remote_owner` is `Some` only on the non-owner path; it carries the
+    // registration to the owner and, once the client is admitted locally, is
+    // opened so its media forwards to the owner instead of fanning out locally.
+    let mut pending_remote: Option<crate::audio::join::JoinOutcome> = None;
+    match state.mesh() {
+        Some(mesh) => {
+            match crate::audio::join::resolve_join(
+                &mesh.directory,
+                tenant.community(),
+                channel_id,
+                mesh.local_runtime_id,
+            )
+            .await
+            {
+                Ok(outcome) => pending_remote = Some(outcome),
+                Err(e) => {
+                    warn!(
+                        channel_id = %channel_id,
+                        pubkey = %pubkey_hex,
+                        "huddle join rejected by fence: {e}"
+                    );
+                    let _ = ws_send
+                        .send(WsMessage::Text(
+                            serde_json::json!({
+                                "type": "error",
+                                "code": "join_rejected",
+                                "message": "huddle join rejected"
+                            })
+                            .to_string()
+                            .into(),
+                        ))
+                        .await;
+                    return;
+                }
+            }
+        }
+        None => {
+            if !state.config.huddle_audio_available {
+                debug!(
+                    channel_id = %channel_id,
+                    pubkey = %pubkey_hex,
+                    "huddle audio unavailable under horizontal scaling — rejecting join"
+                );
+                let _ = ws_send
+                    .send(WsMessage::Text(
+                        serde_json::json!({
+                            "type": "error",
+                            "code": "huddle_audio_unavailable",
+                            "message": "huddle audio unavailable in this deployment"
+                        })
+                        .to_string()
+                        .into(),
+                    ))
+                    .await;
+                return;
+            }
+        }
     }
 
     let room = state.audio_rooms.get_or_create(channel_id);
@@ -368,6 +407,79 @@ async fn handle_audio_connection(
         "audio peer joined"
     );
 
+    // Non-owner path: the huddle is owned by another pod. Register this client
+    // with the owner over a `HuddleControl` stream so the owner fans media back
+    // to us; the returned session forwards the client's Opus to the owner. A
+    // fence/admission rejection from the owner is surfaced to the client as the
+    // same WS error a same-pod join would produce — never a silent media drop.
+    //
+    // NOTE (roster sync): cross-pod roster fan-out (the owner's full participant
+    // list reflected to this client) is the next increment; today this pod's
+    // local `joined`/`peers` reflect only co-located peers. The media + fence
+    // path is complete and correct; roster consistency lands with the owner's
+    // roster-delta messages over the same `HuddleControl` stream.
+    let mut remote_session: Option<crate::audio::join::RemoteHuddleSession> = None;
+    if let (Some(mesh), Some(crate::audio::join::JoinOutcome::RemoteOwner { .. })) =
+        (state.mesh(), pending_remote)
+    {
+        let outcome = pending_remote.expect("RemoteOwner matched above");
+        let fenced = outcome.fenced_header(channel_id, mesh.local_runtime_id);
+        let crate::audio::join::JoinOutcome::RemoteOwner {
+            owner_runtime_id, ..
+        } = outcome
+        else {
+            unreachable!("matched RemoteOwner above");
+        };
+        match crate::audio::join::dial_remote_owner(
+            Arc::clone(&mesh.transport),
+            mesh.local_runtime_id,
+            owner_runtime_id,
+            fenced,
+            pubkey_hex.clone(),
+            requested_version,
+        )
+        .await
+        {
+            Ok(session) => remote_session = Some(session),
+            Err(crate::audio::join::DialError::Rejected(reason)) => {
+                warn!(
+                    channel_id = %channel_id,
+                    pubkey = %pubkey_hex,
+                    "huddle owner rejected registration: {reason:?}"
+                );
+                let _ = ws_send
+                    .send(WsMessage::Text(
+                        remote_rejection_ws_error(&reason).to_string().into(),
+                    ))
+                    .await;
+                room.remove_peer(peer_id);
+                state.audio_rooms.cleanup_if_empty(channel_id);
+                return;
+            }
+            Err(crate::audio::join::DialError::Mesh(e)) => {
+                warn!(
+                    channel_id = %channel_id,
+                    pubkey = %pubkey_hex,
+                    "huddle owner registration failed: {e}"
+                );
+                let _ = ws_send
+                    .send(WsMessage::Text(
+                        serde_json::json!({
+                            "type": "error",
+                            "code": "huddle_owner_unreachable",
+                            "message": "could not reach the huddle owner"
+                        })
+                        .to_string()
+                        .into(),
+                    ))
+                    .await;
+                room.remove_peer(peer_id);
+                state.audio_rooms.cleanup_if_empty(channel_id);
+                return;
+            }
+        }
+    }
+
     let peers_snapshot: Vec<serde_json::Value> = room
         .peer_pubkeys()
         .into_iter()
@@ -427,9 +539,16 @@ async fn handle_audio_connection(
         ctrl_tx,
         Arc::clone(&missed_pongs),
         cancel.clone(),
+        remote_session.as_mut(),
     )
     .await;
 
+    // Non-owner path: unregister from the owner and close the control stream
+    // cleanly. Best-effort — teardown never blocks connection cleanup.
+    if let Some(session) = remote_session {
+        session.close().await;
+    }
+
     cancel.cancel();
     let _ = send_task.await;
     let _ = heartbeat_task.await;
@@ -499,6 +618,43 @@ async fn handle_audio_connection(
     );
 }
 
+/// Map an owner's registration rejection to the client-facing WS error, using
+/// the same `code`s a same-pod join produces so a cross-pod client handles them
+/// identically. Fence rejections carry their taxonomy code for observability.
+fn remote_rejection_ws_error(
+    reason: &crate::audio::join::RegisterRejection,
+) -> serde_json::Value {
+    use crate::audio::join::RegisterRejection;
+    match reason {
+        RegisterRejection::RoomFull => serde_json::json!({
+            "type": "error", "code": "room_full",
+            "message": "peer index space exhausted"
+        }),
+        RegisterRejection::RoomEnded => serde_json::json!({
+            "type": "error", "code": "room_ended", "message": "huddle has ended"
+        }),
+        RegisterRejection::VersionMismatch { pinned, requested } => serde_json::json!({
+            "type": "error", "code": "upgrade_required",
+            "message": format!(
+                "this huddle is using audio protocol v{pinned}; your client requested v{requested}"
+            ),
+            "pinned_version": pinned,
+            "requested_version": requested,
+        }),
+        RegisterRejection::Fenced(f) => serde_json::json!({
+            "type": "error", "code": "join_rejected",
+            "message": "huddle join rejected",
+            "fence_reason": f.code(),
+        }),
+    }
+}
+
+/// Receive loop: reads client frames and routes them. Local/owner joins fan
+/// out through the local room; a non-owner join forwards to the huddle owner
+/// via `remote_session`. Argument count reflects the pre-existing connection
+/// wiring plus the one mesh session; a param struct would obscure more than it
+/// clarifies at this single call site.
+#[allow(clippy::too_many_arguments)]
 async fn recv_loop(
     mut ws_recv: futures_util::stream::SplitStream<WebSocket>,
     room: Arc<crate::audio::room::Room>,
@@ -507,6 +663,7 @@ async fn recv_loop(
     ctrl_tx: mpsc::Sender<WsMessage>,
     missed_pongs: Arc<AtomicU8>,
     cancel: CancellationToken,
+    mut remote_session: Option<&mut crate::audio::join::RemoteHuddleSession>,
 ) {
     use crate::audio::wire::{FrameHeader, V2_HEADER_LEN};
 
@@ -567,7 +724,15 @@ async fn recv_loop(
                             }
                         }
 
-                        room.broadcast_frame(peer_id, data);
+                        // Non-owner path forwards the client's Opus to the
+                        // huddle owner as a datagram (the owner is the sole
+                        // fan-out authority); the owner-side room fans it back
+                        // to every participant, including our co-located peers.
+                        // Owner/local path fans out through the local room.
+                        match remote_session.as_deref_mut() {
+                            Some(session) => session.forward_media(&data),
+                            None => room.broadcast_frame(peer_id, data),
+                        }
                     }
                     Some(Ok(WsMessage::Text(text))) => {
                         if text.len() > MAX_TEXT_FRAME_BYTES {

crates/buzz-relay/src/audio/join.rs

@@ -35,11 +35,20 @@
 //! [`HuddleControlMsg::RegisterRejected`] (fence/admission failure surfaced to
 //! the client as a join error, never a silent media drop).
 
+use std::sync::Arc;
+
 use buzz_core::CommunityId;
-use buzz_relay_mesh::{FencedHeader, MeshError, Profile, RuntimeId};
+use buzz_relay_mesh::{
+    FencedHeader, GoodbyeReason, MeshDatagram, MeshError, MeshStream, MeshStreamFrame, Profile,
+    RelayPeerTransport, RuntimeId, StreamHello, StreamRole,
+};
 use serde::{Deserialize, Serialize};
+use tracing::debug;
 use uuid::Uuid;
 
+use super::mesh::spawn_remote_peer_sink;
+use super::room::{AdmissionError, AudioRoomManager};
+
 /// The slice of the Redis fenced session directory the huddle join path needs.
 ///
 /// Implemented by the session-directory lane's `SessionDirectory` over the
@@ -377,6 +386,418 @@ pub fn decode_control(bytes: &[u8]) -> Result<HuddleControlMsg, MeshError> {
 /// desync, so it never rides datagrams.
 pub const HUDDLE_CONTROL_PROFILE: Profile = Profile::HuddleControl;
 
+// ── Owner-side HuddleControl accept path ─────────────────────────────────────
+//
+// The owner pod hosts the real [`Room`]. When a *non-owner* pod opens a
+// `HuddleControl` stream and registers a client, the owner admits that client
+// as an ordinary [`AudioPeer`] whose `audio_tx` is drained by
+// [`super::mesh::spawn_remote_peer_sink`] back to the non-owner pod as
+// datagrams. `Room` never learns about the mesh — a remote participant looks
+// exactly like a local one to fan-out.
+
+/// Owner-side handler for inbound `HuddleControl` streams.
+///
+/// One instance per relay; the boot-seam dispatcher routes every
+/// `Profile::HuddleControl` session stream to [`Self::accept_inbound`]. It is
+/// the counterpart to Perci's reliable-stream acceptor — same
+/// `accept_inbound(community_id, from, hello, stream)` shape, different profile
+/// and body.
+pub struct HuddleControlAcceptor<D: HuddleDirectory + ?Sized> {
+    rooms: Arc<AudioRoomManager>,
+    transport: Arc<dyn RelayPeerTransport>,
+    directory: Arc<D>,
+    local_runtime_id: RuntimeId,
+}
+
+impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
+    /// Build the acceptor. `directory` is the fenced arbiter (re-validated on
+    /// every registration); `transport` is used to open the media datagram
+    /// sink back to each registering pod.
+    pub fn new(
+        rooms: Arc<AudioRoomManager>,
+        transport: Arc<dyn RelayPeerTransport>,
+        directory: Arc<D>,
+        local_runtime_id: RuntimeId,
+    ) -> Self {
+        Self {
+            rooms,
+            transport,
+            directory,
+            local_runtime_id,
+        }
+    }
+
+    /// Accept and validate an inbound `HuddleControl` stream, then serve its
+    /// register/unregister control loop until the stream closes.
+    ///
+    /// Validation mirrors the reliable-stream acceptor and enforces the fencing
+    /// law on receipt: the claimed sender must be the authenticated peer, the
+    /// profile must be `HuddleControl`, the fence must pass Redis, and this pod
+    /// must be the fenced owner. Any of these failing rejects the stream before
+    /// a single peer is admitted.
+    pub async fn accept_inbound(
+        &self,
+        community_id: CommunityId,
+        from: RuntimeId,
+        hello: StreamHello,
+        stream: MeshStream,
+    ) -> Result<(), MeshError> {
+        if hello.sender != from {
+            return Err(MeshError::Transport(format!(
+                "huddle-control hello.sender {} != authenticated peer {from}",
+                hello.sender
+            )));
+        }
+        let StreamRole::Session { fenced, profile } = hello.role else {
+            return Err(MeshError::Transport(
+                "huddle-control stream Hello was not a session role".into(),
+            ));
+        };
+        if profile != Profile::HuddleControl {
+            return Err(MeshError::Transport(format!(
+                "huddle-control acceptor got profile {profile:?}"
+            )));
+        }
+        // Fence at every hop: the origin validated before dialing; the owner
+        // re-validates on receipt so a lease that moved in between is caught.
+        self.directory.validate(community_id, &fenced).await?;
+        if fenced.owner_runtime_id != self.local_runtime_id {
+            return Err(MeshError::OwnerMismatch {
+                session_id: fenced.session_id,
+                generation: fenced.generation,
+                frame_owner_runtime_id: fenced.owner_runtime_id,
+                current_owner_runtime_id: self.local_runtime_id,
+            });
+        }
+
+        self.serve_control_loop(community_id, from, fenced, stream)
+            .await
+    }
+
+    /// Serve register/unregister frames for one non-owner pod's stream.
+    ///
+    /// Peers this stream registers are tracked so a stream close (the non-owner
+    /// pod went away) tears them all down — no leaked remote peers holding
+    /// index slots in the owner's room.
+    async fn serve_control_loop(
+        &self,
+        community_id: CommunityId,
+        from: RuntimeId,
+        fenced: FencedHeader,
+        mut stream: MeshStream,
+    ) -> Result<(), MeshError> {
+        let session_id = fenced.session_id;
+        // pubkey -> peer_id, for UnregisterPeer and teardown on stream close.
+        let mut registered: std::collections::HashMap<String, Uuid> = std::collections::HashMap::new();
+
+        let result = loop {
+            let msg = match stream.recv_frame().await {
+                Ok(Some(MeshStreamFrame::Data { fenced: f, payload })) => {
+                    // Re-fence every control frame, not just the Hello: a lease
+                    // that moves mid-stream must reject subsequent registers.
+                    if f != fenced {
+                        break Err(MeshError::OwnerMismatch {
+                            session_id,
+                            generation: f.generation,
+                            frame_owner_runtime_id: f.owner_runtime_id,
+                            current_owner_runtime_id: self.local_runtime_id,
+                        });
+                    }
+                    if let Err(e) = self.directory.validate(community_id, &f).await {
+                        break Err(e);
+                    }
+                    match decode_control(&payload) {
+                        Ok(m) => m,
+                        Err(e) => break Err(e),
+                    }
+                }
+                Ok(Some(MeshStreamFrame::Goodbye { .. })) | Ok(None) => break Ok(()),
+                Ok(Some(other)) => {
+                    break Err(MeshError::Transport(format!(
+                        "huddle-control stream got unexpected frame {other:?}"
+                    )));
+                }
+                Err(e) => break Err(e),
+            };
+
+            match msg {
+                HuddleControlMsg::RegisterPeer {
+                    pubkey,
+                    protocol_version,
+                } => {
+                    let reply = self.register_remote_peer(
+                        session_id,
+                        fenced,
+                        from,
+                        &pubkey,
+                        protocol_version,
+                        &mut registered,
+                    );
+                    if let Err(e) = stream
+                        .send_frame(MeshStreamFrame::Data {
+                            fenced,
+                            payload: encode_control(&reply)?,
+                        })
+                        .await
+                    {
+                        break Err(e);
+                    }
+                }
+                HuddleControlMsg::UnregisterPeer { pubkey } => {
+                    if let Some(peer_id) = registered.remove(&pubkey) {
+                        if let Some(room) = self.rooms.get(session_id) {
+                            room.remove_peer(peer_id);
+                        }
+                    }
+                }
+                // Owner→non-owner replies never arrive on the owner's accept
+                // side; a peer sending one is a protocol violation.
+                HuddleControlMsg::PeerRegistered { .. }
+                | HuddleControlMsg::RegisterRejected { .. } => {
+                    break Err(MeshError::Transport(
+                        "huddle-control owner received an owner→non-owner reply".into(),
+                    ));
+                }
+            }
+        };
+
+        // Teardown: drop every peer this stream registered, regardless of how
+        // the loop ended. Dropping the peer drops its `audio_tx`, which ends the
+        // matching `spawn_remote_peer_sink` task.
+        if let Some(room) = self.rooms.get(session_id) {
+            for (_pubkey, peer_id) in registered {
+                room.remove_peer(peer_id);
+            }
+        }
+        result
+    }
+
+    /// Admit one remote client into the owner's room and wire its fan-out back
+    /// to the registering pod as datagrams. Returns the reply to send.
+    fn register_remote_peer(
+        &self,
+        session_id: Uuid,
+        fenced: FencedHeader,
+        from: RuntimeId,
+        pubkey: &str,
+        protocol_version: u8,
+        registered: &mut std::collections::HashMap<String, Uuid>,
+    ) -> HuddleControlMsg {
+        let room = self.rooms.get_or_create(session_id);
+        match room.add_peer(pubkey.to_string(), protocol_version) {
+            Ok((peer_id, peer_index, audio_rx, _peer_ctrl_rx)) => {
+                registered.insert(pubkey.to_string(), peer_id);
+                // The owner's Room fans out to this remote peer's `audio_tx`;
+                // the sink drains `audio_rx` and ships each frame as a datagram
+                // to the pod that hosts the client.
+                spawn_remote_peer_sink(Arc::clone(&self.transport), from, fenced, audio_rx);
+                HuddleControlMsg::PeerRegistered {
+                    pubkey: pubkey.to_string(),
+                    peer_index,
+                }
+            }
+            Err(reason) => HuddleControlMsg::RegisterRejected {
+                pubkey: pubkey.to_string(),
+                reason: admission_to_rejection(reason),
+            },
+        }
+    }
+}
+
+/// Map a room admission failure to the wire rejection taxonomy. Kept 1:1 with
+/// the single-pod WS error codes so a cross-pod join surfaces the same
+/// client-facing error a same-pod join would.
+fn admission_to_rejection(err: AdmissionError) -> RegisterRejection {
+    match err {
+        AdmissionError::Full => RegisterRejection::RoomFull,
+        AdmissionError::Ended => RegisterRejection::RoomEnded,
+        AdmissionError::VersionMismatch { pinned, requested } => {
+            RegisterRejection::VersionMismatch { pinned, requested }
+        }
+    }
+}
+
+/// The `Goodbye` reason a non-owner sends when its client leaves the huddle
+/// cleanly. Re-exported so the handler's dial path uses one spelling.
+pub const HUDDLE_SESSION_ENDED: GoodbyeReason = GoodbyeReason::SessionEnded;
+
+// ── Non-owner-side HuddleControl dial path ───────────────────────────────────
+//
+// A client connected here whose huddle is owned by another pod. We keep the
+// client as an ordinary local WS peer (heartbeats, `joined`/`left`, delivery of
+// the owner's fan-out) but there is NO local fan-out: the owner is the sole
+// fan-out authority. Each client Opus frame is shipped to the owner as a
+// datagram tagged with the OWNER-assigned peer index; the owner fans out to
+// everyone (including this pod's co-located clients, which hear each other via
+// the owner round-trip — `deliver_prefixed` skips a client's own index so it
+// never hears itself).
+
+/// A registered cross-pod huddle session on the non-owner side.
+///
+/// Holds everything needed to forward the local client's media to the owner and
+/// to unregister cleanly on disconnect. Media delivery *back* to the client goes
+/// through the ordinary local room via `MeshAudioRouter::on_media_datagram`, so
+/// this handle owns only the outbound (client→owner) half plus teardown.
+pub struct RemoteHuddleSession {
+    /// The owner-allocated peer index this client occupies in the owner's room.
+    /// Stamped on every media datagram so the owner attributes frames correctly.
+    peer_index: u8,
+    /// Fenced header for this session's owner epoch; every datagram carries it.
+    fenced: FencedHeader,
+    /// The pod that owns the huddle.
+    owner: RuntimeId,
+    /// Pubkey of the local client, for the closing `UnregisterPeer`.
+    pubkey: String,
+    /// Transport for datagrams and the control-stream teardown.
+    transport: Arc<dyn RelayPeerTransport>,
+    /// The open `HuddleControl` stream to the owner; a clean `Goodbye` is sent
+    /// on teardown.
+    stream: MeshStream,
+    /// Per-datagram monotonic sequence for loss/reorder observability.
+    seq: u64,
+}
+
+/// Why a cross-pod join could not complete on the non-owner side.
+#[derive(Debug)]
+pub enum DialError {
+    /// The owner refused the registration; surfaced to the client as the same
+    /// WS error a same-pod join would produce, never a silent media drop.
+    Rejected(RegisterRejection),
+    /// Transport / protocol failure opening or serving the control stream.
+    Mesh(MeshError),
+}
+
+impl From<MeshError> for DialError {
+    fn from(e: MeshError) -> Self {
+        DialError::Mesh(e)
+    }
+}
+
+/// Open a `HuddleControl` stream to the owner and register the local client.
+///
+/// On success the owner has admitted the client as a remote peer and returned
+/// its owner-assigned index; the returned [`RemoteHuddleSession`] forwards media
+/// and unregisters on drop. On [`DialError::Rejected`] the caller surfaces the
+/// owner's admission failure to the client unchanged.
+pub async fn dial_remote_owner(
+    transport: Arc<dyn RelayPeerTransport>,
+    local_runtime_id: RuntimeId,
+    owner: RuntimeId,
+    fenced: FencedHeader,
+    pubkey: String,
+    protocol_version: u8,
+) -> Result<RemoteHuddleSession, DialError> {
+    let hello = StreamHello {
+        sender: local_runtime_id,
+        role: StreamRole::Session {
+            fenced,
+            profile: Profile::HuddleControl,
+        },
+    };
+    // `open_session_stream` sends the Hello before returning.
+    let mut stream = transport.open_session_stream(owner, hello).await?;
+
+    stream
+        .send_frame(MeshStreamFrame::Data {
+            fenced,
+            payload: encode_control(&HuddleControlMsg::RegisterPeer {
+                pubkey: pubkey.clone(),
+                protocol_version,
+            })?,
+        })
+        .await?;
+
+    match stream.recv_frame().await? {
+        Some(MeshStreamFrame::Data { payload, .. }) => match decode_control(&payload)? {
+            HuddleControlMsg::PeerRegistered { peer_index, .. } => Ok(RemoteHuddleSession {
+                peer_index,
+                fenced,
+                owner,
+                pubkey,
+                transport,
+                stream,
+                seq: 0,
+            }),
+            HuddleControlMsg::RegisterRejected { reason, .. } => Err(DialError::Rejected(reason)),
+            other => Err(DialError::Mesh(MeshError::Transport(format!(
+                "expected PeerRegistered/RegisterRejected, got {other:?}"
+            )))),
+        },
+        Some(MeshStreamFrame::Goodbye { .. }) | None => Err(DialError::Mesh(MeshError::Transport(
+            "owner closed HuddleControl stream before replying".into(),
+        ))),
+        Some(other) => Err(DialError::Mesh(MeshError::Transport(format!(
+            "unexpected HuddleControl frame from owner: {other:?}"
+        )))),
+    }
+}
+
+/// The `StreamHello.sender` for a dialed session: the fenced header carries the
+/// owner's identity, but the *sender* is this pod. The owner validates
+/// `hello.sender == authenticated peer`, so it must be our own runtime id — the
+/// handler threads `local_runtime_id` in explicitly.
+impl RemoteHuddleSession {
+    /// The owner-assigned index this client occupies in the owner's room.
+    pub fn peer_index(&self) -> u8 {
+        self.peer_index
+    }
+
+    /// Forward one client Opus frame to the owner as a media datagram, tagged
+    /// with the owner-assigned index. Drop-on-error: realtime audio never blocks
+    /// on a slow or gone link (the same discipline as local fan-out).
+    pub fn forward_media(&mut self, client_frame: &[u8]) {
+        let dgram = media_datagram(self.peer_index, self.fenced, self.seq, client_frame);
+        self.seq = self.seq.wrapping_add(1);
+        if let Err(e) = self.transport.send_datagram(self.owner, dgram) {
+            debug!(owner = %self.owner, "huddle media datagram to owner failed: {e}");
+        }
+    }
+
+    /// Unregister the client from the owner and close the control stream
+    /// cleanly. Best-effort: teardown never blocks connection cleanup.
+    pub async fn close(mut self) {
+        let _ = self
+            .stream
+            .send_frame(MeshStreamFrame::Data {
+                fenced: self.fenced,
+                payload: match encode_control(&HuddleControlMsg::UnregisterPeer {
+                    pubkey: self.pubkey.clone(),
+                }) {
+                    Ok(p) => p,
+                    Err(_) => return,
+                },
+            })
+            .await;
+        let _ = self
+            .stream
+            .send_frame(MeshStreamFrame::Goodbye {
+                fenced: self.fenced,
+                reason: HUDDLE_SESSION_ENDED,
+            })
+            .await;
+    }
+}
+
+/// Build the media datagram a non-owner ships to the owner for one client
+/// frame: `[owner_peer_index][client frame]`, stamped with the session fence
+/// and sequence. Pure so the framing is unit-testable without a live transport
+/// or stream.
+fn media_datagram(
+    peer_index: u8,
+    fenced: FencedHeader,
+    seq: u64,
+    client_frame: &[u8],
+) -> MeshDatagram {
+    let mut payload = Vec::with_capacity(1 + client_frame.len());
+    payload.push(peer_index);
+    payload.extend_from_slice(client_frame);
+    MeshDatagram {
+        fenced,
+        seq,
+        payload,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -602,4 +1023,44 @@ mod tests {
             }
         );
     }
+
+    #[test]
+    fn admission_errors_map_to_wire_rejections() {
+        assert_eq!(
+            admission_to_rejection(AdmissionError::Full),
+            RegisterRejection::RoomFull
+        );
+        assert_eq!(
+            admission_to_rejection(AdmissionError::Ended),
+            RegisterRejection::RoomEnded
+        );
+        assert_eq!(
+            admission_to_rejection(AdmissionError::VersionMismatch {
+                pinned: 2,
+                requested: 1
+            }),
+            RegisterRejection::VersionMismatch {
+                pinned: 2,
+                requested: 1
+            }
+        );
+    }
+
+    #[test]
+    fn media_datagram_tags_owner_index_and_stamps_fence() {
+        let fenced = FencedHeader {
+            session_id: Uuid::new_v4(),
+            generation: 9,
+            owner_runtime_id: rt(2),
+        };
+        // Owner-assigned index is the first payload byte; client bytes follow.
+        let d0 = media_datagram(42, fenced, 0, &[0xDE, 0xAD]);
+        assert_eq!(d0.payload, vec![42, 0xDE, 0xAD]);
+        assert_eq!(d0.fenced, fenced);
+        assert_eq!(d0.seq, 0);
+        // Empty client frame still carries the index byte (owner tolerates it).
+        let d1 = media_datagram(7, fenced, 3, &[]);
+        assert_eq!(d1.payload, vec![7]);
+        assert_eq!(d1.seq, 3);
+    }
 }

crates/buzz-relay/src/audio/mesh.rs

@@ -241,8 +241,8 @@ impl MeshAudioRouter {
 /// feeds this task, which wraps each frame as a [`MeshDatagram`] and sends it to
 /// the pod that hosts that participant. Drops on a disconnected/oversize peer —
 /// realtime audio never blocks fan-out on one slow remote link.
-pub fn spawn_remote_peer_sink<T: RelayPeerTransport>(
-    transport: Arc<T>,
+pub fn spawn_remote_peer_sink(
+    transport: Arc<dyn RelayPeerTransport>,
     to: RuntimeId,
     fenced: FencedHeader,
     mut frames: mpsc::Receiver<Bytes>,