Chat file attachments, stage 2: file-transfer network plane

Add the dedicated FILES_ALPN data plane that moves attachment bytes
off-gossip via direct QUIC streams.

- FileRouter (ProtocolHandler on the persistent router, mirroring
  AudioRouter): bound to the active session's Shared on join, cleared on
  leave. On an inbound stream it authenticates the peer via the ALPN
  handshake, gates on live room membership (reuses audio_sender_admitted,
  so a former member cannot pull files), reads exactly one 32-byte
  attachment id (bounded request read), and streams back the matching
  blob from the session serve store — or an empty body for an unknown id.
- Shared gains served_files (id -> bytes), populated by serve_attachment
  and cleared on leave.
- IrohTransport::serve_attachment + fetch_attachment (inherent methods;
  transport is used concretely). fetch dials the sender on FILES_ALPN
  (preferring a known full address), writes the id, and reads bounded by
  the descriptor's declared size, with a 30s connect/read timeout so a
  stalled sender can't hang the fetch.
- Register FILES_ALPN in the router; bind/clear file_router in lock-step
  with audio_router at every join/leave site.

Builds + clippy clean, 349 lib tests pass (plane is runtime I/O,
field-tested in stage 5).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/core/mod.rs

@@ -8,7 +8,7 @@ use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
 use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
 use crate::network::{
     NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket,
-    iroh_impl::{IrohTransport, AudioRouter},
+    iroh_impl::{IrohTransport, AudioRouter, FileRouter},
     gossip::IrohGossipState,
 };
 use crate::core::messages::{CoreCommand, UiEvent};
@@ -578,6 +578,9 @@ struct NetStack {
     /// The persistent inbound-audio handler on `router`; per-join we bind the
     /// active session's transport into it, and clear it on leave.
     audio_router: AudioRouter,
+    /// The persistent chat-file-transfer handler on `router`; bound/cleared in
+    /// lock-step with `audio_router` (same session lifecycle).
+    file_router: FileRouter,
     /// In-memory address book (ticket + gossip fed), shared with every session.
     memory_lookup: iroh::address_lookup::memory::MemoryLookup,
 }
@@ -689,6 +692,7 @@ async fn build_net_stack(
         .spawn(endpoint.clone());
 
     let audio_router = AudioRouter::new();
+    let file_router = FileRouter::new();
     // The friends presence listener (W7 B2) rides this same persistent router as a
     // third ALPN — it MUST be a handler here, not a standalone accept loop, since
     // the router owns endpoint.accept(). Policy (who we answer / what room we
@@ -696,6 +700,7 @@ async fn build_net_stack(
     let router = Router::builder(endpoint.clone())
         .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
         .accept(crate::protocol::AUDIO_ALPN, audio_router.clone())
+        .accept(crate::protocol::FILES_ALPN, file_router.clone())
         .accept(
             crate::presence_net::FRIENDS_ALPN,
             crate::presence_net::FriendsProtocol::new(friends_handler),
@@ -707,6 +712,7 @@ async fn build_net_stack(
         gossip,
         router,
         audio_router,
+        file_router,
         memory_lookup,
     })
 }
@@ -1071,6 +1077,7 @@ async fn run_core_loop(
                 if let Some(session) = active_session.take() {
                     session.shutdown(audio_backend.clone()).await;
                     net.audio_router.clear();
+                    net.file_router.clear();
                 }
                 *current_room.lock().unwrap() = None;
 
@@ -1093,6 +1100,7 @@ async fn run_core_loop(
                     crate::log_msg("Shutting down existing active session");
                     session.shutdown(audio_backend.clone()).await;
                     net.audio_router.clear();
+                    net.file_router.clear();
                 }
 
                 // If a network-mode / identity change was deferred while a call was
@@ -1149,6 +1157,7 @@ async fn run_core_loop(
                 // NetStack; the session just subscribes its topic below.
                 let transport = Arc::new(IrohTransport::new(endpoint.clone()));
                 net.audio_router.bind(&transport);
+                net.file_router.bind(&transport);
 
                 let room_state = Arc::new(IrohGossipState::new(
                     endpoint.clone(),
@@ -1193,6 +1202,7 @@ async fn run_core_loop(
                     crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
                     net.audio_router.clear();
+                    net.file_router.clear();
                     continue;
                 }
                 crate::log_msg("Joined room successfully via room_state");
@@ -1245,6 +1255,7 @@ async fn run_core_loop(
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
                     let _ = room_state.leave().await;
                     net.audio_router.clear();
+                    net.file_router.clear();
                     continue;
                 }
 
@@ -1257,6 +1268,7 @@ async fn run_core_loop(
                     let _ = audio_backend.stop();
                     let _ = room_state.leave().await;
                     net.audio_router.clear();
+                    net.file_router.clear();
                     continue;
                 }
 
@@ -1829,6 +1841,7 @@ async fn run_core_loop(
                     session.shutdown(audio_backend.clone()).await;
                     // Stop routing inbound audio links — the endpoint/router stay up.
                     net.audio_router.clear();
+                    net.file_router.clear();
                     // No longer in a gathering — friends who ping see us as just online.
                     *current_room.lock().unwrap() = None;
                     let _ = ui_tx.send(UiEvent::RoomLeft).await;

src/network/iroh_impl.rs

@@ -9,7 +9,8 @@ use std::collections::{HashMap, HashSet};
 use std::time::Duration;
 use async_trait::async_trait;
 
-use crate::protocol::AUDIO_ALPN;
+use crate::protocol::{AUDIO_ALPN, FILES_ALPN};
+use crate::files::{AttachmentId, ChatAttachment};
 
 /// Per-peer datagram send queue depth. Audio is real-time, so a backlog is
 /// useless latency — keep it shallow and drop the oldest frame when full.
@@ -31,6 +32,10 @@ const MAX_BACKOFF: Duration = Duration::from_secs(5);
 /// already means "the peer closed this on purpose."
 const GOODBYE_CODE: u32 = 1;
 
+/// Bound on each phase (connect, read) of a chat-attachment fetch, so a slow or
+/// stalled sender can't hang the fetch indefinitely.
+const FILE_FETCH_TIMEOUT: Duration = Duration::from_secs(30);
+
 /// State shared between the transport, its protocol handler, and every per-peer
 /// supervisor task. One supervisor owns a peer's whole connection lifecycle.
 struct Shared {
@@ -60,6 +65,11 @@ struct Shared {
     /// verified gossip roster plus peers still inside reconnect grace; transport
     /// connections alone never mutate this set.
     admitted_audio: StdMutex<HashSet<EndpointId>>,
+    /// Chat file attachments we're serving to room members this session, keyed by
+    /// the random attachment id. Populated when we send a chat file; read by the
+    /// file protocol handler to answer a member's fetch. Cleared on leave. Each
+    /// blob is already byte-capped at send time.
+    served_files: StdMutex<HashMap<crate::files::AttachmentId, Arc<Vec<u8>>>>,
     incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
     /// Best-effort link-state notifications for the UI (connecting / connected).
     conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -402,6 +412,88 @@ impl iroh::protocol::ProtocolHandler for AudioRouter {
     }
 }
 
+/// Protocol handler for the file-transfer plane (`FILES_ALPN`). Mirrors
+/// [`AudioRouter`]: it's persistent on the router and bound to the active
+/// session's [`Shared`] on join. On an inbound stream it authenticates the peer
+/// (iroh ALPN handshake gives us `remote_id`), gates on **live room membership**
+/// (same invariant as audio admission, so a former member can't pull files),
+/// reads a single 32-byte attachment id, and streams back the matching blob from
+/// the session serve store — or nothing if the id is unknown.
+#[derive(Clone, Default)]
+pub struct FileRouter {
+    current: Arc<StdMutex<Option<Arc<Shared>>>>,
+}
+
+impl std::fmt::Debug for FileRouter {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("FileRouter").finish_non_exhaustive()
+    }
+}
+
+impl FileRouter {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Route inbound file fetches to `transport`'s session (called on join).
+    pub fn bind(&self, transport: &IrohTransport) {
+        *self.current.lock().unwrap() = Some(transport.shared.clone());
+    }
+
+    /// Stop serving files until the next [`bind`](Self::bind) (called on leave).
+    pub fn clear(&self) {
+        *self.current.lock().unwrap() = None;
+    }
+}
+
+/// Max bytes we'll read for a fetch *request* frame. A request is exactly one
+/// 32-byte id; this small ceiling rejects a peer trying to stream us a huge
+/// "request" as a cheap DoS.
+const FILE_REQUEST_MAX: usize = 64;
+
+impl iroh::protocol::ProtocolHandler for FileRouter {
+    fn accept(
+        &self,
+        connection: Connection,
+    ) -> impl std::future::Future<Output = Result<(), iroh::protocol::AcceptError>> + Send {
+        let peer_id = connection.remote_id();
+        let shared = self.current.lock().unwrap().clone();
+        async move {
+            // No active call → nothing to serve.
+            let Some(shared) = shared else {
+                return Ok(());
+            };
+            // Member gating: only current room members may fetch our files. Reuses
+            // the audio admission roster (the authoritative room membership set).
+            if !shared.audio_sender_admitted(peer_id) {
+                crate::log_msg(&format!(
+                    "Transport: rejected file fetch from non-member {}",
+                    crate::short_id(&peer_id.to_string())
+                ));
+                return Ok(());
+            }
+            // Accept one bidirectional stream: read the id, write the bytes.
+            let Ok((mut send, mut recv)) = connection.accept_bi().await else {
+                return Ok(());
+            };
+            let Ok(req) = recv.read_to_end(FILE_REQUEST_MAX).await else {
+                return Ok(());
+            };
+            let Some(id) = crate::files::parse_request(&req) else {
+                return Ok(());
+            };
+            let blob = shared.served_files.lock().unwrap().get(&id).cloned();
+            if let Some(blob) = blob {
+                let _ = send.write_all(&blob).await;
+            }
+            // Finish either way: an unknown id closes with an empty body, which
+            // the fetcher reads as a zero-length result and treats as "gone".
+            let _ = send.finish();
+            Ok(())
+        }
+    }
+}
+
 pub struct IrohTransport {
     shared: Arc<Shared>,
     incoming_rx: tokio::sync::Mutex<Option<mpsc::Receiver<(EndpointId, Bytes)>>>,
@@ -426,6 +518,7 @@ impl IrohTransport {
             peers: tokio::sync::Mutex::new(HashMap::new()),
             live_conns: StdMutex::new(HashMap::new()),
             admitted_audio: StdMutex::new(HashSet::new()),
+            served_files: StdMutex::new(HashMap::new()),
             incoming_tx,
             conn_events_tx,
         });
@@ -455,6 +548,7 @@ impl IrohTransport {
         self.shared.senders.lock().unwrap().clear();
         self.shared.addrs.lock().unwrap().clear();
         self.shared.admitted_audio.lock().unwrap().clear();
+        self.shared.served_files.lock().unwrap().clear();
         // Give the CONNECTION_CLOSE frames a moment to flush before the caller
         // shuts the endpoint/router down (the `conns` clones are still alive
         // here, so the endpoint can still transmit them).
@@ -483,6 +577,59 @@ impl IrohTransport {
     pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
         self.shared.audio_sender_admitted(peer_id)
     }
+
+    /// Make `bytes` available to room members under `id` for the rest of this
+    /// session (served by the [`FileRouter`] handler). Called by core when we
+    /// send a chat file. The blob is cleared on leave.
+    pub fn serve_attachment(&self, id: AttachmentId, bytes: Arc<Vec<u8>>) {
+        self.shared.served_files.lock().unwrap().insert(id, bytes);
+    }
+
+    /// Fetch a chat attachment's bytes from its sender over the file plane. Dials
+    /// the sender on `FILES_ALPN` (preferring a known full address), writes the
+    /// 32-byte id, and reads the response bounded by the descriptor's declared
+    /// size (which the caller has already validated against the global cap). The
+    /// read limit means a malicious sender can't stream us more than advertised.
+    pub async fn fetch_attachment(
+        &self,
+        from: EndpointId,
+        att: &ChatAttachment,
+    ) -> Result<Vec<u8>, NetError> {
+        if !crate::files::size_within_cap(att.size) {
+            return Err(NetError::Other("attachment size out of range".to_string()));
+        }
+        let addr = self.shared.addrs.lock().unwrap().get(&from).cloned();
+        let connect = async {
+            match addr {
+                Some(addr) => self.shared.endpoint.connect(addr, FILES_ALPN).await,
+                None => self.shared.endpoint.connect(from, FILES_ALPN).await,
+            }
+        };
+        let conn = tokio::time::timeout(FILE_FETCH_TIMEOUT, connect)
+            .await
+            .map_err(|_| NetError::Other("file fetch: connect timed out".to_string()))?
+            .map_err(|e| NetError::Other(format!("file fetch: connect failed: {e}")))?;
+
+        let (mut send, mut recv) = conn
+            .open_bi()
+            .await
+            .map_err(|e| NetError::Other(format!("file fetch: open stream failed: {e}")))?;
+        send.write_all(&att.id)
+            .await
+            .map_err(|e| NetError::Other(format!("file fetch: request write failed: {e}")))?;
+        send.finish()
+            .map_err(|e| NetError::Other(format!("file fetch: request finish failed: {e}")))?;
+
+        let read = recv.read_to_end(att.size as usize);
+        let bytes = tokio::time::timeout(FILE_FETCH_TIMEOUT, read)
+            .await
+            .map_err(|_| NetError::Other("file fetch: read timed out".to_string()))?
+            .map_err(|e| NetError::Other(format!("file fetch: read failed: {e}")))?;
+        if bytes.is_empty() {
+            return Err(NetError::Other("file fetch: sender no longer has the file".to_string()));
+        }
+        Ok(bytes)
+    }
 }
 
 #[async_trait]