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

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,
 }

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

crates/rutster-tap/src/tap_client.rs

@@ -103,6 +103,7 @@ pub async fn run_tap_client<T>(
     tx_audio_out: mpsc::Sender<PcmFrame>,
     tx_function_call: mpsc::Sender<FunctionCallEvent>,
     rx_function_call_output: &mut mpsc::Receiver<FunctionCallOutputEvent>,
+    tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent>,
     metrics: Arc<TapMetrics>,
     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<u64>,
     tx_audio_out: &mpsc::Sender<PcmFrame>,
     tx_function_call: &mpsc::Sender<FunctionCallEvent>,
+    tx_advisory: &mpsc::Sender<rutster_media::AdvisoryEvent>,
     metrics: &Arc<TapMetrics>,
     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::<FunctionCallEvent>(8);
         let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
+        let (tx_advisory, _rx_advisory) = mpsc::channel::<rutster_media::AdvisoryEvent>(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::<FunctionCallEvent>(8);
         let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
+        let (tx_advisory, mut rx_advisory) = mpsc::channel::<rutster_media::AdvisoryEvent>(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<u64> = 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));
     }
 }

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::<rutster_media::AdvisoryEvent>(16);
+                let (pipe, conn) = spawn_tap_engine(id, tap_url_clone, app_state, advisory_tx);
                 s.set_pipe(pipe);
                 s.channel.tap = Some(TapHandle::new());
                 info!(channel_id = %id, "tap engine spawned on Connected");

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<rutster_media::AdvisoryEvent>,
 ) -> (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<FunctionCallEvent>,
     mut rx_function_call_output: mpsc::Receiver<FunctionCallOutputEvent>,
+    tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent>,
     metrics: Arc<TapMetrics>,
 ) {
     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::<rutster_media::AdvisoryEvent>(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::<rutster_media::AdvisoryEvent>(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::<rutster_media::AdvisoryEvent>(16);
+        let (_pipe, conn) = spawn_tap_engine(
+            id,
+            url,
+            crate::session_map::AppState::default(),
+            advisory_tx,
+        );
+        let _ = conn.close_tx.send(());
+        conn.join.abort();
+    }
 }

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::<rutster_media::AdvisoryEvent>(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::<rutster_media::AdvisoryEvent>(16);
+    let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state.clone(), advisory_tx);
 
     // Wait for the engine to connect + handshake so the brain-side pump
     // can react to the injected frame. This is the same wait pattern
@@ -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::<rutster_media::AdvisoryEvent>(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

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::<rutster_media::AdvisoryEvent>(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