feat(brain-realtime): MockRealtimeBrain advisory schedule (slice-4 §4.5)

The mock can now emit speech_started/speech_stopped on a programmable
schedule keyed by input_audio_buffer.append count. Used by the slice-4
secondary-path e2e to exercise the brain advisory → reflex path. The
schedule is shared via an Arc<Mutex<_>> so set_advisory_schedule right
after start() is picked up by accepted connections.

Signed-off-by: Aaron D. Lee <himself@adlee.work>

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<oneshot::Sender<()>>,
     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<Mutex<Vec<AdvisoryTrigger>>>,
 }
 
 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<AdvisoryTrigger>) {
+        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<Mutex<Vec<AdvisoryTrigger>>>,
+) {
     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<Mutex<Vec<AdvisoryTrigger>>>,
+) {
     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");
+    }
 }