test(slice-4): secondary-path barge-in e2e — brain advisory as backstop trigger

Quiet caller audio (sub-VAD-threshold) so the local VAD stays silent; the
MockRealtimeBrain advisory schedule emits speech_started after 2 appends
and the kill flows through slice-3's real advisory plumbing:
mock -> translator -> tap protocol -> tap_client -> advisory_tx -> Reflex.
Proves the ASR-VAD backstop path (spec §3.1, §6; done-criteria #8).

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

crates/rutster/tests/barge_in_integration.rs

@@ -1,16 +1,158 @@
 // crates/rutster/tests/barge_in_integration.rs
 //
 // PRIMARY-path barge-in e2e proof (slice-4 spec §3.4, §6.1, §7 done-criteria #8).
+// SECONDARY-path barge-in e2e proof (slice-4 spec §3.1, §6.1, §7 done-criteria #9).
 //
-// Constructs LocalVadReflex<Reflex<TapAudioPipe>> directly and proves that loud
+// The PRIMARY test constructs the reflex stack directly and proves that loud
-// caller audio kills playout within one tick, with only the local VAD as the
+// caller audio kills playout with only the local VAD as the trigger source — no
-// trigger source — no brain advisory is involved. This is wedge #1 from
+// brain advisory is involved. This is wedge #1 from README:98-100 and
-// README:98-100 and ARCHITECTURE.md:79-81.
+// ARCHITECTURE.md:79-81.
+//
+// The SECONDARY test exercises slice-3's advisory plumbing end-to-end:
+// MockRealtimeBrain -> translator -> tap protocol -> tap_client -> advisory_tx ->
+// Reflex. The local VAD is intentionally kept silent (zeroed caller frames), so
+// the kill can only be advisory-driven.
 
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use rutster::session_map::AppState;
+use rutster::tap_engine::spawn_tap_engine;
+use rutster_brain_realtime::mock::{AdvisoryKind, AdvisoryTrigger, MockRealtimeBrain};
+use rutster_brain_realtime::openai_client::run_openai_pump;
+use rutster_call_model::ChannelId;
 use rutster_media::{AudioSink, AudioSource, PcmFrame, VAD_DEBOUNCE_FRAMES};
 use rutster_media::{LocalVadReflex, Reflex, ReflexMetrics};
-use rutster_tap::{TapAudioPipe, TapMetrics};
+use rutster_tap::{DecodedPayload, TapAudioPipe, TapMetrics, decode_envelope};
+use tokio::net::TcpListener;
 use tokio::sync::mpsc;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tracing::info;
+use url::Url;
+
+// === Brain-shim helpers (duplicated from realtime_integration.rs) ===
+// The brain process's accept loop is inlined here so this test exercises the
+// real OpenAI-client pump (`run_openai_pump`) against the mock brain without
+// spawning a subprocess or depending on private helpers in another integration-
+// test binary.
+
+/// Handle returned by `start_brain_shim`. Drop to tear down.
+struct BrainShim {
+    addr: std::net::SocketAddr,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    join: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for BrainShim {
+    fn drop(&mut self) {
+        if let Some(tx) = self.shutdown.take() {
+            let _ = tx.send(());
+        }
+        self.join.abort();
+    }
+}
+
+/// Start an in-process brain-process-equivalent WS server.
+async fn start_brain_shim(mock_url: String) -> BrainShim {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+    let join = tokio::spawn(async move {
+        brain_accept_loop(listener, shutdown_rx, mock_url).await;
+    });
+    BrainShim {
+        addr,
+        shutdown: Some(shutdown_tx),
+        join,
+    }
+}
+
+/// Accept loop: spawns a per-connection task for each tap WS dial.
+async fn brain_accept_loop(
+    listener: TcpListener,
+    mut shutdown: tokio::sync::oneshot::Receiver<()>,
+    mock_url: String,
+) {
+    loop {
+        tokio::select! {
+            biased;
+            _ = &mut shutdown => {
+                info!("brain_shim accept loop shutting down");
+                return;
+            }
+            res = listener.accept() => {
+                let Ok((stream, peer)) = res else { continue };
+                let url = mock_url.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = handle_tap_connection(stream, peer, &url).await {
+                        info!(%peer, error = ?e, "brain_shim connection ended");
+                    }
+                });
+            }
+        }
+    }
+}
+
+/// Handle one tap WS connection: handshake, split sink+stream, dial the mock
+/// OpenAI side, and run `run_openai_pump`.
+async fn handle_tap_connection(
+    stream: tokio::net::TcpStream,
+    peer: std::net::SocketAddr,
+    openai_url: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut tap_ws = tokio_tungstenite::accept_async(stream).await?;
+    info!(%peer, "brain_shim tap WS connection accepted");
+
+    let hello_in = tap_ws
+        .next()
+        .await
+        .ok_or("tap connection closed before hello")??;
+    let hello_text = hello_in.into_text().map_err(|_| "hello not text")?;
+    let decoded = decode_envelope(&hello_text)?;
+    let session_id = match decoded.payload {
+        DecodedPayload::Hello(p) => p.session_id,
+        _ => return Err("first tap frame not hello".into()),
+    };
+    info!(%peer, %session_id, "brain_shim tap hello received");
+    let ack = rutster_tap::encode_hello(&session_id, 0, 0)?;
+    tap_ws.send(Message::Text(ack)).await?;
+
+    let (tap_via, pump_tap_rx) = mpsc::channel::<String>(64);
+    let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(64);
+    let (mut tap_sink, mut tap_stream) = tap_ws.split();
+
+    let in_fwd = tokio::spawn(async move {
+        while let Some(msg_res) = tap_stream.next().await {
+            if let Ok(m) = msg_res {
+                if let Ok(text) = m.into_text() {
+                    if tap_via.send(text).await.is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+    let out_fwd = tokio::spawn(async move {
+        while let Some(text) = tap_out_rx.recv().await {
+            if tap_sink.send(Message::Text(text)).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    let request = openai_url.into_client_request()?;
+    let (openai_ws, _resp) = tokio_tungstenite::connect_async(request).await?;
+    info!(%peer, %openai_url, "brain_shim OpenAI side connected");
+
+    let pump_result =
+        run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, "alloy".to_string()).await;
+    info!(%peer, ?pump_result, "brain_shim pump exited");
+    in_fwd.abort();
+    out_fwd.abort();
+    Ok(())
+}
 
 #[tokio::test]
 async fn primary_path_local_vad_kills_playout_without_brain() {
@@ -59,3 +201,111 @@ async fn primary_path_local_vad_kills_playout_without_brain() {
         "first fresh audio_out post-barge must resume playout"
     );
 }
+
+#[tokio::test]
+async fn secondary_path_brain_advisory_kills_playout_as_backstop() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "rutster=info".into()),
+        )
+        .try_init();
+
+    let mut mock = MockRealtimeBrain::start().await.expect("mock bind ok");
+    // Emit speech_started after the mock has received 2 audio_in frames.
+    mock.set_advisory_schedule(vec![AdvisoryTrigger {
+        after_audio_in_frames: 2,
+        event: AdvisoryKind::SpeechStarted,
+    }])
+    .await;
+
+    let shim = start_brain_shim(mock.url()).await;
+    let tap_url = Url::parse(&format!("ws://{}/", shim.addr)).unwrap();
+
+    let app_state = AppState::default();
+    let session_id = ChannelId::new();
+    let (advisory_tx, advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
+    let reflex_metrics = ReflexMetrics::new();
+    let (pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx.clone());
+    // Compose the same stack the media thread wires on Connected:
+    // LocalVadReflex (primary VAD) wrapping Reflex (advisory state machine).
+    let reflex = Reflex::new(pipe, advisory_rx, reflex_metrics.clone());
+    let mut stack = LocalVadReflex::new(reflex, advisory_tx);
+
+    // Give the engine a moment to complete the tap handshake and dial the mock.
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    let quiet = PcmFrame::zeroed();
+
+    // Step 1 — prove the brain→core audio_out path is alive. Send one quiet
+    // frame (below the local VAD threshold, so it cannot trip the primary
+    // trigger) and wait for the mock's canned response.audio.delta.
+    stack.on_pcm_frame(quiet.clone());
+    let mut audio_path_alive = false;
+    for _ in 0..25 {
+        if stack.next_pcm_frame().is_some() {
+            audio_path_alive = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        audio_path_alive,
+        "mock brain must deliver at least one audio_out frame before the kill"
+    );
+
+    // Step 2 — trigger the advisory schedule. The mock counts
+    // input_audio_buffer.append events; this second append reaches the
+    // threshold and causes the mock to emit speech_started unprompted.
+    stack.on_pcm_frame(quiet.clone());
+
+    // Step 3 — bounded wait for the advisory to cross the real plumbing and
+    // kill playout.
+    let kill_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    let mut killed = false;
+    while tokio::time::Instant::now() < kill_deadline {
+        stack.next_pcm_frame();
+        if reflex_metrics.barge_in_count.load(Ordering::Relaxed) == 1 {
+            killed = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        killed,
+        "brain advisory must trigger barge-in within the bounded window"
+    );
+    assert_eq!(
+        reflex_metrics.barge_in_count.load(Ordering::Relaxed),
+        1,
+        "advisory-driven barge-in must fire exactly once"
+    );
+
+    // Playout is killed: even though the ring may still have brain frames in
+    // flight, barge_in_flush cleared them before the Reflex returned None.
+    assert!(
+        stack.next_pcm_frame().is_none(),
+        "playout must stay killed after the advisory"
+    );
+
+    // Step 4 — resume. Another quiet frame prompts the mock to send a fresh
+    // audio_out delta; the first post-barge audio_out un-mutes playout.
+    let resume_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    let mut resumed = false;
+    while tokio::time::Instant::now() < resume_deadline {
+        stack.on_pcm_frame(quiet.clone());
+        if stack.next_pcm_frame().is_some() {
+            resumed = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        resumed,
+        "first fresh audio_out post-barge must resume playout"
+    );
+
+    // Teardown — close the engine; the shim + mock drop on scope exit.
+    let _ = conn.close_tx.send(());
+    conn.join.abort();
+}