From d696536bdd6834500805eacc2dcb90836a594cc9 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 17:52:26 +0000 Subject: [PATCH] slice-4 (finisher): secondary-path e2e + CI seam gate (Tasks 9.2 + 10) (#13) Co-authored-by: Aaron D. Lee Co-committed-by: Aaron D. Lee --- .github/workflows/ci.yml | 23 ++ crates/rutster/tests/barge_in_integration.rs | 260 ++++++++++++++++++- 2 files changed, 278 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 62a7815..774c338 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,29 @@ jobs: toolchain: [stable, "1.85"] steps: - uses: actions/checkout@v4 + + # Seam gate (slice-4 §7 #3 / PORT_PLAN phasing): + # + # `loop_driver.rs` and `rtc_session.rs` are the hot-path media seam + # preserved from slice-3. They MUST remain byte-identical across + # slice-4 so the reflex wrapper can be added without touching the + # core poll loop. If a legitimate change is needed, update the pinned + # blob hashes below in the same PR — loud and reviewable by design. + - name: Seam gate — loop_driver + rtc_session byte-identical to slice-3 + run: | + EXPECTED_LOOP_DRIVER='744bf314edf7f4925c8bb3bd0f5176dbc88f8113' + EXPECTED_RTC_SESSION='a4c9f2ae64e56c08e1990956391514929535b526' + GOT_LOOP_DRIVER=$(git hash-object crates/rutster-media/src/loop_driver.rs) + GOT_RTC_SESSION=$(git hash-object crates/rutster-media/src/rtc_session.rs) + if [ "$GOT_LOOP_DRIVER" != "$EXPECTED_LOOP_DRIVER" ]; then + echo "::error::loop_driver.rs blob mismatch: $GOT_LOOP_DRIVER != $EXPECTED_LOOP_DRIVER" + exit 1 + fi + if [ "$GOT_RTC_SESSION" != "$EXPECTED_RTC_SESSION" ]; then + echo "::error::rtc_session.rs blob mismatch: $GOT_RTC_SESSION != $EXPECTED_RTC_SESSION" + exit 1 + fi + - uses: dtolnay/rust-toolchain@master with: toolchain: ${{ matrix.toolchain }} diff --git a/crates/rutster/tests/barge_in_integration.rs b/crates/rutster/tests/barge_in_integration.rs index 2a44731..651584f 100644 --- a/crates/rutster/tests/barge_in_integration.rs +++ b/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> directly and proves that loud -// caller audio kills playout within one tick, with only the local VAD as the -// trigger source — no brain advisory is involved. This is wedge #1 from -// README:98-100 and ARCHITECTURE.md:79-81. +// The PRIMARY test constructs the reflex stack directly and proves that loud +// caller audio kills playout with only the local VAD as the trigger source — no +// brain advisory is involved. This is wedge #1 from README:98-100 and +// 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>, + 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> { + 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::(64); + let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::(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::(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(); +}