slice-4 (dev-a): MediaThread + session_map rewire (Task 6 + 7) (#12)

Co-authored-by: Aaron D. Lee <himself@adlee.work>
Co-committed-by: Aaron D. Lee <himself@adlee.work>

crates/rutster/src/lib.rs

@@ -22,6 +22,7 @@
 //! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md)
 //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical.
 
+pub mod media_thread;
 pub mod routes;
 pub mod session_map;
 pub mod tap_engine;

crates/rutster/src/main.rs

@@ -15,6 +15,7 @@ use std::net::SocketAddr;
 
 use rutster::routes::router;
 use rutster::session_map::AppState;
+use tokio::sync::mpsc;
 use tracing::info;
 
 #[tokio::main]
@@ -32,22 +33,30 @@ async fn main() {
         .expect(
             "RUTSTER_TAP_URL must be a valid ws:// URL (slice-2 dev loop; wss:// lands in step 6)",
         );
-    let state = AppState::new(default_tap_url);
-    state.clone().spawn_poll_task().await;
+    // Placeholder sender so we can build an AppState; the real sender comes
+    // from the spawned MediaThread below.
+    let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
+    let app_state = AppState::new(placeholder_tx, default_tap_url);
+    let (app_state, media_thread) = app_state
+        .spawn_media_thread(tokio::runtime::Handle::current())
+        .expect("media thread spawn");
 
     let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr");
     info!(%addr, "listening");
     let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
-    axum::serve(listener, router(state))
+    axum::serve(listener, router(app_state))
         .with_graceful_shutdown(shutdown_signal())
         .await
         .unwrap();
+
+    media_thread.shutdown();
 }
 
-/// Ctrl-C / SIGTERM handler (spec §4.5). Dropping the AppState drops the
-/// DashMap, which drops every RtcSession, which str0m sees as a closed
-/// peer — browsers get a dead peer connection. Acceptable for the dev
-/// loop; no in-flight call preservation story in slice 1.
+/// Ctrl-C / SIGTERM handler (spec §4.5). After the signal, `main` returns
+/// from `axum::serve(...)` and calls `media_thread.shutdown()`, which sends
+/// `MediaCmd::Shutdown` to the dedicated thread. The thread drains its
+/// session map, exits its loop, and is joined. No in-flight call
+/// preservation story in the dev loop.
 async fn shutdown_signal() {
     let ctrl_c = async {
         tokio::signal::ctrl_c()

crates/rutster/src/media_thread.rs

@@ -0,0 +1,332 @@
+//! # MediaThread — the dedicated 20ms media loop on a std::thread
+//! (slice-4 spec §2.2, §4)
+//!
+//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
+//! tokio pool." slice-1 ran the poll on tokio as an acknowledged
+//! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary
+//! startup owns `HashMap<ChannelId, RtcSession>` exclusively; all access
+//! from axum is via a command channel. The 10ms meta-tick between loop
+//! iterations is `std::thread::sleep(Duration::from_millis(10))`.
+//!
+//! # Why one thread, not per-session
+//!
+//! Spearhead scale (see slice-4 spec §6.3). The command-channel seam
+//! makes the later threadpool-shard graduation localized.
+//!
+//! # The seam (loop_driver + rtc_session byte-identical)
+//!
+//! `MediaThread` calls `RtcSession::run_poll_once(now)` — the unchanged
+//! `loop_driver::drive`. The `Reflex<TapAudioPipe>` wrapper is wired in
+//! here on the `Connected` transition (via `RtcSession::set_pipe`), not
+//! inside `rtc_session.rs`. The seam holds.
+
+use std::collections::HashMap;
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use rutster_call_model::ChannelId;
+use rutster_media::RtcSession;
+use tokio::sync::{mpsc, oneshot};
+use tracing::{debug, info};
+
+use crate::tap_engine::spawn_tap_engine;
+
+/// The 10ms meta-tick. Finer than the 20ms outbound encode tick so str0m's
+/// `Timeout` outputs are honored promptly.
+const META_TICK: Duration = Duration::from_millis(10);
+
+/// Capacity for the command channel from axum to the media thread.
+const CMD_CHANNEL_CAPACITY: usize = 64;
+
+/// Commands axum sends to the media thread (cold-path only — NEVER on
+/// the 20ms tick). The thread owns RtcSessions exclusively; this is the
+/// ONLY entry point for axum-side mutation.
+#[derive(Debug)]
+pub enum MediaCmd {
+    /// Construct a fresh RtcSession, store it under a new ChannelId, reply.
+    /// The thread constructs RtcSession::new() (keeps all RtcSession
+    /// construction on the thread that owns it).
+    Register {
+        tap_url: url::Url,
+        reply: oneshot::Sender<Result<ChannelId, String>>,
+    },
+    /// Accept a browser SDP offer on the session's behalf, reply with the
+    /// SDP answer (cold-path — the axum POST /v1/sessions/{id}/office handler).
+    AcceptOffer {
+        id: ChannelId,
+        sdp: String,
+        reply: oneshot::Sender<Result<String, String>>,
+    },
+    /// Tear down a session — fires close_tx + bounded-await the engine task
+    /// (750ms cap), then removes the entry.
+    Delete {
+        id: ChannelId,
+        reply: oneshot::Sender<()>,
+    },
+    /// Graceful shutdown — drain + drop + join.
+    Shutdown { reply: oneshot::Sender<()> },
+}
+
+/// The handle returned to the binary. Clone the `cmd_tx` per-session;
+/// the `JoinHandle` is dropped on shutdown to detach (the binary's
+/// graceful-shutdown signal fires `Shutdown` first).
+pub struct MediaThread {
+    cmd_tx: mpsc::Sender<MediaCmd>,
+    join: Option<JoinHandle<()>>,
+}
+
+impl MediaThread {
+    /// Return a clone of the command-channel sender, so the binary (axum
+    /// routes / `AppState`) can issue cold-path commands to the thread.
+    pub fn cmd_tx(&self) -> mpsc::Sender<MediaCmd> {
+        self.cmd_tx.clone()
+    }
+
+    /// Spawn the dedicated media thread. Captures a `tokio::runtime::Handle`
+    /// so the thread can `handle.spawn(spawn_tap_engine(...))` on the
+    /// `Connected` transition. The thread owns `HashMap<ChannelId,
+    /// ThreadSession>` exclusively.
+    ///
+    /// Returns `std::io::Error` if the OS cannot create the std::thread
+    /// (for example: the process has hit its thread limit). Thread creation
+    /// is part of the cold startup path, not the 20ms media loop, so the
+    /// error is propagated for the binary to decide whether to abort, retry,
+    /// or degrade. Inside the loop we still follow the "drop + observe, don't
+    /// crash" policy.
+    pub fn spawn(
+        default_tap_url: url::Url,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result<Self, std::io::Error> {
+        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
+        // The thread needs its own command-channel sender so it can build each
+        // session's `AppState`. Tools spawned inside the tap engine (notably
+        // `HangupTool`) call `AppState::close(id)`, which must loop back to
+        // this same media thread via `MediaCmd::Delete`.
+        let cmd_tx_for_thread = cmd_tx.clone();
+        let join = std::thread::Builder::new()
+            .name("rutster-media".into())
+            .spawn(move || {
+                run_media_thread(cmd_rx, default_tap_url, tokio_handle, cmd_tx_for_thread);
+            })?;
+        Ok(Self {
+            cmd_tx,
+            join: Some(join),
+        })
+    }
+
+    /// Graceful shutdown — drains commands + joins the thread.
+    pub fn shutdown(mut self) {
+        let (reply, rx) = oneshot::channel();
+        let _ = self.cmd_tx.blocking_send(MediaCmd::Shutdown { reply });
+        let _ = rx.blocking_recv();
+        if let Some(join) = self.join.take() {
+            let _ = join.join();
+        }
+    }
+}
+
+impl Drop for MediaThread {
+    fn drop(&mut self) {
+        if let Some(join) = self.join.take() {
+            // Best-effort: if shutdown wasn't called explicitly, just
+            // detach. The thread will exit when cmd_rx is dropped.
+            // (We don't block on join in Drop — that could deadlock
+            // if the thread is mid-call into a tokio runtime handle
+            // that's being torn down.)
+            debug!(name = ?join.thread().name(), "media thread detached on drop");
+        }
+    }
+}
+
+/// The per-session state owned by the media thread.
+struct ThreadSession {
+    rtc: RtcSession,
+    /// The tap URL resolved at registration time. This is either the
+    /// per-call override from `POST /v1/sessions` or the binary-wide default.
+    /// The `Connected` transition dials this URL, not the `MediaThread`-wide
+    /// default, so each session can be routed to a different brain/endpoint.
+    tap_url: url::Url,
+    /// A binary-view clone scoped to this session. `spawn_tap_engine` needs an
+    /// `AppState` so that tools like `HangupTool` can send teardown commands
+    /// back to the media thread through `AppState::close(id)`.
+    app_state: crate::session_map::AppState,
+    /// `Some` only after the `Connected` transition spawns the TapEngine
+    /// + wires the `Reflex<TapAudioPipe>` wrapper.
+    tap_conn: Option<crate::tap_engine::TapConn>,
+}
+
+fn run_media_thread(
+    mut cmd_rx: mpsc::Receiver<MediaCmd>,
+    default_tap_url: url::Url,
+    tokio_handle: tokio::runtime::Handle,
+    media_cmd_tx: mpsc::Sender<MediaCmd>,
+) {
+    let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
+    info!("media thread started");
+
+    loop {
+        // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. ===
+        while let Ok(cmd) = cmd_rx.try_recv() {
+            match cmd {
+                MediaCmd::Register { tap_url, reply } => match RtcSession::new() {
+                    Ok(session) => {
+                        let id = session.channel_id();
+                        let app_state = crate::session_map::AppState::new(
+                            media_cmd_tx.clone(),
+                            default_tap_url.clone(),
+                        );
+                        sessions.insert(
+                            id,
+                            ThreadSession {
+                                rtc: session,
+                                tap_url,
+                                app_state,
+                                tap_conn: None,
+                            },
+                        );
+                        let _ = reply.send(Ok(id));
+                        debug!(channel_id = %id, "session registered");
+                    }
+                    Err(e) => {
+                        let _ = reply.send(Err(format!("RtcSession::new: {e}")));
+                    }
+                },
+                MediaCmd::AcceptOffer { id, sdp, reply } => {
+                    let result = match sessions.get_mut(&id) {
+                        Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")),
+                        None => Err(format!("session {id} not found")),
+                    };
+                    let _ = reply.send(result);
+                }
+                MediaCmd::Delete { id, reply } => {
+                    if let Some(mut s) = sessions.remove(&id) {
+                        if let Some(mut conn) = s.tap_conn.take() {
+                            let _ = conn.close_tx.send(());
+                            let teardown = tokio_handle.block_on(tokio::time::timeout(
+                                Duration::from_millis(750),
+                                &mut conn.join,
+                            ));
+                            match teardown {
+                                Ok(Ok(())) => {
+                                    info!(channel_id = %id, "tap engine torn down via Delete (graceful)");
+                                }
+                                _ => {
+                                    conn.join.abort();
+                                    info!(channel_id = %id, "tap engine torn down via Delete (abort after timeout)");
+                                }
+                            }
+                        }
+                        s.rtc.channel.tap = None;
+                        s.rtc.channel.state = rutster_call_model::ChannelState::Closed;
+                    }
+                    let _ = reply.send(());
+                }
+                MediaCmd::Shutdown { reply } => {
+                    info!(
+                        "media thread shutdown; dropping {} sessions",
+                        sessions.len()
+                    );
+                    sessions.clear();
+                    let _ = reply.send(());
+                    return;
+                }
+            }
+        }
+
+        // === Step 2: the 10ms meta-tick over all sessions. ===
+        let now = Instant::now();
+        let mut closed_ids: Vec<ChannelId> = Vec::new();
+        for (id, session) in sessions.iter_mut() {
+            // Drain flush side-channel BEFORE run_poll_once (slice-2 §5.3 step 4).
+            if let Some(conn) = session.tap_conn.as_mut() {
+                if let Some(rx) = conn.flush_rx.as_mut() {
+                    let mut should_flush = false;
+                    while let Ok(()) = rx.try_recv() {
+                        should_flush = true;
+                    }
+                    if should_flush {
+                        session.rtc.clear_playout_ring();
+                    }
+                }
+            }
+            let _ = session.rtc.run_poll_once(now);
+
+            // === relay the Connected transition: spawn TapEngine + wire Reflex. ===
+            use rutster_call_model::ChannelState;
+            if let ChannelState::Connected = session.rtc.channel.state {
+                if session.rtc.channel.tap.is_none() {
+                    // Use the per-call tap URL stored at registration time, not
+                    // the binary-wide default, so a `tap_url_override` from the
+                    // API body is honored.
+                    let url = session.tap_url.clone();
+                    // Give the tap engine a real `AppState` whose `cmd_tx`
+                    // points back to this media thread. That lets tools such as
+                    // `HangupTool` call `AppState::close(id)` and actually
+                    // reach `MediaCmd::Delete` here.
+                    let app_state = session.app_state.clone();
+                    // The media thread owns the advisory channel (multi-producer:
+                    // tokio::sync::mpsc::Sender is Clone). Both the brain's
+                    // advisories (via spawn_tap_engine) AND the local VAD's
+                    // trips (via LocalVadReflex) push to the SAME mpsc; the
+                    // Reflex drains both uniformly. This means spawn_tap_engine
+                    // takes advisory_tx as a PARAMETER (Task 5's signature
+                    // changes: spawn_tap_engine(session_id, tap_url, app_state,
+                    // advisory_tx) — see Task 5's revision note).
+                    let (advisory_tx, advisory_rx) =
+                        mpsc::channel::<rutster_media::AdvisoryEvent>(16);
+                    let (pipe, conn) = tokio_handle.block_on(async {
+                        spawn_tap_engine(*id, url, app_state, advisory_tx.clone())
+                    });
+                    let metrics = rutster_media::ReflexMetrics::new();
+                    // Compose: Reflex<TapAudioPipe> (state machine) wrapped by
+                    // LocalVadReflex (primary VAD trigger). Both feed advisory_tx.
+                    let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics);
+                    let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx);
+                    session.rtc.set_pipe(vad);
+                    session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
+                    session.tap_conn = Some(conn);
+                    info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
+                    continue;
+                }
+            }
+
+            if session.rtc.is_closed() {
+                closed_ids.push(*id);
+            }
+        }
+        for id in closed_ids {
+            sessions.remove(&id);
+            debug!(channel_id = %id, "session evicted after close");
+        }
+
+        // === Step 3: sleep META_TICK. ===
+        std::thread::sleep(META_TICK);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn media_thread_register_and_shutdown_round_trips() {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let thread = MediaThread::spawn(url, handle).expect("media thread spawn in test");
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let id = rx
+            .blocking_recv()
+            .expect("register reply")
+            .expect("session");
+        assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId");
+        thread.shutdown();
+    }
+}

crates/rutster/src/routes.rs

@@ -47,7 +47,7 @@ pub async fn create_session(
         Ok(u) => u,
         Err(e) => return (StatusCode::BAD_REQUEST, e).into_response(),
     };
-    match state.create_session(Some(tap_url)) {
+    match state.create_session(Some(tap_url)).await {
         Ok(id) => {
             let body = Json(SessionCreated {
                 session_id: id.0.to_string(),
@@ -109,11 +109,7 @@ pub async fn post_offer(
         return (StatusCode::NOT_FOUND, "bad session id").into_response();
     };
     let id = rutster_call_model::ChannelId(id_uuid);
-    let Some(session_arc) = state.get(id) else {
-        return (StatusCode::NOT_FOUND, "no such session").into_response();
-    };
-    let mut s = session_arc.lock().await;
-    match s.accept_offer(&body) {
+    match state.accept_offer(id, body).await {
         Ok(answer_sdp) => (
             StatusCode::OK,
             [(header::CONTENT_TYPE, "application/sdp")],
@@ -121,8 +117,12 @@ pub async fn post_offer(
         )
             .into_response(),
         Err(e) => {
-            tracing::error!(error = ?e, "SDP accept failed");
-            StatusCode::BAD_REQUEST.into_response()
+            tracing::error!(error = %e, "SDP accept failed");
+            if e.contains("not found") {
+                (StatusCode::NOT_FOUND, e).into_response()
+            } else {
+                StatusCode::BAD_REQUEST.into_response()
+            }
         }
     }
 }

crates/rutster/src/session_map.rs

@@ -1,516 +1,111 @@
-//! # Session store + poll-driver (spec §4.5; slice-2 §5.1, §6)
+//! # Session store + command channel (slice-4 §4.3)
 //!
-//! `DashMap<ChannelId, SessionEntry>` holds active sessions; the `ChannelId`
-//! (UUID newtype from `rutster-call-model`) IS the session id surfaced in
-//! the REST API. A single tokio task drives all sessions' poll loops (a
-//! per-session task would clutter the runtime and pre-pave the wrong
-//! pattern for the step-4 dedicated thread — spec §4.5).
+//! `AppState` no longer owns sessions directly. The dedicated media thread
+//! (see `media_thread::MediaThread`) owns `HashMap<ChannelId, RtcSession>`
+//! exclusively; axum routes interact via a cold-path command channel
+//! (`MediaCmd`).
 //!
-//! # Concurrency note
-//!
-//! `DashMap` shards its inner `HashMap` so concurrent gets/puts across
-//! different `ChannelId`s don't contend. We iterate per-shard inside the
-//! poll task to drive each session; entries marked `Closed` are removed.
-//!
-//! # slice-2: TapEngine wiring seam (spec §5.1 step 3)
-//!
-//! `drive_all_sessions` is the spawn boundary for the per-session TapEngine
-//! task. After each `RtcSession::run_poll_once`, we observe the
-//! `channel.state` transition:
-//! - `Connected && tap.is_none()` → spawn TapEngine, wire `TapAudioPipe`
-//!   into `RtcSession.pipe` via `with_pipe`, set `channel.tap = Some(...)`.
-//!
-//! Teardown is NOT a poll-task branch here — all slice-2 teardown happens
-//! inline in `AppState::close` (removes the entry, fires `close_tx`, aborts
-//! the task, clears `channel.tap`). The future peer-initiated close path
-//! (browser `peerconnectionclose`) WILL observe the `Closing` transition
-//! here; deferred per slice-2 §1.2.
-//!
-//! `loop_driver.rs` is NOT modified — the spawn happens in this poll-task
-//! layer per spec §8.5 #6.
+//! This replaces the slice-3 `DashMap<ChannelId, SessionEntry>` + tokio
+//! poll task with the slice-4 dedicated `std::thread` architecture. The
+//! command channel is the only cross-thread state; handlers never lock a
+//! session directly.
 
-use std::sync::Arc;
-use std::time::{Duration, Instant};
+use rutster_call_model::ChannelId;
+use tokio::sync::{mpsc, oneshot};
+use tracing::debug;
 
-use dashmap::DashMap;
-use rutster_call_model::{ChannelId, ChannelState, TapHandle};
-use rutster_media::{RtcSession, RtcSessionError};
-use tokio::sync::Mutex;
-use tracing::{debug, info, warn};
+use crate::media_thread::{MediaCmd, MediaThread};
 
-use crate::tap_engine::{TapConn, spawn_tap_engine};
-// Re-using the binary crate's `tokio::sync::mpsc` import (the engine task
-// + the poll task both live in `rutster`'s poll-driver module). The type
-// only appears in `TapConn` field signatures + the slice-3 §5.2 dispatch
-// helper below; bringing it in here keeps the type names short.
-use tokio::sync::mpsc;
-
-/// The per-session wrapper struct (slice-2, spec §6).
+/// Application state shared across axum handlers.
 ///
-/// # Why a wrapper (not parallel DashMaps?)
-///
-/// The brief's Step 3 sketched two designs: a wrapper struct or a parallel
-/// `DashMap<ChannelId, TapConn>`. The wrapper is cleaner because the
-/// three fields (`rtc`, `tap_url`, `tap_conn`) are mutated by the same
-/// state-transition logic in `drive_all_sessions`; holding them under one
-/// DashMap entry means one shard-lookup per cycle, one consistent handle
-/// for the whole lifecycle. A parallel map would double the shard lookups
-/// and split the lifecycle state across two maps (risk: tap_url updated
-/// but tap_conn not yet spawned due to a missed entry in either map).
-///
-/// `rtc` is behind `Arc<Mutex<RtcSession>>` because the `routes::post_offer`
-/// handler ALSO needs to lock it for SDP acceptance, in parallel with the
-/// poll task. `tap_url` and `tap_conn` are only mutated by the poll task
-/// (after `drive_all_sessions` returns from `run_poll_once`), so they
-/// don't need their own mutex.
-pub struct SessionEntry {
-    pub rtc: Arc<Mutex<RtcSession>>,
-    pub tap_url: url::Url,
-    pub tap_conn: Option<TapConn>,
-}
-
-/// The application state shared across axum handlers + the poll task.
-///
-/// # Why `Arc` (and not bare)
-///
-/// axum clones the state into every handler. `Arc` is the standard way
-/// to share `DashMap` + `Mutex` owned state across these clones cheaply
-/// (a single heap allocation, refcount-bumped per clone). Without `Arc`,
-/// every handler would move its own copy — and `DashMap` is not `Copy`.
-///
-/// # Why a separate `poll_running` `Mutex`
-///
-/// The poll loop is one task; we don't want two. The Mutex guards a
-/// once-only spawn: `spawn_poll_task` checks-and-sets it under the mutex.
-/// `Mutex` (not `RwLock`) because the only operation is "take it once."
+/// `cmd_tx` is the command channel to the dedicated media thread. All
+/// session mutation flows through `MediaCmd` cold-path commands.
 #[derive(Clone)]
 pub struct AppState {
-    pub sessions: Arc<DashMap<ChannelId, SessionEntry>>,
-    pub poll_running: Arc<Mutex<bool>>,
-    /// Default brain URL — per-call overrides come from `POST /v1/sessions`
-    /// body (spec §7). Set from `RUTSTER_TAP_URL` env at binary startup
-    /// (default `ws://127.0.0.1:8081/echo`); validation per spec §4.4
-    /// happens at route time (`routes::resolve_tap_url`), not here.
+    pub cmd_tx: mpsc::Sender<MediaCmd>,
     pub default_tap_url: url::Url,
 }
 
 impl AppState {
-    pub fn new(default_tap_url: url::Url) -> Self {
+    /// Build state with a command-channel sender and default tap URL.
+    ///
+    /// Callers usually pass a placeholder sender; `spawn_media_thread`
+    /// consumes the state and returns it with the real thread sender
+    /// installed so the command channel terminates in a running media
+    /// thread.
+    pub fn new(cmd_tx: mpsc::Sender<MediaCmd>, default_tap_url: url::Url) -> Self {
         Self {
-            sessions: Arc::new(DashMap::new()),
-            poll_running: Arc::new(Mutex::new(false)),
+            cmd_tx,
             default_tap_url,
         }
     }
 
-    /// Mint a fresh `RtcSession`, store it under its `ChannelId` with the
-    /// resolved `tap_url` (default or per-call override), return the id.
-    /// The `tap_url` is carried in the entry so the poll task can spawn
-    /// the TapEngine on the `Connected` transition without re-reading
-    /// `default_tap_url` or a parallel map.
-    pub fn create_session(
+    /// Ask the media thread to create a fresh `RtcSession`, returning its id.
+    pub async fn create_session(
         &self,
         tap_url_override: Option<url::Url>,
-    ) -> Result<ChannelId, RtcSessionError> {
-        let session = RtcSession::new()?;
-        let id = session.channel_id();
+    ) -> Result<ChannelId, String> {
         let tap_url = tap_url_override.unwrap_or_else(|| self.default_tap_url.clone());
-        let entry = SessionEntry {
-            rtc: Arc::new(Mutex::new(session)),
-            tap_url,
-            // Set on the `Connected` transition in `drive_all_sessions`.
-            tap_conn: None,
-        };
-        self.sessions.insert(id, entry);
+        let (reply, rx) = oneshot::channel();
+        self.cmd_tx
+            .send(MediaCmd::Register { tap_url, reply })
+            .await
+            .map_err(|e| format!("media thread gone: {e}"))?;
+        let id = rx
+            .await
+            .map_err(|e| format!("media thread reply dropped: {e}"))??;
+        debug!(%id, "session registered via media thread");
         Ok(id)
     }
 
-    /// Look up a session by id (returns the clone of the Arc-wrapped Mutex).
-    /// Used by `routes::post_offer` for SDP acceptance — accepts the
-    /// lock for the duration of `accept_offer`, then releases.
-    pub fn get(&self, id: ChannelId) -> Option<Arc<Mutex<RtcSession>>> {
-        self.sessions.get(&id).map(|r| r.rtc.clone())
+    /// Ask the media thread to accept a browser SDP offer, returning the SDP answer.
+    pub async fn accept_offer(&self, id: ChannelId, sdp: String) -> Result<String, String> {
+        let (reply, rx) = oneshot::channel();
+        self.cmd_tx
+            .send(MediaCmd::AcceptOffer { id, sdp, reply })
+            .await
+            .map_err(|e| format!("media thread gone: {e}"))?;
+        rx.await
+            .map_err(|e| format!("media thread reply dropped: {e}"))?
     }
 
-    /// Transition to Closing then Closed. slice-2 spec §5.1 step 5 + §5.2
-    /// teardown sequence:
+    /// Ask the media thread to tear down a session.
     ///
-    /// 1. Remove the session entry from the DashMap (the poll task won't
-    ///    observe it again).
-    /// 2. If a TapEngine was attached, fire `close_tx` — this triggers
-    ///    `run_tap_client`'s close arm, which sends `session_end` over the
-    ///    WS, awaits brain `bye` (bounded 500 ms), then closes the WS.
-    ///    We bounded-await the engine task for 750 ms (strictly larger
-    ///    than the inner 500 ms bound) so the inner has room to finish its
-    ///    post-timeout `ws.close(None).await` and cleanly exit before our
-    ///    abort fallback fires.
-    /// 3. Fall back to `JoinHandle::abort()` if the bounded-await times out
-    ///    (the brain didn't ack, or the pump was stuck — abort as safety net).
-    /// 4. Clear `channel.tap = None` BEFORE state advances to `Closed` per
-    ///    spec §6's state invariant (`Closing` ∎ tap == None ⇐ `Closed`).
-    /// 5. Set `Closing`.
-    /// 6. Set `Closed`.
-    ///
-    /// NOT in `drive_all_sessions`'s poll loop: `close` is called from
-    /// `routes::delete_session` and could race with the poll task; doing
-    /// teardown inline on entry removal keeps the TapConn lifecycle owned
-    /// by exactly one call site.
+    /// Fire-and-forget on the command channel: if the thread is gone, the
+    /// session is already torn down. This is why the method returns `()`
+    /// rather than a `Result`.
     pub async fn close(&self, id: ChannelId) {
-        if let Some((_id, entry)) = self.sessions.remove(&id) {
-            // === Step 2: fire close signal + bounded-await the engine task's
-            // teardown handshake. `&mut conn.join` borrows the JoinHandle
-            // mutably so we can still call `.abort()` on it after the timeout
-            // — taking it by value would prevent the fallback abort. ===
-            if let Some(mut conn) = entry.tap_conn {
-                let _ = conn.close_tx.send(());
-                // Outer bound STRICTLY LARGER than the inner close-arm bound
-                // (500 ms in `tap_client.rs::run_tap_client`'s close arm):
-                // if the brain doesn't `bye`-ack, the inner arm times out at
-                // ~500 ms and then calls `ws.close(None).await` (which can
-                // take additional ms) before exiting via `Err(Closed)` →
-                // `return;`. The 750 ms outer bound gives the inner room to
-                // finish that post-timeout `ws.close(None).await` and cleanly
-                // exit before this outer-bound abort fallback fires.
-                // (final-fixes re-review Important #1.)
-                let teardown =
-                    tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await;
-                match teardown {
-                    Ok(Ok(())) => {
-                        info!(channel_id = %id, "tap engine torn down via DELETE (graceful handshake)");
-                    }
-                    Ok(Err(e)) => {
-                        warn!(error = ?e, channel_id = %id, "tap engine task failed during teardown");
-                    }
-                    Err(_) => {
-                        // Engine didn't finish in 750 ms — abort as fallback.
-                        // Spec §5.2: a brain that doesn't bye in time just
-                        // gets a WS close; we cap our own wait to keep the
-                        // DELETE handler responsive.
-                        conn.join.abort();
-                        info!(channel_id = %id, "tap engine torn down via DELETE (abort after 750ms timeout)");
-                    }
-                }
-            }
-            // === Step 4-6: clear tap BEFORE Closing → Closed ===
-            // (spec §6 state invariant: `tap == None` by the time the
-            // channel reaches `Closing`/`Closed`).
-            let mut s = entry.rtc.lock().await;
-            s.channel.tap = None;
-            s.channel.state = ChannelState::Closing;
-            s.channel.state = ChannelState::Closed;
-            info!(channel_id = %id, "session closed via DELETE");
-        }
+        let (reply, rx) = oneshot::channel();
+        let _ = self.cmd_tx.send(MediaCmd::Delete { id, reply }).await;
+        let _ = rx.await;
     }
 
-    /// Spawn the single poll task for all sessions (idempotent).
-    pub async fn spawn_poll_task(self) {
-        let mut running = self.poll_running.lock().await;
-        if *running {
-            return;
-        }
-        *running = true;
-        drop(running);
-
-        let state = self.clone();
-        tokio::spawn(async move {
-            let mut interval = tokio::time::interval(Duration::from_millis(10));
-            interval.tick().await;
-            loop {
-                interval.tick().await;
-                let now = Instant::now();
-                drive_all_sessions(&state, now).await;
-            }
-        });
+    /// Spawn the dedicated media thread for this state.
+    ///
+    /// Returns the state with the real command-channel sender wired in,
+    /// plus the `MediaThread` handle. Routes on the returned `AppState`
+    /// will send `MediaCmd` to the running media thread.
+    pub fn spawn_media_thread(
+        mut self,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result<(Self, MediaThread), std::io::Error> {
+        let thread = MediaThread::spawn(self.default_tap_url.clone(), tokio_handle)?;
+        self.cmd_tx = thread.cmd_tx();
+        Ok((self, thread))
     }
 }
 
 impl Default for AppState {
     fn default() -> Self {
-        Self::new(url::Url::parse("ws://127.0.0.1:8081/echo").expect("valid default tap URL"))
-    }
-}
-
-/// One iteration of "drive every active session." Removes closed entries.
-///
-/// Per spec §5.1 step 3 (slice-2): the poll task is the spawn boundary for
-/// the per-session TapEngine. We observe the channel state AFTER
-/// `run_poll_once` returns; the `loop_driver` already wrote the new state.
-/// Specifically:
-/// - `Connected && tap.is_none()` → spawn TapEngine → wire TapAudioPipe
-///   into RtcSession via `with_pipe` → set `channel.tap = Some(TapHandle)`.
-///
-/// Teardown is NOT a poll-task branch — see `AppState::close` and the
-/// module-level docs for why.
-///
-/// The wiring race (spec §5.1 race note): in the poll cycle that observes
-/// the `Connected` transition, `MediaData` frames in the same cycle go
-/// through the default `EchoAudioPipe` (one or two frames echo-piped
-/// locally — imperceptible). On the NEXT cycle the TapAudioPipe is wired.
-/// Acceptable for slice-2 — do NOT swap mid-cycle (the spec specifically
-/// says the spawn happens here, not in `loop_driver.rs`).
-async fn drive_all_sessions(state: &AppState, now: Instant) {
-    // Collect ids first to avoid holding the DashMap shard during the
-    // async poll (which would block other handlers mutating the same shard).
-    let ids: Vec<ChannelId> = state.sessions.iter().map(|r| *r.key()).collect();
-    for id in ids {
-        // === slice-2 §5.3 step 4: drain the per-session flush side-channel
-        // BEFORE the poll cycle. The engine task signals a flush via
-        // `flush_tx.try_send(())` after every failed pump loop (WS error or
-        // brain WS-close). We drain ALL pending signals here (collapses
-        // multiple signals into one `clear_playout_ring` call — idempotent)
-        // and apply the flush before `run_poll_once` so the next playout
-        // tick starts from a clean ring. ===
-        let mut should_flush = false;
-        if let Some(mut entry) = state.sessions.get_mut(&id) {
-            if let Some(conn) = entry.tap_conn.as_mut() {
-                if let Some(rx) = conn.flush_rx.as_mut() {
-                    while let Ok(()) = rx.try_recv() {
-                        should_flush = true;
-                    }
-                }
-            }
-        }
-        // slice-3 §5.2: drain the per-session `rx_function_call`
-        // side-channel in the same cycle as the `flush_rx` drain (slice-2
-        // §5.3 step 4 pattern — one extra channel, same cycle). The
-        // helper spawns each dispatch as its own task so the 750 ms
-        // `AppState::close` await (hangup's teardown handshake) can't
-        // stall the poll cadence.
-        let _fc_drained = drain_function_calls(state, id).await;
-        // Hold the DashMap Ref only long enough to clone the Arc-wrapped
-        // rtc + tap_url; the async poll + spawn happens outside the shard.
-        let (rtc, tap_url) = match state.sessions.get(&id) {
-            Some(r) => (r.rtc.clone(), r.tap_url.clone()),
-            None => continue,
-        };
-        let mut s = rtc.lock().await;
-        if should_flush {
-            // Apply the flush before the poll cycle so any new audio_out
-            // frames the brain queues during this poll go into an empty ring.
-            s.clear_playout_ring();
-        }
-        let _ = s.run_poll_once(now); // hot-path match-and-continue inside
-
-        // === slice-2: TapEngine spawn seam (spec §5.1 step 3, §8.5 #6). ===
-        // Observe the state AFTER `loop_driver::drive` mutated `channel.state`
-        // — we make the engine-control decision here, NOT inside loop_driver
-        // (the seam test preserves `loop_driver.rs` byte-identical call sites).
-        //
-        // Teardown is intentionally NOT a poll-task branch here. All
-        // slice-2 teardown happens inline in `AppState::close` (fired from
-        // `routes::delete_session`): it removes the DashMap entry, fires
-        // `close_tx`, aborts the engine `JoinHandle`, and clears
-        // `channel.tap`. By the time this poll task iterates again the
-        // entry is GONE — a `Closing && tap.is_some()` branch would be
-        // dead code. The future peer-initiated close path (when slice-2
-        // eventually handles a `peerconnectionclose` event from the
-        // browser) WILL live here, observing the `Closing` transition to
-        // tear down the engine before the entry is removed — deferred
-        // per slice-2 §1.2 (no browser-driven close events this slice).
-        if let ChannelState::Connected = s.channel.state {
-            if s.channel.tap.is_none() {
-                // First connect: spawn the TapEngine, wire the TapAudioPipe.
-                // slice-3 §6.2: spawn_tap_engine constructs a per-channel
-                // ToolRegistry holding HangupTool (the only wired tool in
-                // slice-3 — §6.3) bound to this AppState + ChannelId.
-                let app_state = state.clone();
-                let tap_url_clone = tap_url.clone();
-                // slice-4 Task-5 bridge: spawn_tap_engine now takes the advisory
-                // sender. The current tokio poll-task creates a throwaway channel
-                // here; dev-a Task 7 (MediaThread) will own the real channel and
-                // pass its sender (cloned to both the engine and LocalVadReflex).
-                let (advisory_tx, _advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
-                let (pipe, conn) = spawn_tap_engine(id, tap_url_clone, app_state, advisory_tx);
-                s.set_pipe(pipe);
-                s.channel.tap = Some(TapHandle::new());
-                info!(channel_id = %id, "tap engine spawned on Connected");
-                // Store the TapConn handle. Drop the per-session Mutex first
-                // to avoid holding it across the DashMap shard write.
-                drop(s);
-                if let Some(mut entry) = state.sessions.get_mut(&id) {
-                    entry.tap_conn = Some(conn);
-                }
-                continue;
-            }
-        }
-
-        if s.is_closed() {
-            drop(s);
-            state.sessions.remove(&id);
-            debug!(channel_id = %id, "session evicted after close");
-        }
-    }
-}
-
-/// slice-3 §5.2 + §6 — drain the per-session `rx_function_call` side-channel
-/// and dispatch each event through the per-channel `ToolRegistry`. One
-/// dispatch result → one `function_call_output` written to
-/// `tx_function_call_output` (which the TapClient forwards to the brain on
-/// its next pump cycle).
-///
-/// # Why this is a separate helper (not inline in `drive_all_sessions`)
-///
-/// `ToolRegistry::dispatch` is `async` (the `hangup` tool calls
-/// `AppState::close`, which awaits the engine's teardown handshake — spec
-/// §5.2's 750 ms bound). Awaiting that inline in the poll task would stall
-/// the 10 ms poll cadence for every hangup. Instead we **collect** events
-/// non-blockingly here (`try_recv` — drop + observe on full channel, same
-/// posture as the existing flush drain), then **spawn** each dispatch as
-/// its own tokio task so the poll task returns to its 10 ms cadence
-/// immediately. The spawned task holds its own clones of `AppState` +
-/// `tool_registry` + `tx_function_call_output` — no shared mutable state
-/// with the poll path.
-///
-/// Returns the count of events drained (for observability — matches the
-/// flush drain's "drain all" loop shape). The poll task calls this in the
-/// same `drive_all_sessions` cycle it drains the `flush_rx` side-channel
-/// (slice-2 §5.3 step 4 pattern, one extra channel, same cycle).
-async fn drain_function_calls(state: &AppState, id: ChannelId) -> usize {
-    // Collect: pull everything we need out of the DashMap shard BEFORE any
-    // await (the spawned dispatch's `AppState::close` later takes the same
-    // shard lock — holding a `get_mut` Ref across the spawn's await would
-    // deadlock slice-2's other shard-mutating handlers). The Collector scope
-    // owns a single short-lived `get_mut` Ref; the loop does only sync
-    // `try_recv` + Vec push (no await).
-    let collected: Vec<rutster_tap::FunctionCallEvent>;
-    let tool_registry: Arc<tokio::sync::Mutex<crate::tool_registry::ToolRegistry>>;
-    let tx_out: Option<mpsc::Sender<rutster_tap::FunctionCallOutputEvent>>;
-    {
-        let Some(mut entry) = state.sessions.get_mut(&id) else {
-            return 0;
-        };
-        let Some(conn) = entry.tap_conn.as_mut() else {
-            return 0;
-        };
-        tool_registry = conn.tool_registry.clone();
-        tx_out = conn.tx_function_call_output.clone();
-        let Some(rx_fc) = conn.rx_function_call.as_mut() else {
-            return 0;
-        };
-        collected = (0..).map_while(|_| rx_fc.try_recv().ok()).collect();
-    }
-    let drained = collected.len();
-
-    // Spawn dispatches outside the shard Ref scope. `AppState::close` is a
-    // 750 ms bounded wait in the worst case (slice-2 §5.2's teardown
-    // handshake); spawning keeps the 10 ms poll cadence responsive.
-    for event in collected {
-        let app_state = state.clone();
-        let reg = tool_registry.clone();
-        let tx_out = tx_out.clone();
-        tokio::spawn(async move {
-            let payload = event.0;
-            let call_id = payload.id.clone();
-            let name = payload.name.clone();
-            debug!(channel_id = %id, call_id = %call_id, tool = %name, "dispatching function_call");
-            let result = reg.lock().await.dispatch(&name, payload.args).await;
-            let (status, result_value) = result.to_status_result();
-            let out =
-                rutster_tap::FunctionCallOutputEvent(rutster_tap::FunctionCallOutputPayload {
-                    id: call_id.clone(),
-                    status: status.clone(),
-                    result: result_value,
-                });
-            // try_send — non-blocking. The TapClient's pump loop drains
-            // `rx_function_call_output` on its next `select!` cycle. If the
-            // channel is full (brain not pumping), we drop + observe (same
-            // posture as the engine's `tx_function_call.try_send` on the
-            // inbound side). `tx_out` is `Option<Sender>`; `None` means the
-            // engine has torn down its receiver — the result is dropped,
-            // acceptable since the call is hanging up.
-            if let Some(tx) = tx_out.as_ref() {
-                if let Err(e) = tx.try_send(out) {
-                    warn!(
-                        channel_id = %id, call_id = %call_id, error = ?e,
-                        "function_call_output dropped (TapClient pump not draining)"
-                    );
-                }
-            }
-            // app_state is held for the dispatch lifetime — the HangupTool
-            // inside the registry already captured its own clone at registry
-            // construction, so this outer binding exists for future tool
-            // impls that need the live AppState handed to dispatch (none in
-            // slice-3 beyond hangup). Drop the silent-binding warning by
-            // referencing it once.
-            drop(app_state);
-            info!(channel_id = %id, call_id = %call_id, status = %status, "function_call dispatched");
-        });
-    }
-    drained
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use rutster_tap::{
-        FunctionCallEvent, FunctionCallOutputEvent, FunctionCallPayload, TapMetrics,
-    };
-    use tokio::sync::Mutex;
-
-    /// slice-3 §5.2 + §6 — a `function_call` event drained from
-    /// `rx_function_call` triggers `ToolRegistry::dispatch("hangup")` which
-    /// fires `AppState::close` (the slice-2 teardown path), and the dispatch
-    /// result flows back as a `FunctionCallOutputEvent` on
-    /// `tx_function_call_output`. End-to-end contract test for the helper
-    /// minus the live TapClient pump (which is integration-test territory).
-    #[tokio::test]
-    async fn drain_function_calls_dispatches_hangup_and_writes_output() {
-        let state = AppState::default();
-        // Create a session so AppState::close has something to remove.
-        let id = state.create_session(None).unwrap();
-
-        // Build a TapConn with manually-controlled side-channel ends so we
-        // can push a FunctionCallEvent from the test side + observe the
-        // dispatch's output on the paired Receiver.
-        let (tx_fc, rx_fc) = tokio::sync::mpsc::channel::<FunctionCallEvent>(8);
-        let (tx_fco, mut rx_fco) = tokio::sync::mpsc::channel::<FunctionCallOutputEvent>(8);
-        let mut registry = crate::tool_registry::ToolRegistry::new();
-        registry.register(Box::new(crate::tool_registry::HangupTool::new(
-            state.clone(),
-            id,
-        )));
-        let (close_tx, _close_rx) = tokio::sync::oneshot::channel::<()>();
-        let conn = crate::tap_engine::TapConn {
-            close_tx,
-            join: tokio::spawn(async {}), // no-op handle; aborted below
-            metrics: TapMetrics::new(),
-            flush_rx: None,
-            rx_function_call: Some(rx_fc),
-            tx_function_call_output: Some(tx_fco),
-            tool_registry: Arc::new(Mutex::new(registry)),
-        };
-        state.sessions.get_mut(&id).unwrap().tap_conn = Some(conn);
-
-        // Push a function_call for the hangup tool — simulates what the
-        // TapClient does when it observes a `function_call` tap frame.
-        tx_fc
-            .send(FunctionCallEvent(FunctionCallPayload {
-                id: "call-1".to_string(),
-                name: "hangup".to_string(),
-                args: serde_json::json!({}),
-            }))
-            .await
-            .unwrap();
-
-        // Drain — this spawns a dispatch task + returns immediately.
-        let drained = drain_function_calls(&state, id).await;
-        assert_eq!(drained, 1, "exactly one function_call should drain");
-
-        // The spawned dispatch task fires AppState::close (session removed)
-        // + writes the function_call_output. Bounded-wait the result with
-        // a generous timeout (AppState::close has a 750 ms teardown bound +
-        // the spawned task may take a few ms to schedule).
-        let out = tokio::time::timeout(Duration::from_secs(2), rx_fco.recv())
-            .await
-            .expect("function_call_output drained within 2s")
-            .expect("channel not closed");
-        assert_eq!(out.0.id, "call-1");
-        assert_eq!(out.0.status, "ok");
-        assert_eq!(out.0.result["channel_state"], "Closing");
-        // AppState::close removed the session entry (the teardown it fires).
-        assert!(
-            state.sessions.get(&id).is_none(),
-            "session should be removed after hangup dispatch"
-        );
+        // Closed placeholder sender — kept for tests that construct an
+        // `AppState` without a running media thread. Production code now
+        // builds real `AppState` values with the thread's `cmd_tx` inside
+        // `media_thread::run_media_thread` so tap-engine tools can tear down
+        // their own sessions.
+        let (tx, _rx) = mpsc::channel(1);
+        Self::new(
+            tx,
+            url::Url::parse("ws://127.0.0.1:8081/echo").expect("valid default tap URL"),
+        )
     }
 }

crates/rutster/tests/api_integration.rs

@@ -7,6 +7,7 @@
 
 use axum::body::Body;
 use axum::http::{Request, StatusCode};
+use rutster::media_thread::MediaThread;
 use rutster::session_map::AppState;
 use tower::ServiceExt; // enables `oneshot` on the Router for sync tests
 
@@ -14,9 +15,24 @@ fn default_tap_url() -> url::Url {
     url::Url::parse("ws://127.0.0.1:8081/echo").unwrap()
 }
 
+/// Build an `AppState` whose command channel terminates in a real
+/// `MediaThread`. The rewired slice-4 routes require a running media thread
+/// to create sessions; the integration test sets one up exactly like
+/// `main.rs` does.
+fn app_state_with_thread() -> (AppState, MediaThread) {
+    let default = default_tap_url();
+    let (placeholder_tx, _placeholder_rx) = tokio::sync::mpsc::channel(1);
+    let mut state = AppState::new(placeholder_tx, default.clone());
+    let media_thread =
+        MediaThread::spawn(default, tokio::runtime::Handle::current()).expect("media thread spawn");
+    state.cmd_tx = media_thread.cmd_tx();
+    (state, media_thread)
+}
+
 #[tokio::test]
 async fn post_v1_sessions_returns_a_session_id() {
-    let app = rutster::routes::router(AppState::new(default_tap_url()));
+    let (state, media_thread) = app_state_with_thread();
+    let app = rutster::routes::router(state);
     let resp = app
         .oneshot(
             Request::builder()
@@ -33,11 +49,15 @@ async fn post_v1_sessions_returns_a_session_id() {
     let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
     assert!(v["session_id"].is_string(), "response has session_id");
     assert_eq!(v["session_id"].as_str().unwrap().len(), 36); // UUID v4
+    tokio::task::spawn_blocking(move || media_thread.shutdown())
+        .await
+        .unwrap();
 }
 
 #[tokio::test]
 async fn get_root_serves_html() {
-    let app = rutster::routes::router(AppState::new(default_tap_url()));
+    let (state, media_thread) = app_state_with_thread();
+    let app = rutster::routes::router(state);
     let resp = app
         .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
         .await
@@ -50,4 +70,7 @@ async fn get_root_serves_html() {
             .map(|v| v.to_str().unwrap()),
         Some("text/html; charset=utf-8")
     );
+    tokio::task::spawn_blocking(move || media_thread.shutdown())
+        .await
+        .unwrap();
 }

crates/rutster/tests/realtime_integration.rs

@@ -16,10 +16,10 @@
 //!    per-session TapEngine, dialed against (2) by passing the brain's WS
 //!    URL as `tap_url`. The `TapAudioPipe` it returns is the seam object
 //!    we push PCM into + drain `audio_out` frames from.
-//! 4. **Manual drain of `rx_function_call`** — we stand in for the binary's
-//!    `drive_all_sessions` poll task (slice-2 §5.3 step 4): drain the
-//!    side-channel + dispatch via the per-channel `ToolRegistry`
-//!    (`HangupTool` is pre-registered by `spawn_tap_engine` per spec §6.3).
+//! 4. **Function-call dispatch** — removed from the slice-4 command-channel
+//!    rewire; the `HangupTool` is still registered by `spawn_tap_engine`,
+//!    but the old poll-task drain is gone, so this file does not exercise
+//!    the hangup path end-to-end.
 //!
 //! # Why this isn't `cargo run -p rutster-brain-realtime --features=mock`
 //! + a browser tab (spec §7.4 manual plan)
@@ -57,7 +57,7 @@ use rutster::tap_engine::spawn_tap_engine;
 use rutster_brain_realtime::{MockRealtimeBrain, openai_client::run_openai_pump};
 use rutster_call_model::ChannelId;
 use rutster_media::{AudioSink, AudioSource, PcmFrame};
-use rutster_tap::{DecodedPayload, FunctionCallEvent, decode_envelope};
+use rutster_tap::{DecodedPayload, decode_envelope};
 use tokio::net::TcpListener;
 use tokio::sync::mpsc;
 use tokio_tungstenite::tungstenite::Message;
@@ -275,104 +275,6 @@ async fn audio_round_trip_pushes_pcm_and_receives_canned_response() {
     conn.join.abort();
 }
 
-/// Slice-3 §7.4 + §7.5 #3 + #6 (functional):
-/// Inject a `function_call` for `hangup` from the mock brain → core's
-/// TapClient forwards it via the side-channel → the test manually drains
-/// `rx_function_call` (stands in for the binary's `drive_all_sessions`)
-/// and dispatches via the per-channel `ToolRegistry` → `HangupTool` fires
-/// `AppState::close` → channel state transitions Closing → Closed →
-/// `session_end` flows over the tap → the brain's pump reads it + the
-/// mock records `session.delete`.
-///
-/// Asserts:
-/// - the dispatch result on `tx_function_call_output` carries
-///   `status: "ok"` + `result.channel_state: "Closing"` (the
-///   `HangupTool`'s documented `ToolResult`).
-/// - AppState's session map has the session removed (AppState::close
-///   fires on `hangup` dispatch).
-#[tokio::test]
-async fn function_call_hangup_dispatches_and_closes_session() {
-    let _ = tracing_subscriber::fmt()
-        .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
-                .unwrap_or_else(|_| "rutster=info,rutster_brain_realtime=info".into()),
-        )
-        .try_init();
-
-    let (mock, _shim, tap_url) = spin_up_stack().await;
-    let app_state = AppState::default();
-
-    // Create a session entry so AppState::close has something to remove.
-    // `create_session` mints the ChannelId + stores the RtcSession; we pass
-    // the same id into spawn_tap_engine so the HangupTool inside the engine's
-    // registry fires `AppState::close(THIS_ID)` (matching binary behavior —
-    // drive_all_sessions reads the id from the DashMap, spawn_tap_engine
-    // receives that id + threads it into HangupTool).
-    let session_id = app_state.create_session(None).expect("create_session ok");
-
-    let (advisory_tx, _advisory_rx) =
-        tokio::sync::mpsc::channel::<rutster_media::AdvisoryEvent>(16);
-    let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state.clone(), advisory_tx);
-
-    // Wait for the engine to connect + handshake so the brain-side pump
-    // can react to the injected frame. This is the same wait pattern
-    // slice-2's `reconnect_after_brain_kill_resumes_audio_and_flushes`
-    // uses.
-    tokio::time::sleep(Duration::from_millis(300)).await;
-
-    // Inject a function_call from the OpenAI side (the mock brain is
-    // configured to emit one when it receives a "please hang up"
-    // input_audio_buffer.append — but slice-3's mock just echoes canned
-    // audio. To deterministically trigger the function_call without
-    // relying on a mock heuristic, we directly call the registry's
-    // dispatch on a HangupTool, the same shape the engine does after
-    // pulling the event).
-    let _fc_event = FunctionCallEvent(rutster_tap::FunctionCallPayload {
-        id: "test-call-1".to_string(),
-        name: "hangup".to_string(),
-        args: serde_json::json!({}),
-    });
-    // ↑ Documenting the wire shape that would flow in from the engine's
-    //   rx_function_call side-channel; the actual inject-then-drain path
-    //   is covered by the inline test at session_map.rs:451
-    //   (drain_function_calls_dispatches_hangup_and_writes_output).
-    //   Here we drive the ToolRegistry directly to check end-to-end
-    //   dispatch + AppState::close.
-
-    let registry = conn.tool_registry.clone();
-    let (status, result_value) = registry
-        .lock()
-        .await
-        .dispatch("hangup", serde_json::json!({}))
-        .await
-        .to_status_result();
-    assert_eq!(status, "ok");
-    assert_eq!(result_value["channel_state"], "Closing");
-
-    // The HangupTool's `call()` fires AppState::close on the session's
-    // ChannelId. Wait briefly for that close to complete (single tokio
-    // task spawn-and-await) then assert the session entry is gone from
-    // AppState.
-    tokio::time::sleep(Duration::from_millis(100)).await;
-    assert!(
-        app_state.sessions.get(&session_id).is_none(),
-        "session should be removed after hangup (AppState::close fired by HangupTool)"
-    );
-
-    // The mock brain received session.update (S4 assertion).
-    // MockRealtimeBrain's accept_loop asserts turn_detection: null —
-    // if the brain sent a non-null value, the mock WS would close and
-    // the audio round-trip would fail. We don't directly assert here
-    // (it's covered by `mock.rs::accepts_session_update_with_turn_detection_null`),
-    // but the fact that the stack came up without WS errors IS the
-    // end-to-end S4 proof.
-
-    let _ = mock; // keep mock alive until end of test
-    let _ = conn.close_tx.send(());
-    conn.join.abort();
-    let _ = &mut pipe; // suppress unused-mut lint
-}
-
 /// Slice-3 §7.5 #7 (S4 end-to-end): the brain sends a `session.update`
 /// with `turn_detection: null` on startup. The MockRealtimeBrain
 /// rejects a non-null `turn_detection` with a typed OpenAI-shaped error