From 3c63eb6688b3620b3ed7eaf62404f16473270082 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 04:37:28 +0000 Subject: [PATCH] slice-4 (dev-a): MediaThread + session_map rewire (Task 6 + 7) (#12) Co-authored-by: Aaron D. Lee Co-committed-by: Aaron D. Lee --- crates/rutster/src/lib.rs | 1 + crates/rutster/src/main.rs | 23 +- crates/rutster/src/media_thread.rs | 332 +++++++++++ crates/rutster/src/routes.rs | 16 +- crates/rutster/src/session_map.rs | 555 +++---------------- crates/rutster/tests/api_integration.rs | 27 +- crates/rutster/tests/realtime_integration.rs | 108 +--- 7 files changed, 462 insertions(+), 600 deletions(-) create mode 100644 crates/rutster/src/media_thread.rs diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 6f98e10..0d89088 100644 --- a/crates/rutster/src/lib.rs +++ b/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; diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index db17b4e..b72f0d7 100644 --- a/crates/rutster/src/main.rs +++ b/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() diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs new file mode 100644 index 0000000..26c7406 --- /dev/null +++ b/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` 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` 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>, + }, + /// 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>, + }, + /// 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, + join: Option>, +} + +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 { + 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` 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 { + 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` wrapper. + tap_conn: Option, +} + +fn run_media_thread( + mut cmd_rx: mpsc::Receiver, + default_tap_url: url::Url, + tokio_handle: tokio::runtime::Handle, + media_cmd_tx: mpsc::Sender, +) { + let mut sessions: HashMap = 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 = 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::(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 (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(); + } +} diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs index c464755..cea1551 100644 --- a/crates/rutster/src/routes.rs +++ b/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() + } } } } diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 26c72b5..05eb55d 100644 --- a/crates/rutster/src/session_map.rs +++ b/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` 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` +//! 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` + 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`. 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>` 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>, - pub tap_url: url::Url, - pub tap_conn: Option, -} - -/// 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>, - pub poll_running: Arc>, - /// 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, 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, 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, - ) -> Result { - let session = RtcSession::new()?; - let id = session.channel_id(); + ) -> Result { 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>> { - 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 { + 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 = 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::(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; - let tool_registry: Arc>; - let tx_out: Option>; - { - 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`; `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::(8); - let (tx_fco, mut rx_fco) = tokio::sync::mpsc::channel::(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"), + ) } } diff --git a/crates/rutster/tests/api_integration.rs b/crates/rutster/tests/api_integration.rs index 377824b..70081b8 100644 --- a/crates/rutster/tests/api_integration.rs +++ b/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(); } diff --git a/crates/rutster/tests/realtime_integration.rs b/crates/rutster/tests/realtime_integration.rs index d2f8755..3d95e71 100644 --- a/crates/rutster/tests/realtime_integration.rs +++ b/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::(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