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"); + } }