From 276dc5ac45f93af1874d7956cd45b9eff48a822a Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 01:43:41 +0000 Subject: [PATCH] slice-4 (dev-b): TapAudioPipe::barge_in_flush + advisory_tx + MockRealtimeBrain schedule (#9) Co-authored-by: Aaron D. Lee Co-committed-by: Aaron D. Lee --- crates/rutster-brain-realtime/src/mock.rs | 127 ++++++++++++++++++- crates/rutster-tap/src/metrics.rs | 7 + crates/rutster-tap/src/tap_audio_pipe.rs | 74 +++++++++++ crates/rutster-tap/src/tap_client.rs | 96 ++++++++++---- crates/rutster/src/session_map.rs | 7 +- crates/rutster/src/tap_engine.rs | 38 +++++- crates/rutster/tests/realtime_integration.rs | 12 +- crates/rutster/tests/tap_integration.rs | 4 +- 8 files changed, 329 insertions(+), 36 deletions(-) diff --git a/crates/rutster-brain-realtime/src/mock.rs b/crates/rutster-brain-realtime/src/mock.rs index e7df45d..e8cc7c4 100644 --- a/crates/rutster-brain-realtime/src/mock.rs +++ b/crates/rutster-brain-realtime/src/mock.rs @@ -28,15 +28,32 @@ //! hook, not via the mock's own drive). use std::net::SocketAddr; +use std::sync::Arc; use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, json}; use tokio::net::TcpListener; -use tokio::sync::oneshot; +use tokio::sync::{Mutex, oneshot}; use tokio::task::JoinHandle; use tokio_tungstenite::tungstenite::Message; use tracing::{debug, info, warn}; +/// A trigger for the advisory schedule. The mock counts +/// `input_audio_buffer.append` events; when the count reaches +/// `after_audio_in_frames`, it emits `event` unprompted (simulating +/// the brain's VAD firing). +#[derive(Debug, Clone)] +pub struct AdvisoryTrigger { + pub after_audio_in_frames: u32, + pub event: AdvisoryKind, +} + +#[derive(Debug, Clone, Copy)] +pub enum AdvisoryKind { + SpeechStarted, + SpeechStopped, +} + /// The handle returned by `MockRealtimeBrain::start`. Drop or call `shutdown` /// to stop the mock server + abort its accept-loop task. pub struct MockRealtimeBrain { @@ -46,6 +63,10 @@ pub struct MockRealtimeBrain { pub addr: SocketAddr, pub shutdown: Option>, pub join: JoinHandle<()>, + /// slice-4: shared advisory schedule. Created empty in `start` and + /// cloned to every accepted connection; `set_advisory_schedule` + /// mutates it in place so live connections pick it up. + advisory_schedule: Arc>>, } impl MockRealtimeBrain { @@ -56,12 +77,14 @@ impl MockRealtimeBrain { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); - let join = tokio::spawn(accept_loop(listener, shutdown_rx)); + let schedule = Arc::new(Mutex::new(Vec::new())); + let join = tokio::spawn(accept_loop(listener, shutdown_rx, schedule.clone())); info!(%addr, "MockRealtimeBrain listening (fake OpenAI Realtime)"); Ok(Self { addr, shutdown: Some(shutdown_tx), join, + advisory_schedule: schedule, }) } @@ -72,6 +95,14 @@ impl MockRealtimeBrain { pub fn url(&self) -> String { format!("ws://{}/v1/realtime?model=mock", self.addr) } + + /// Set a schedule of advisory events the mock emits UNPROMPTED after + /// observing N `input_audio_buffer.append` events. Used by the slice-4 + /// barge-in e2e test to drive the reflex via the secondary trigger path. + pub async fn set_advisory_schedule(&mut self, schedule: Vec) { + let mut locked = self.advisory_schedule.lock().await; + *locked = schedule; + } } impl Drop for MockRealtimeBrain { @@ -86,7 +117,11 @@ impl Drop for MockRealtimeBrain { } } -async fn accept_loop(listener: TcpListener, mut shutdown: oneshot::Receiver<()>) { +async fn accept_loop( + listener: TcpListener, + mut shutdown: oneshot::Receiver<()>, + schedule: Arc>>, +) { loop { tokio::select! { biased; @@ -102,7 +137,8 @@ async fn accept_loop(listener: TcpListener, mut shutdown: oneshot::Receiver<()>) continue; } }; - tokio::spawn(handle_connection(stream, peer)); + let schedule = schedule.clone(); + tokio::spawn(handle_connection(stream, peer, schedule)); } } } @@ -111,7 +147,11 @@ async fn accept_loop(listener: TcpListener, mut shutdown: oneshot::Receiver<()>) /// Handle one brain-process → mock-OpenAI WS connection. Each connection is /// one OpenAI Realtime session (stateless across reconnects — same as the /// brain process's own contract for the tap side). -async fn handle_connection(stream: tokio::net::TcpStream, peer: SocketAddr) { +async fn handle_connection( + stream: tokio::net::TcpStream, + peer: SocketAddr, + schedule: Arc>>, +) { let ws = tokio_tungstenite::accept_async(stream).await; let Ok(mut ws) = ws else { warn!(%peer, "MockRealtimeBrain WS handshake failed"); @@ -119,6 +159,9 @@ async fn handle_connection(stream: tokio::net::TcpStream, peer: SocketAddr) { }; debug!(%peer, "MockRealtimeBrain connection accepted"); + // slice-4: per-connection counter for advisory schedule triggers. + let mut audio_in_count: u32 = 0; + // First event MUST be session.update with turn_detection: null. // Wait for it (bounded 2s — the brain process sends it immediately // after the WS handshake per `run_openai_pump`). @@ -191,7 +234,7 @@ async fn handle_connection(stream: tokio::net::TcpStream, peer: SocketAddr) { match event_type { "input_audio_buffer.append" => { // Canned response: one `response.audio.delta` per append - // carrying 480 zeroed samples as base64 LE i16 (the same + // carrying 480 zeroed samples as base64 LE i16 PCM (the same // wire shape as slice-2's audio_out — verified by the // translator's existing round-trip test). Identical bytes // every time: deterministic for test assertions. @@ -206,6 +249,25 @@ async fn handle_connection(stream: tokio::net::TcpStream, peer: SocketAddr) { debug!(%peer, error = ?e, "MockRealtimeBrain send delta failed; closing"); return; } + + // slice-4: count the append + emit any scheduled advisories. + audio_in_count += 1; + for trigger in schedule.lock().await.iter() { + if trigger.after_audio_in_frames == audio_in_count { + let evt = match trigger.event { + AdvisoryKind::SpeechStarted => { + json!({ "type": "input_audio_buffer.speech_started" }) + } + AdvisoryKind::SpeechStopped => { + json!({ "type": "input_audio_buffer.speech_stopped" }) + } + }; + if let Err(e) = ws.send(Message::Text(evt.to_string())).await { + debug!(%peer, error = ?e, "MockRealtimeBrain send advisory failed; closing"); + return; + } + } + } } "input_audio_buffer.speech_started" => { // Pass-through echo so the brain's translator forwards it as @@ -325,4 +387,57 @@ mod tests { assert_eq!(err["type"], "error"); assert_eq!(err["error"]["code"], "invalid_session_update"); } + + /// slice-4: MockRealtimeBrain can emit `speech_started`/`speech_stopped` + /// on a programmable schedule, simulating the brain's VAD firing. This + /// is what the slice-4 barge-in e2e test drives. + #[tokio::test] + async fn emits_speech_started_on_schedule_after_n_audio_in_frames() { + let mut mock = MockRealtimeBrain::start().await.unwrap(); + mock.set_advisory_schedule(vec![ + AdvisoryTrigger { + after_audio_in_frames: 2, + event: AdvisoryKind::SpeechStarted, + }, + AdvisoryTrigger { + after_audio_in_frames: 4, + event: AdvisoryKind::SpeechStopped, + }, + ]) + .await; + let url = mock.url(); + let req = url.as_str().into_client_request().unwrap(); + let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap(); + + // Send session.update first (the mock's contract). + let session_update = json!({ + "type": "session.update", + "session": { "turn_detection": null } + }); + ws.send(Message::Text(session_update.to_string())) + .await + .unwrap(); + + // Send 2 audio_in appends → expect a speech_started. + for _ in 0..2 { + let append = json!({ "type": "input_audio_buffer.append", "audio": "AAAA" }); + ws.send(Message::Text(append.to_string())).await.unwrap(); + } + + // Skip the canned response.audio.delta replies; wait for speech_started. + let mut saw_started = false; + for _ in 0..10 { + let msg = tokio::time::timeout(Duration::from_millis(500), ws.next()) + .await + .expect("event within 500ms") + .unwrap() + .unwrap(); + let text = msg.into_text().unwrap(); + if text.contains("speech_started") { + saw_started = true; + break; + } + } + assert!(saw_started, "mock must emit speech_started after N appends"); + } } diff --git a/crates/rutster-tap/src/metrics.rs b/crates/rutster-tap/src/metrics.rs index b0a1e73..dc7cbca 100644 --- a/crates/rutster-tap/src/metrics.rs +++ b/crates/rutster-tap/src/metrics.rs @@ -25,6 +25,11 @@ pub struct TapMetrics { pub unknown_frames: AtomicU64, pub malformed_frames: AtomicU64, pub reconnect_attempts: AtomicU64, + /// slice-4 §3.3: count of in-flight `audio_out` frames dropped from + /// `rx_audio_out` during `barge_in_flush`. The drain makes the resume + /// condition race-free — the first `audio_out` observed post-barge is + /// provably post-barge. + pub barge_drained_inflight: AtomicU64, } impl TapMetrics { @@ -45,6 +50,7 @@ impl TapMetrics { unknown_frames: self.unknown_frames.load(Ordering::Relaxed), malformed_frames: self.malformed_frames.load(Ordering::Relaxed), reconnect_attempts: self.reconnect_attempts.load(Ordering::Relaxed), + barge_drained_inflight: self.barge_drained_inflight.load(Ordering::Relaxed), } } } @@ -59,4 +65,5 @@ pub struct MetricsSnapshot { pub unknown_frames: u64, pub malformed_frames: u64, pub reconnect_attempts: u64, + pub barge_drained_inflight: u64, } diff --git a/crates/rutster-tap/src/tap_audio_pipe.rs b/crates/rutster-tap/src/tap_audio_pipe.rs index 91e2bd9..98b5955 100644 --- a/crates/rutster-tap/src/tap_audio_pipe.rs +++ b/crates/rutster-tap/src/tap_audio_pipe.rs @@ -130,11 +130,36 @@ impl rutster_media::AudioPipe for TapAudioPipe { debug!(cleared, "playout ring flushed on brain disconnect"); } } + + /// slice-4 spec §3.3 — barge-in flush: clear the playout ring AND + /// drain `rx_audio_out` of any frames queued before the barge. Without + /// this drain, a stale brain frame in the mpsc would un-mute + /// immediately on the next tick — defeating the "first fresh audio_out" + /// resume condition. Hot-path: try_recv loop, bounded, no blocking. + fn barge_in_flush(&mut self) { + let cleared = self.playout_ring.len(); + self.playout_ring.clear(); + if cleared > 0 { + debug!(cleared, "playout ring flushed on barge-in"); + } + + let mut drained = 0usize; + while self.rx_audio_out.try_recv().is_ok() { + drained += 1; + } + if drained > 0 { + self.metrics + .barge_drained_inflight + .fetch_add(drained as u64, Ordering::Relaxed); + debug!(drained, "in-flight brain frames drained on barge-in"); + } + } } #[cfg(test)] mod tests { use super::*; + use rutster_media::AudioPipe; #[allow(clippy::type_complexity)] // test helper: 5-tuple of channel ends; not worth a struct. fn channels() -> ( @@ -229,4 +254,53 @@ mod tests { // Now next_pcm_frame should return None (silence) — Disconnected path. assert!(pipe.next_pcm_frame().is_none()); } + + /// slice-4 §3.3: barge_in_flush clears the playout ring AND drains the + /// inbound `rx_audio_out` mpsc of any frames queued before the barge. + /// Without draining the mpsc, a stale pre-barge frame would un-mute + /// immediately on the next tick — defeating the "first fresh audio_out" + /// resume condition. + #[test] + fn barge_in_flush_clears_ring_and_drains_rx_audio_out() { + let (_tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone()); + + // Put some frames into the playout ring first (simulating steady-state + // playout that already drained from the mpsc). + for i in 0..2 { + let mut f = PcmFrame::zeroed(); + f.samples[0] = i as i16; + tx_audio_out.blocking_send(f).unwrap(); + } + let _ = pipe.next_pcm_frame(); // drains mpsc into ring, pops first + + // Now queue MORE frames into rx_audio_out that have NOT been pulled + // into the ring yet — these are the "in-flight" stale frames that + // must be drained during a barge-in to keep resume race-free. + for i in 0..3 { + let mut f = PcmFrame::zeroed(); + f.samples[0] = (10 + i) as i16; + tx_audio_out.blocking_send(f).unwrap(); + } + + // Barge-in: ring should clear + the 3 mpsc frames drain. + pipe.barge_in_flush(); + + assert!(pipe.next_pcm_frame().is_none()); + assert_eq!( + metrics.barge_drained_inflight.load(Ordering::Relaxed), + 3, + "three in-flight mpsc frames should be drained on barge-in" + ); + } + + /// slice-4 §3.3: barge_in_flush when empty is a no-op and leaves the + /// counter at zero. + #[test] + fn barge_in_flush_when_already_empty_is_noop() { + let (_tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(_tx_pcm_in, rx_audio_out, metrics.clone()); + pipe.barge_in_flush(); + assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 0); + } } diff --git a/crates/rutster-tap/src/tap_client.rs b/crates/rutster-tap/src/tap_client.rs index ed906d4..5c7d39f 100644 --- a/crates/rutster-tap/src/tap_client.rs +++ b/crates/rutster-tap/src/tap_client.rs @@ -103,6 +103,7 @@ pub async fn run_tap_client( tx_audio_out: mpsc::Sender, tx_function_call: mpsc::Sender, rx_function_call_output: &mut mpsc::Receiver, + tx_advisory: mpsc::Sender, metrics: Arc, close: &mut oneshot::Receiver<()>, ) -> Result<(), TapClientError> @@ -260,9 +261,15 @@ where // silently dropped (v1 is text-JSON only — spec §3.4). if let Ok(text) = msg.into_text() { handle_brain_frame( - &text, &mut last_seq_ingress, &tx_audio_out, - &tx_function_call, &metrics, session_start, - ).await; + &text, + &mut last_seq_ingress, + &tx_audio_out, + &tx_function_call, + &tx_advisory, + &metrics, + session_start, + ) + .await; } } } @@ -325,6 +332,7 @@ async fn handle_brain_frame( last_seq_ingress: &mut Option, tx_audio_out: &mpsc::Sender, tx_function_call: &mpsc::Sender, + tx_advisory: &mpsc::Sender, metrics: &Arc, session_start: Instant, ) { @@ -403,12 +411,27 @@ async fn handle_brain_frame( metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); warn!("unexpected function_call_output from brain; dropping"); } - // Slice-3 advisory — same "logged + counted, not forwarded" posture - // as `Unknown`. The FOB reflex loop in step 4 will act on these; - // slice-3 only pre-paves the wire event. - DecodedPayload::SpeechStarted | DecodedPayload::SpeechStopped => { - metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); - debug!("advisory interruption event observed; not acted on in slice-3"); + // slice-4: advisory events forward to the Reflex via the dedicated + // `advisory_tx` channel. The FOB reflex is authoritative: a local + // in-core VAD is the PRIMARY trigger, and the brain's ASR-grade + // advisory is the slower SECONDARY/confirmation trigger (~300 ms + // later). Both sources feed the same advisory mpsc; `Reflex` drains + // them uniformly on the 20 ms tick. + DecodedPayload::SpeechStarted => { + let ev = rutster_media::AdvisoryEvent::SpeechStarted { at: Instant::now() }; + if tx_advisory.try_send(ev).is_err() { + // Channel full → drop + observe (hot-path policy). The + // Reflex counts dropped advisories in its own metrics. + metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed); + warn!("advisory SpeechStarted dropped (advisory_tx full)"); + } + } + DecodedPayload::SpeechStopped => { + let ev = rutster_media::AdvisoryEvent::SpeechStopped { at: Instant::now() }; + if tx_advisory.try_send(ev).is_err() { + metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed); + warn!("advisory SpeechStopped dropped (advisory_tx full)"); + } } DecodedPayload::ToolsUpdate(_) => { metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); @@ -482,6 +505,7 @@ mod tests { async fn handle_brain_frame_forwards_function_call_to_side_channel() { let (tx_fc, mut rx_fc) = mpsc::channel::(8); let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let (tx_advisory, _rx_advisory) = mpsc::channel::(8); let metrics = Arc::new(TapMetrics::new()); // Build a wire function_call frame: id="call-1", name="hangup", args={}. @@ -493,6 +517,7 @@ mod tests { &mut last_seq, &tx_audio_out, &tx_fc, + &tx_advisory, &metrics, Instant::now(), ) @@ -511,18 +536,18 @@ mod tests { assert_eq!(last_seq, Some(1)); } - /// slice-3 spec §5.2 — the *advisory* interrupt events (`speech_started` - /// /`speech_stopped`) and `tools.update` are observed (logged + counted) - /// but do NOT flow through the function_call side-channel (only - /// `function_call` does — that's the only event with a binary-side - /// disposal). This pins that boundary: an advisory event must NOT - /// produce a `FunctionCallEvent` even with the channel plumbed. + /// slice-4: `speech_started`/`speech_stopped` advisories now flow to the + /// dedicated `advisory_tx` side-channel for the Reflex to drain, and + /// STILL do NOT flow through the function_call side-channel (different + /// bus). This pins that boundary. #[tokio::test] - async fn advisory_events_are_logged_not_forwarded_to_function_call_channel() { + async fn advisory_events_forwarded_to_advisory_channel_only() { let (tx_fc, mut rx_fc) = mpsc::channel::(8); let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let (tx_advisory, mut rx_advisory) = mpsc::channel::(8); let metrics = Arc::new(TapMetrics::new()); + // speech_started forwards to advisory_tx. let wire = crate::protocol::encode_speech_started(2, 200).unwrap(); let mut last_seq: Option = None; handle_brain_frame( @@ -530,23 +555,48 @@ mod tests { &mut last_seq, &tx_audio_out, &tx_fc, + &tx_advisory, &metrics, Instant::now(), ) .await; - - // No FunctionCallEvent forwarded — the channel stays empty. Pick a - // tight bounded receive so the test fails fast if a future refactor - // starts forwarding advisory events here. + let advisory = tokio::time::timeout(Duration::from_millis(200), rx_advisory.recv()) + .await + .expect("advisory drained within 200ms") + .expect("channel not closed"); + assert!(matches!( + advisory, + rutster_media::AdvisoryEvent::SpeechStarted { .. } + )); + // function_call channel stays empty. assert!( tokio::time::timeout(Duration::from_millis(50), rx_fc.recv()) .await .is_err(), "no FunctionCallEvent expected for advisory events" ); - // The advisory event IS still observed via metrics (seq gap tracking - // + the unknown-slot counter remains 0 — speech_started is now a - // known payload variant). assert_eq!(last_seq, Some(2)); + + // speech_stopped forwards to advisory_tx. + let wire = crate::protocol::encode_speech_stopped(3, 300).unwrap(); + handle_brain_frame( + &wire, + &mut last_seq, + &tx_audio_out, + &tx_fc, + &tx_advisory, + &metrics, + Instant::now(), + ) + .await; + let advisory = tokio::time::timeout(Duration::from_millis(200), rx_advisory.recv()) + .await + .expect("advisory drained within 200ms") + .expect("channel not closed"); + assert!(matches!( + advisory, + rutster_media::AdvisoryEvent::SpeechStopped { .. } + )); + assert_eq!(last_seq, Some(3)); } } diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index d538fc4..26c72b5 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -315,7 +315,12 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { // slice-3 — §6.3) bound to this AppState + ChannelId. let app_state = state.clone(); let tap_url_clone = tap_url.clone(); - let (pipe, conn) = spawn_tap_engine(id, tap_url_clone, app_state); + // 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"); diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index bb8a580..94e30fc 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -132,6 +132,7 @@ pub fn spawn_tap_engine( session_id: ChannelId, tap_url: Url, app_state: crate::session_map::AppState, + advisory_tx: mpsc::Sender, ) -> (TapAudioPipe, TapConn) { // Two mpsc channels. The naming convention is "from the engine's POV": // - `tx_pcm_in`/`rx_pcm_in`: peer PCM flowing INTO the engine (sink side @@ -193,6 +194,7 @@ pub fn spawn_tap_engine( flush_tx, tx_function_call, rx_function_call_output, + advisory_tx, metrics, ) .await; @@ -246,6 +248,7 @@ async fn run_engine_loop( flush_tx: mpsc::Sender<()>, tx_function_call: mpsc::Sender, mut rx_function_call_output: mpsc::Receiver, + tx_advisory: mpsc::Sender, metrics: Arc, ) { let mut backoff = Backoff::default(); @@ -290,6 +293,7 @@ async fn run_engine_loop( tx_audio_out.clone(), tx_function_call.clone(), &mut rx_function_call_output, + tx_advisory.clone(), metrics.clone(), &mut close, ) @@ -487,7 +491,13 @@ mod tests { // increment. We abort the task on test drop to avoid leak. let id = ChannelId::new(); let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // port 1 = unreachable - let (mut pipe, conn) = spawn_tap_engine(id, url, crate::session_map::AppState::default()); + let (advisory_tx, _advisory_rx) = mpsc::channel::(16); + let (mut pipe, conn) = spawn_tap_engine( + id, + url, + crate::session_map::AppState::default(), + advisory_tx, + ); // TapAudioPipe is the seam object — should default to silent underflow. assert!(pipe.next_pcm_frame().is_none()); // TapConn carries the close oneshot + JoinHandle + metrics. @@ -506,7 +516,13 @@ mod tests { async fn spawn_returns_tap_conn_with_function_call_side_channels() { let id = ChannelId::new(); let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // unreachable brain - let (_pipe, conn) = spawn_tap_engine(id, url, crate::session_map::AppState::default()); + let (advisory_tx, _advisory_rx) = mpsc::channel::(16); + let (_pipe, conn) = spawn_tap_engine( + id, + url, + crate::session_map::AppState::default(), + advisory_tx, + ); // rx_function_call: Some(Receiver) — engine owns the paired Sender. assert!( @@ -528,4 +544,22 @@ mod tests { let _ = conn.close_tx.send(()); conn.join.abort(); } + + /// slice-4: spawn_tap_engine takes advisory_tx as a parameter because + /// the media thread owns the channel and needs to clone the sender to + /// both the engine and the LocalVadReflex wrapper. + #[tokio::test] + async fn spawn_accepts_advisory_tx_parameter() { + let id = ChannelId::new(); + let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); + let (advisory_tx, _advisory_rx) = mpsc::channel::(16); + let (_pipe, conn) = spawn_tap_engine( + id, + url, + crate::session_map::AppState::default(), + advisory_tx, + ); + let _ = conn.close_tx.send(()); + conn.join.abort(); + } } diff --git a/crates/rutster/tests/realtime_integration.rs b/crates/rutster/tests/realtime_integration.rs index 65db97e..d2f8755 100644 --- a/crates/rutster/tests/realtime_integration.rs +++ b/crates/rutster/tests/realtime_integration.rs @@ -261,7 +261,9 @@ async fn audio_round_trip_pushes_pcm_and_receives_canned_response() { let (_mock, _shim, tap_url) = spin_up_stack().await; let app_state = AppState::default(); let session_id = ChannelId::new(); - let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state); + let (advisory_tx, _advisory_rx) = + tokio::sync::mpsc::channel::(16); + let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx); let frame = push_pcm_and_wait_audio_out(&mut pipe).await; // MockRealtimeBrain sends 480 zeroed samples per response.audio.delta. @@ -308,7 +310,9 @@ async fn function_call_hangup_dispatches_and_closes_session() { // receives that id + threads it into HangupTool). let session_id = app_state.create_session(None).expect("create_session ok"); - let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state.clone()); + 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 @@ -395,7 +399,9 @@ async fn s4_brain_sends_session_update_with_turn_detection_null_end_to_end() { let (_mock, _shim, tap_url) = spin_up_stack().await; let app_state = AppState::default(); let session_id = ChannelId::new(); - let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state); + let (advisory_tx, _advisory_rx) = + tokio::sync::mpsc::channel::(16); + let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx); // If the brain sent turn_detection != null, MockRealtimeBrain would // close the OpenAI WS — the brain's pump would exit + the engine's diff --git a/crates/rutster/tests/tap_integration.rs b/crates/rutster/tests/tap_integration.rs index 634ee19..3684ec4 100644 --- a/crates/rutster/tests/tap_integration.rs +++ b/crates/rutster/tests/tap_integration.rs @@ -78,7 +78,9 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() { // Fix-3 playout-ring-flush contract end-to-end. let session_id = ChannelId::new(); let app_state = AppState::default(); - let (mut pipe, mut conn) = spawn_tap_engine(session_id, url, app_state); + let (advisory_tx, _advisory_rx) = + tokio::sync::mpsc::channel::(16); + let (mut pipe, mut conn) = spawn_tap_engine(session_id, url, app_state, advisory_tx); // 3. Push TWO frames with the same marker `samples[0] = 7` back-to-back // before the kill so the playout ring has buffered content that the