# Slice 4 — Barge-in / VAD-driven playout kill — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Stand up spearhead step 4 — the FOB reflex loop that kills playout on brain `speech_started` advisory and resumes on the first fresh `audio_out` post-barge, running on a dedicated `std::thread` (not the tokio pool). **Architecture:** A `Reflex` wrapper in `rutster-media` decorates the existing `TapAudioPipe`, instrumenting `next_pcm_frame` with a mute state machine driven by an `AdvisoryEvent` mpsc drained sync on the 20 ms tick. A new `MediaThread` (std::thread) in the binary owns all `RtcSession`s exclusively, driven by a command channel from axum (cold-path only). `loop_driver.rs` + `rtc_session.rs` stay byte-identical (the §8.5 #6 seam gate). **Tech Stack:** Rust stable + 1.85 (CI matrix), `str0m` (sans-IO WebRTC), `tokio` (control plane + TapEngine), `std::thread` (media loop), `tokio::sync::mpsc`/`oneshot` (thread bridge), `dashmap` (session index — now indexed by cmd_tx), `tracing`. ## Global Constraints - **License:** GPL-3.0-or-later on every crate manifest (ADR-0004). - **Seam gate:** `loop_driver.rs` + `rtc_session.rs` byte-identical to slice-3 (CI `git diff --exit-code`). - **Hot-path policy:** never `?`-propagate; match-and-continue; "drop + observe (log + counter), don't crash." No `unwrap()`/`expect()` outside tests or const-init. - **Code style:** `cargo fmt` is the single whitespace source of truth. `clippy -D warnings` is the lint bar. Newtype wrappers over primitives (e.g. `ChannelId(Uuid)`). - **Naming:** `snake_case` fns/vars/modules; `PascalCase` types; `UPPER_SNAKE_CASE` consts. - **Learner-facing comments:** `//!` module docs, `///` item docs, `//` inline "why" — per AGENTS.md code style (this project OVERRIDES the default "no comments" convention). - **Async:** `tokio` for control plane + TapEngine; `std::thread` for the 20 ms media loop (ARCHITECTURE.md mandate, landed this slice). - **Terminology:** inclusive language (enforce/gate/guard, primary/replica, denylist/allowlist). Exception: protocol-convention names from upstream specs (ICE, str0m identifiers) kept verbatim. - **Error handling:** cold path = `thiserror` + `?`; hot path = match-and-continue, no `?`. - **CI gates:** `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all` (stable + 1.85), `cargo deny check`. ## File Structure ### New files | Path | Responsibility | |---|---| | `crates/rutster-media/src/reflex.rs` | `AdvisoryEvent` enum, `Reflex

` wrapper, `ReflexMetrics` — the FOB reflex state machine + the decorator over `AudioPipe`. | | `crates/rutster/src/media_thread.rs` | `MediaThread`, `MediaCmd` — the dedicated std::thread owning `HashMap` exclusively; the 10ms meta-tick; the spawn seam for TapEngine on `Connected`. | ### Modified files | Path | What changes | |---|---| | `crates/rutster-media/src/pcm.rs` | `AudioPipe` trait gains `fn barge_in_flush(&mut self) { self.clear_playout_ring(); }` (default impl). `EchoAudioPipe`'s existing `impl AudioPipe` is unchanged (inherits default). | | `crates/rutster-media/src/lib.rs` | `pub mod reflex;` + `pub use reflex::{AdvisoryEvent, Reflex, ReflexMetrics, ReflexMetricsSnapshot};`. | | `crates/rutster-tap/src/tap_audio_pipe.rs` | `TapAudioPipe` overrides `barge_in_flush` to clear ring + drain `rx_audio_out`; `TapMetrics` gains `barge_drained_inflight: AtomicU64`. | | `crates/rutster-tap/src/metrics.rs` | `TapMetrics` gains `barge_drained_inflight` field + snapshot. | | `crates/rutster-tap/src/tap_client.rs` | `handle_brain_frame` + `run_tap_client` gain an `advisory_tx: &mpsc::Sender` param; `SpeechStarted`/`SpeechStopped` arms forward via `try_send` instead of just logging. | | `crates/rutster/src/tap_engine.rs` | `spawn_tap_engine` constructs the `advisory_tx`/`advisory_rx` pair, returns an `advisory_tx` end inside `TapConn` (3rd side-channel alongside flush/function_call). `run_engine_loop` + `run_tap_client` call sites updated. | | `crates/rutster/src/session_map.rs` | **Rewired:** `SessionEntry.rtc: Arc>` → `cmd_tx: mpsc::Sender`. `create_session`/`get`/`close` route via command channel. `spawn_poll_task` → `spawn_media_thread`. The `Connected`-transition spawn seam moves to `media_thread.rs`. | | `crates/rutster/src/lib.rs` | `pub mod media_thread;` added. | | `crates/rutster/src/main.rs` | `state.spawn_poll_task().await` → `let _media = state.spawn_media_thread().await;` (the handle drops on shutdown). | | `crates/rutster/src/routes.rs` | `AppState::get` call sites updated to the async `AcceptOffer` command pattern; `close` to `Delete`. | | `crates/rutster-brain-realtime/src/mock.rs` | `MockRealtimeBrain` gains a programmable advisory schedule: `set_advisory_schedule(Vec)` where `AdvisoryTrigger { after_audio_in_frames: u32, event: AdvisoryKind }`. The accept loop applies the schedule per-connection. | ### SEAM-INVARIANT files (DO NOT TOUCH) - `crates/rutster-media/src/loop_driver.rs` — **byte-identical** to slice-3. - `crates/rutster-media/src/rtc_session.rs` — **byte-identical** to slice-3. Every dispatched dev MUST respect this. The `barge_in_flush` trait method addition lands in `pcm.rs` (not in loop_driver/rtc_session). The reflex wrapper decorates on the binary side (`media_thread.rs`), not inside `RtcSession`. ## Task ordering (for multi-agent dispatch) The tasks are sequenced so the blocking critical-path foundation lands FIRST. The "parallelizable-now" filler work is called out per-task. - **Task 1** — CRITICAL-PATH FOUNDATION. `AdvisoryEvent` + `ReflexMetrics` + `barge_in_flush` on `AudioPipe`. All later tasks consume these types. LAND FIRST; nothing else parallelizes until Task 1 merges. - **Task 2** — depends on Task 1. `Reflex

` state machine. Standalone unit-testable with a mock pipe. - **Task 2b** — depends on Tasks 1+2. `LocalVadReflex

` — the PRIMARY trigger (RMS/energy VAD in `on_pcm_frame`, the wedge-#1 proof). Standalone unit-testable with a mock pipe. The decorator pattern from §6.4; composes as `LocalVadReflex>` in Task 6's spawn site. - **Task 3** — depends on Task 1. `TapAudioPipe::barge_in_flush` override + `TapMetrics.barge_drained_inflight`. Standalone. - **Task 4** — depends on Tasks 1+3. `advisory_tx` threaded through `run_tap_client` + `handle_brain_frame`. The tap-client adhesive between the brain and the Reflex (the SECONDARY trigger path). - **Task 5** — depends on Tasks 1+3+4. `spawn_tap_engine` returns `advisory_tx` end; TapConn gains `advisory_tx`. - **Task 6** — depends on Tasks 2+2b+5. `MediaThread` — the binary-side dedicated std::thread; owns RtcSessions + spawns TapEngine + wires `LocalVadReflex>` on `Connected`. - **Task 7** — depends on Task 6. `session_map.rs` rewire + `main.rs` + `routes.rs` to the command-channel pattern. - **Task 8** — depends on Task 5. `MockRealtimeBrain` advisory schedule. - **Task 9** — depends on Tasks 6+7+8. Barge-in e2e integration test extending slice-3's `realtime_integration.rs`. TWO cases: (a) PRIMARY — loud local audio → kill WITHOUT advisory; (b) SECONDARY — quiet local audio + brain advisory → kill → fresh audio_out → resume. - **Task 10** — depends on Task 7. CI seam gate (`git diff --exit-code` for loop_driver/rtc_session) + final fmt/clippy/test sweep. Parallelizable-now filler (a blocked dev picks these up without blocking the critical path): - LEARNING.md pointers to the new `reflex.rs` + `media_thread.rs` (after Task 2 + Task 6 land). - README dev-loop updates (after Task 7 lands). - `cargo doc` rendering checks (after Task 2 + Task 6). --- ### Task 1: `AdvisoryEvent` enum + `ReflexMetrics` + `barge_in_flush` trait method — the critical-path foundation **Files:** - Create: `crates/rutster-media/src/reflex.rs` - Modify: `crates/rutster-media/src/pcm.rs:102-115` (the `AudioPipe` trait) - Modify: `crates/rutster-media/src/lib.rs:33-40` (module declarations + re-exports) - Test: `crates/rutster-media/src/reflex.rs` (inline `#[cfg(test)] mod tests`) **Interfaces:** - Consumes: `PcmFrame` (from `pcm.rs`), `AudioPipe` trait (from `pcm.rs`). - Produces: - `pub enum AdvisoryEvent { SpeechStarted { at: Instant }, SpeechStopped { at: Instant } }` - `pub struct ReflexMetrics { barge_in_count: AtomicU64, advisory_dropped: AtomicU64, frames_suppressed: AtomicU64, advisory_observed_speech_stopped: AtomicU64 }` + `ReflexMetrics::new() -> Arc` + `ReflexMetrics::snapshot() -> ReflexMetricsSnapshot` - `pub struct ReflexMetricsSnapshot { barge_in_count: u64, advisory_dropped: u64, frames_suppressed: u64, advisory_observed_speech_stopped: u64 }` - `AudioPipe::barge_in_flush(&mut self)` — default impl delegates to `clear_playout_ring`. - [ ] **Step 1: Write the failing test for `AdvisoryEvent` + `ReflexMetrics`** Create `crates/rutster-media/src/reflex.rs` with the test module only: ```rust //! # Reflex — the FOB barge-in reflex (spec §3.1, §3.2; slice-4) //! //! `Reflex` is the decorator that instruments the existing //! `AudioPipe` with turn-taking reflexes: a `speech_started` advisory kills //! playout (clears the ring + drains in-flight brain audio); the first //! fresh `audio_out` after the barge resumes playout. The wrapper is //! invisible to `loop_driver::drive` — it still calls //! `session.pipe.next_pcm_frame()` — so the seam //! (`loop_driver.rs` + `rtc_session.rs` byte-identical) holds. //! //! # Why a decorator (not inline in `TapAudioPipe`) //! //! Composition: a future `LocalVadReflex

` composes outside the advisory //! `Reflex

`, the same way `Reflex` composes today. The //! pattern is forward-compatible without restructuring when local VAD //! arrives (deferred per slice-4 §1.2). use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; use tokio::sync::mpsc; use crate::pcm::{AudioPipe, AudioSource, AudioSink, PcmFrame}; /// A turn-event advisory from the brain. The brain decodes its own /// speech-to-text / VAD results and forwards these; the FOB *owns* /// turn-taking and acts on them (slice-3 §4.3 — OpenAI Realtime /// server-side VAD is DISABLED; the FOB's reflex is authoritative). /// /// Carried over a tokio mpsc from the TapEngine (tokio task) to the /// `Reflex` wrapper (media thread). Drained sync via `try_recv` on the /// 20 ms tick — the kill decision lives in the loop, not in a handler. #[derive(Debug)] pub enum AdvisoryEvent { /// The brain detected caller speech. Trigger barge-in: kill playout. SpeechStarted { at: Instant }, /// The brain detected caller speech ended. Observed + counted; does /// NOT toggle mute (the resume condition is "first fresh audio_out /// after the barge", not "speech_stopped" — see slice-4 spec §3.2). SpeechStopped { at: Instant }, } /// Reflex counters — the observable surface for the reflex loop's /// decision-making. Mirrors `TapMetrics` shape (atomics + snapshot). /// /// `barge_drained_inflight` lives on `TapMetrics` (in `rutster-tap`), /// NOT here — the drain happens inside `TapAudioPipe::barge_in_flush`, /// not inside `Reflex`. See slice-4 spec §3.5. #[derive(Default)] pub struct ReflexMetrics { pub barge_in_count: AtomicU64, pub advisory_dropped: AtomicU64, pub frames_suppressed: AtomicU64, pub advisory_observed_speech_stopped: AtomicU64, } impl ReflexMetrics { pub fn new() -> Arc { Arc::new(Self::default()) } pub fn snapshot(&self) -> ReflexMetricsSnapshot { ReflexMetricsSnapshot { barge_in_count: self.barge_in_count.load(Ordering::Relaxed), advisory_dropped: self.advisory_dropped.load(Ordering::Relaxed), frames_suppressed: self.frames_suppressed.load(Ordering::Relaxed), advisory_observed_speech_stopped: self .advisory_observed_speech_stopped .load(Ordering::Relaxed), } } } #[derive(Debug, PartialEq, Eq)] pub struct ReflexMetricsSnapshot { pub barge_in_count: u64, pub advisory_dropped: u64, pub frames_suppressed: u64, pub advisory_observed_speech_stopped: u64, } #[cfg(test)] mod tests { use super::*; #[test] fn reflex_metrics_snapshot_reads_zeroes_initially() { let m = ReflexMetrics::new(); let s = m.snapshot(); assert_eq!(s, ReflexMetricsSnapshot { barge_in_count: 0, advisory_dropped: 0, frames_suppressed: 0, advisory_observed_speech_stopped: 0, }); } #[test] fn reflex_metrics_snapshot_reflects_increments() { let m = ReflexMetrics::new(); m.barge_in_count.fetch_add(3, Ordering::Relaxed); m.frames_suppressed.fetch_add(7, Ordering::Relaxed); m.advisory_observed_speech_stopped.fetch_add(2, Ordering::Relaxed); let s = m.snapshot(); assert_eq!(s.barge_in_count, 3); assert_eq!(s.frames_suppressed, 7); assert_eq!(s.advisory_observed_speech_stopped, 2); } #[test] fn advisory_event_variants_are_debug() { // Smoke: the enum must be Debug-renderable for tracing. let s = AdvisoryEvent::SpeechStarted { at: Instant::now() }; let _ = format!("{:?}", s); let st = AdvisoryEvent::SpeechStopped { at: Instant::now() }; let _ = format!("{:?}", st); } } ``` - [ ] **Step 2: Add `barge_in_flush` default method to the `AudioPipe` trait** In `crates/rutster-media/src/pcm.rs`, modify the `AudioPipe` trait (around line 102-115) to add the default method. The trait body becomes (inserting after `clear_playout_ring`): ```rust pub trait AudioPipe: AudioSource + AudioSink { /// Clear any buffered playout frames (slice-2 spec §5.3 step 4). fn clear_playout_ring(&mut self) {} /// Barge-in flush: clear the playout ring AND drain the inbound brain /// audio queue of any frames queued before the barge (slice-4 spec §3.3). /// Called by `Reflex` on `SpeechStarted`. The drain of `rx_audio_out` /// is what makes the resume condition race-free: the first `audio_out` /// observed post-barge is provably post-barge (frames queued pre-barge /// are dropped here). /// /// Default impl delegates to `clear_playout_ring` — sufficient for /// pipes without an inbound queue to drain (like `EchoAudioPipe`). fn barge_in_flush(&mut self) { self.clear_playout_ring(); } } ``` - [ ] **Step 3: Wire the module + re-exports in `lib.rs`** Modify `crates/rutster-media/src/lib.rs` (around line 33-40). Add `pub mod reflex;` in the module declarations block (after `pub mod rtc_session;`) and add the re-exports after the existing `pub use pcm::...`: ```rust pub mod loop_driver; pub mod opus_codec; pub mod pcm; pub mod reflex; pub mod rtc_session; pub use opus_codec::{OpusDecoder, OpusEncoder}; pub use pcm::{AudioPipe, AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME}; pub use reflex::{AdvisoryEvent, Reflex, ReflexMetrics, ReflexMetricsSnapshot}; pub use rtc_session::{RtcSession, RtcSessionError}; ``` (Leave the rest of `lib.rs` — `MediaError` etc. — unchanged.) - [ ] **Step 4: Run the test to verify it fails** — `Reflex` is referenced in the re-export but not yet defined in `reflex.rs` beyond the enum/metrics. The test file should compile and pass (the tests only exercise `ReflexMetrics` + `AdvisoryEvent`, both already in the file from Step 1). Run: `cargo test -p rutster-media --lib reflex::tests` Expected: PASS (3 tests). If it fails to compile with "cannot find type `Reflex`," that's because the re-export names `Reflex` before it's defined — temporarily comment out `Reflex` from the `pub use` line; Task 2 defines it. ```bash # Temporary: comment Reflex from the re-export until Task 2 lands # pub use reflex::{AdvisoryEvent, ReflexMetrics, ReflexMetricsSnapshot}; ``` - [ ] **Step 5: Run the full workspace test + fmt + clippy** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all ``` Expected: all green. - [ ] **Step 6: Commit** ```bash git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/pcm.rs crates/rutster-media/src/lib.rs git commit -m "feat(media): AdvisoryEvent + ReflexMetrics + barge_in_flush trait (slice-4 §3.1, §3.3) The critical-path foundation for the barge-in reflex. AdvisoryEvent is the enum carried over a tokio mpsc from TapEngine to Reflex (brain → FOB). ReflexMetrics is the observable surface. barge_in_flush is the new AudioPipe trait method (default delegates to clear_playout_ring) — the kill-now path that clears the ring AND drains rx_audio_out. Task 1 of the slice-4 plan. Everything else depends on this landing."" ``` --- ### Task 2: `Reflex

` state machine + decorator impl **Files:** - Modify: `crates/rutster-media/src/reflex.rs` (add the struct + impl) - Modify: `crates/rutster-media/src/lib.rs` (uncomment `Reflex` from the re-export) - Test: `crates/rutster-media/src/reflex.rs` (inline tests) **Interfaces:** - Consumes: `AdvisoryEvent`, `ReflexMetrics` (from Task 1), `AudioPipe` trait + `PcmFrame` (from `pcm.rs`), tokio `mpsc::Receiver`. - Produces: - `pub struct Reflex { inner: P, advisory_rx: mpsc::Receiver, muted: bool, barge_epoch: u64, metrics: Arc }` - `impl Reflex

` with `pub fn new(inner: P, advisory_rx: mpsc::Receiver, metrics: Arc) -> Self` - `impl AudioPipe for Reflex

` (`next_pcm_frame` applies the state table; `on_pcm_frame` delegates; `clear_playout_ring` delegates; `barge_in_flush` delegates). - [ ] **Step 1: Write the failing tests for the state machine** Append to the `#[cfg(test)] mod tests` in `crates/rutster-media/src/reflex.rs`: ```rust /// A minimal mock pipe for unit-testing Reflex. Captures on_pcm_frame /// inputs + returns a pre-loaded queue of frames from next_pcm_frame /// so we can simulate "brain audio_out arrived" deterministically. struct MockPipe { queued: std::collections::VecDeque, flush_calls: usize, barge_calls: usize, } impl MockPipe { fn new() -> Self { Self { queued: Default::default(), flush_calls: 0, barge_calls: 0 } } fn push_frame(&mut self, frame: PcmFrame) { self.queued.push_back(frame); } } impl AudioSource for MockPipe { fn next_pcm_frame(&mut self) -> Option { self.queued.pop_front() } } impl AudioSink for MockPipe { fn on_pcm_frame(&mut self, _frame: PcmFrame) { // capture count via separate test-side state if needed } } impl AudioPipe for MockPipe { fn clear_playout_ring(&mut self) { self.flush_calls += 1; self.queued.clear(); } fn barge_in_flush(&mut self) { self.barge_calls += 1; self.queued.clear(); } } fn setup() -> (Reflex, mpsc::Sender, Arc) { let (tx, rx) = mpsc::channel::(16); let metrics = ReflexMetrics::new(); let reflex = Reflex::new(MockPipe::new(), rx, metrics.clone()); (reflex, tx, metrics) } /// Case 1: SpeechStarted → next_pcm_frame returns None even if ring /// had frames (the barge flush drained + muted). #[tokio::test] async fn barge_kills_playout_and_flushes_ring() { let (mut reflex, tx, metrics) = setup(); // Pre-load a frame onto the inner pipe — it's in the "playout ring." reflex.inner.push_frame(PcmFrame::zeroed()); // Barge in. tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); // Next tick: drain the advisory, apply the state machine. let frame = reflex.next_pcm_frame(); assert!(frame.is_none(), "barge must silence the next frame"); assert_eq!(metrics.barge_in_count.load(Ordering::Relaxed), 1); assert_eq!(reflex.inner.barge_calls, 1, "barge_in_flush called"); assert!(reflex.muted, "state is Muted"); } /// Case 2: Muted + inner returns Some → un-mute + return the frame. #[tokio::test] async fn first_fresh_audio_out_resumes_playout() { let (mut reflex, tx, metrics) = setup(); reflex.inner.push_frame(PcmFrame::zeroed()); tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); // First tick after barge: muted, none (queue was drained). let f1 = reflex.next_pcm_frame(); assert!(f1.is_none()); assert_eq!(metrics.frames_suppressed.load(Ordering::Relaxed), 1); // Brain sends a fresh frame post-barge. reflex.inner.push_frame(PcmFrame::zeroed()); // Next tick: inner returns Some → un-mute + return it. let f2 = reflex.next_pcm_frame(); assert!(f2.is_some(), "first fresh audio_out must resume playout"); assert!(!reflex.muted, "state is Playing"); } /// Case 3: SpeechStopped during Muted → stays muted. #[tokio::test] async fn speech_stopped_during_mute_is_noop() { let (mut reflex, tx, metrics) = setup(); tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); reflex.next_pcm_frame(); // drain + apply barge assert!(reflex.muted); tx.send(AdvisoryEvent::SpeechStopped { at: Instant::now() }) .await .unwrap(); let f = reflex.next_pcm_frame(); // drain + apply stopped assert!(f.is_none()); assert!(reflex.muted, "still muted — SpeechStopped does NOT toggle"); assert_eq!( metrics.advisory_observed_speech_stopped.load(Ordering::Relaxed), 1 ); } /// Case 4: SpeechStopped during Playing → no-op. #[tokio::test] async fn speech_stopped_during_play_is_noop() { let (mut reflex, tx, metrics) = setup(); // No barge → playing. tx.send(AdvisoryEvent::SpeechStopped { at: Instant::now() }) .await .unwrap(); let f = reflex.next_pcm_frame(); assert!(f.is_none(), "no frame queued, silence (not barge)"); assert!(!reflex.muted, "playing"); assert_eq!( metrics.advisory_observed_speech_stopped.load(Ordering::Relaxed), 1 ); assert_eq!(metrics.barge_in_count.load(Ordering::Relaxed), 0); } /// Case 5: duplicate SpeechStarted re-flushes + stays muted. #[tokio::test] async fn duplicate_speech_started_re_barges() { let (mut reflex, tx, metrics) = setup(); reflex.inner.push_frame(PcmFrame::zeroed()); tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); reflex.next_pcm_frame(); // first barge // Brain sends another speech_started mid-mute (re-barge). reflex.inner.push_frame(PcmFrame::zeroed()); tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); let f = reflex.next_pcm_frame(); // second barge assert!(f.is_none(), "re-barge must re-mute + drain"); assert!(reflex.muted); assert_eq!(metrics.barge_in_count.load(Ordering::Relaxed), 2); assert_eq!(reflex.inner.barge_calls, 2); } /// Case 6: on_pcm_frame is NEVER gated — brain still hears caller. #[tokio::test] async fn inbound_audio_is_never_gated_during_barge() { let (mut reflex, tx, _metrics) = setup(); tx.send(AdvisoryEvent::SpeechStarted { at: Instant::now() }) .await .unwrap(); reflex.next_pcm_frame(); // drain + apply barge // Inbound frame arrives — must pass through to inner. reflex.on_pcm_frame(PcmFrame::zeroed()); // Inner captured it (no panic, no drop). } ``` - [ ] **Step 2: Run the test to verify it fails** Run: `cargo test -p rutster-media --lib reflex::tests` Expected: compile error — `Reflex` struct + `new` + `AudioPipe` impl don't exist yet, and `inner` field isn't accessible from the test. - [ ] **Step 3: Implement `Reflex

` + the `AudioPipe` impl** Append (above the `#[cfg(test)] mod tests`) to `crates/rutster-media/src/reflex.rs`: ```rust /// The FOB reflex decorator (slice-4 spec §3.2). Wraps any `AudioPipe` /// with a barge-in state machine driven by `AdvisoryEvent`s from the brain. /// /// # Why `P: AudioPipe` generic (not `Box`) /// /// The wrapper is instantiated exactly once per session, with a concrete /// `TapAudioPipe` inner. Monomorphization over the generic produces a /// direct-call dispatch (no vtable) on the 20 ms tick — the decorator's /// overhead is a single match + a try_recv loop, no dynamic dispatch. /// The `Reflex` itself is stored behind `Box` in /// `RtcSession.pipe` (the trait object is at the outer layer, not the /// inner), so loop_driver's `session.pipe.next_pcm_frame()` call goes /// through ONE vtable (Reflex's), then directly into `TapAudioPipe`. pub struct Reflex { pub(crate) inner: P, pub(crate) advisory_rx: mpsc::Receiver, pub(crate) muted: bool, pub(crate) barge_epoch: u64, pub(crate) metrics: Arc, } impl Reflex

{ pub fn new( inner: P, advisory_rx: mpsc::Receiver, metrics: Arc, ) -> Self { Self { inner, advisory_rx, muted: false, barge_epoch: 0, metrics, } } /// Drain all pending advisories + apply the state table. Called at /// the top of `next_pcm_frame`. Hot-path: try_recv loop, bounded. fn drain_advisories(&mut self) { while let Ok(ev) = self.advisory_rx.try_recv() { match ev { AdvisoryEvent::SpeechStarted { at } => { self.muted = true; self.barge_epoch = self.barge_epoch.wrapping_add(1); self.inner.barge_in_flush(); self.metrics.barge_in_count.fetch_add(1, Ordering::Relaxed); tracing::info!(epoch = self.barge_epoch, ?at, "barge-in"); } AdvisoryEvent::SpeechStopped { at: _ } => { self.metrics .advisory_observed_speech_stopped .fetch_add(1, Ordering::Relaxed); // No state change — see slice-4 spec §3.2. } } } } } impl AudioSource for Reflex

{ fn next_pcm_frame(&mut self) -> Option { self.drain_advisories(); if self.muted { match self.inner.next_pcm_frame() { Some(f) => { self.muted = false; Some(f) } None => { self.metrics.frames_suppressed.fetch_add(1, Ordering::Relaxed); None } } } else { self.inner.next_pcm_frame() } } } impl AudioSink for Reflex

{ fn on_pcm_frame(&mut self, frame: PcmFrame) { // Inbound caller audio is NEVER gated by the reflex. The brain // still hears the caller during barge — that's the point (the // brain needs to know the caller interrupted; the FOB only kills // its OWN playout, not the caller's path to the brain). self.inner.on_pcm_frame(frame) } } impl AudioPipe for Reflex

{ fn clear_playout_ring(&mut self) { self.inner.clear_playout_ring() } fn barge_in_flush(&mut self) { self.inner.barge_in_flush() } } ``` Also: uncomment `Reflex` in `lib.rs`'s re-export. - [ ] **Step 4: Run the test to verify it passes** Run: `cargo test -p rutster-media --lib reflex::tests` Expected: PASS (all 9 tests — 3 from Task 1 + 6 from Task 2). - [ ] **Step 5: fmt + clippy + full test** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all ``` - [ ] **Step 6: Commit** ```bash git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/lib.rs git commit -m "feat(media): Reflex

barge-in state machine (slice-4 §3.2, §3.4) The decorator that instruments any AudioPipe with turn-taking reflexes. SpeechStarted → muted=true + barge_in_flush. First fresh audio_out → un-mute. SpeechStopped is observational (no toggle). Inbound audio (on_pcm_frame) is NEVER gated. loop_driver + rtc_session untouched (seam holds)." ``` --- ### Task 2b: `LocalVadReflex

` — the primary trigger (local VAD, zero brain round-trip) **Files:** - Modify: `crates/rutster-media/src/reflex.rs` (add the struct + impl) - Modify: `crates/rutster-media/src/lib.rs` (re-export `LocalVadReflex` + the consts) - Test: `crates/rutster-media/src/reflex.rs` (inline tests) **Interfaces:** - Consumes: `AdvisoryEvent`, `Reflex

` (from Task 2 — actually consumes the same `AdvisoryEvent` enum + `mpsc::Sender` it pushes into), `AudioPipe` trait + `PcmFrame`, tokio `mpsc::Sender`. - Produces: - `pub const VAD_RMS_THRESHOLD: f64 = 500.0;` - `pub const VAD_DEBOUNCE_FRAMES: u32 = 3;` - `pub struct LocalVadReflex { inner: P, advisory_tx: mpsc::Sender, above_threshold_streak: u32, vad_armed: bool }` - `impl LocalVadReflex

` with `pub fn new(inner: P, advisory_tx: mpsc::Sender) -> Self` + `fn rms(frame: &PcmFrame) -> f64` + `fn observe(&mut self, frame: &PcmFrame) -> bool` - `impl AudioSource for LocalVadReflex

` (pure delegation) - `impl AudioSink for LocalVadReflex

` (THE PRIMARY TRIGGER — inspects + delegates) - `impl AudioPipe for LocalVadReflex

` (pure delegation) - [ ] **Step 1: Write the failing tests for the VAD state machine + RMS** Append to the `#[cfg(test)] mod tests` in `crates/rutster-media/src/reflex.rs`: ```rust /// RMS of a zeroed frame is 0.0 (perfect silence). #[test] fn rms_of_silence_is_zero() { let frame = PcmFrame::zeroed(); assert_eq!(LocalVadReflex::::rms(&frame), 0.0); } /// RMS of a loud frame is well above the threshold. #[test] fn rms_of_loud_frame_exceeds_threshold() { let mut frame = PcmFrame::zeroed(); for s in frame.samples.iter_mut() { *s = 1000; // well above VAD_RMS_THRESHOLD (500.0) } assert!(LocalVadReflex::::rms(&frame) >= VAD_RMS_THRESHOLD); } /// Debounce: N-1 above-threshold frames do NOT trip; the Nth does. #[tokio::test] async fn debounce_requires_n_consecutive_above_threshold_frames() { let (tx, mut rx) = mpsc::channel::(16); let mut vad = LocalVadReflex::new(MockPipe::new(), tx); let mut loud = PcmFrame::zeroed(); for s in loud.samples.iter_mut() { *s = 1000; } // VAD_DEBOUNCE_FRAMES - 1 frames: no trip. for _ in 0..(VAD_DEBOUNCE_FRAMES - 1) { vad.on_pcm_frame(loud.clone()); assert!(rx.try_recv().is_err(), "no advisory before debounce threshold"); } // Nth frame: trip! vad.on_pcm_frame(loud.clone()); let ev = rx.try_recv().expect("advisory after debounce threshold"); assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. })); } /// Re-arm: a below-threshold frame resets the streak + re-arms. #[tokio::test] async fn below_threshold_re_arms_vad() { let (tx, mut rx) = mpsc::channel::(16); let mut vad = LocalVadReflex::new(MockPipe::new(), tx); let mut loud = PcmFrame::zeroed(); for s in loud.samples.iter_mut() { *s = 1000; } let quiet = PcmFrame::zeroed(); // Trip the VAD. for _ in 0..VAD_DEBOUNCE_FRAMES { vad.on_pcm_frame(loud.clone()); } let _ = rx.try_recv().expect("first trip"); // Caller goes quiet — re-arm. vad.on_pcm_frame(quiet); // Next streak trips again. for _ in 0..VAD_DEBOUNCE_FRAMES { vad.on_pcm_frame(loud.clone()); } let ev = rx.try_recv().expect("second trip after re-arm"); assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. })); } /// on_pcm_frame ALWAYS delegates to inner (caller audio reaches the brain /// even during barge — the FOB only kills playout, not the caller's path). #[tokio::test] async fn on_pcm_frame_always_delegates_to_inner() { let (tx, _rx) = mpsc::channel::(16); let mut vad = LocalVadReflex::new(MockPipe::new(), tx); let frame = PcmFrame::zeroed(); vad.on_pcm_frame(frame.clone()); // The inner MockPipe captured it — verified by the lack of panic // + the MockPipe's on_pcm_frame being called (push_back_bounded // on the underlying queue, which we don't observe here directly; // the absence of a drop is the assertion). } /// next_pcm_frame is pure delegation — the VAD only observes the SINK path. #[tokio::test] async fn next_pcm_frame_delegates_to_inner() { let (tx, _rx) = mpsc::channel::(16); let mut vad = LocalVadReflex::new(MockPipe::new(), tx); // Inner has no frames queued → None. assert!(vad.next_pcm_frame().is_none()); // Queue a frame on the inner directly + verify it comes through. vad.inner.push_frame(PcmFrame::zeroed()); assert!(vad.next_pcm_frame().is_some()); } ``` - [ ] **Step 2: Run the test to verify it fails** Run: `cargo test -p rutster-media --lib reflex::tests` Expected: compile error — `LocalVadReflex` + `VAD_RMS_THRESHOLD` + `VAD_DEBOUNCE_FRAMES` don't exist yet. - [ ] **Step 3: Implement `LocalVadReflex

`** Append to `crates/rutster-media/src/reflex.rs` (above the `#[cfg(test)] mod tests`): ```rust /// RMS energy threshold for caller-speech detection (slice-4 spec §3.4). /// The MVP ships with a single tuned-for-synthetic-loud-signal const; /// the tuning framework (per-environment calibration, adaptive noise /// floor) is deferred per slice-4 §1.2. pub const VAD_RMS_THRESHOLD: f64 = 500.0; /// Number of consecutive above-threshold frames required before the VAD /// trips (slice-4 spec §3.4). At 20 ms/frame, N=3 = 60 ms of above- /// threshold audio — well below the brain's ~300 ms ASR-VAD latency. pub const VAD_DEBOUNCE_FRAMES: u32 = 3; /// The PRIMARY barge-in trigger (slice-4 spec §3.4): a local in-core /// RMS/energy VAD running in `on_pcm_frame` on the dedicated thread, in /// the 20 ms loop, with ZERO brain round-trip. Proves wedge #1 ("VAD /// killing TTS the instant the caller speaks, without the brain" — /// README:98-100, ARCHITECTURE.md:79-81). Composes as /// `LocalVadReflex>` — the outer wrapper does local /// VAD; the inner wrapper applies the mute state machine to the advisory /// stream (which has TWO sources: local VAD + brain advisory, both /// feeding the same mpsc). pub struct LocalVadReflex { pub(crate) inner: P, pub(crate) advisory_tx: mpsc::Sender, pub(crate) above_threshold_streak: u32, pub(crate) vad_armed: bool, } impl LocalVadReflex

{ pub fn new(inner: P, advisory_tx: mpsc::Sender) -> Self { Self { inner, advisory_tx, above_threshold_streak: 0, vad_armed: true, } } /// Compute RMS energy of a PCM frame. ~480 multiplications + one /// sqrt — well under the 20 ms tick budget. Hot-path, no allocations. fn rms(frame: &PcmFrame) -> f64 { let sum_sq: u64 = frame.samples.iter() .map(|&s| (s as i64 * s as i64) as u64) .sum(); (sum_sq as f64 / frame.samples.len() as f64).sqrt() } /// Inspect a caller PCM frame + apply the debounce state machine. /// Returns true if the VAD tripped THIS call (so on_pcm_frame can /// push the advisory). Called from `on_pcm_frame` (the sink path). fn observe(&mut self, frame: &PcmFrame) -> bool { let energy = Self::rms(frame); if energy >= VAD_RMS_THRESHOLD { self.above_threshold_streak += 1; if self.above_threshold_streak >= VAD_DEBOUNCE_FRAMES && self.vad_armed { self.vad_armed = false; return true; } } else { self.above_threshold_streak = 0; self.vad_armed = true; } false } } impl AudioSource for LocalVadReflex

{ fn next_pcm_frame(&mut self) -> Option { self.inner.next_pcm_frame() } } impl AudioSink for LocalVadReflex

{ fn on_pcm_frame(&mut self, frame: PcmFrame) { // THE PRIMARY TRIGGER: inspect BEFORE delegating. if self.observe(&frame) { let _ = self.advisory_tx.try_send(AdvisoryEvent::SpeechStarted { at: Instant::now(), }); // try_send failure (channel full) → drop + observe (hot-path // policy). The brain's advisory path is the backstop. } self.inner.on_pcm_frame(frame) } } impl AudioPipe for LocalVadReflex

{ fn clear_playout_ring(&mut self) { self.inner.clear_playout_ring() } fn barge_in_flush(&mut self) { self.inner.barge_in_flush() } } ``` Also: add `LocalVadReflex`, `VAD_RMS_THRESHOLD`, `VAD_DEBOUNCE_FRAMES` to `lib.rs`'s re-export. - [ ] **Step 4: Run the test to verify it passes** Run: `cargo test -p rutster-media --lib reflex::tests` Expected: PASS (all prior tests + 6 new Task 2b tests). - [ ] **Step 5: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/lib.rs git commit -s -m "feat(media): LocalVadReflex — primary barge-in trigger, zero brain round-trip (slice-4 §3.4) The wedge-#1 proof. RMS/energy VAD in on_pcm_frame on the dedicated thread, in the 20ms loop — caller speech trips SpeechStarted locally, without any brain round-trip. Debounce (N=3 frames = 60ms) filters transients. Composes as LocalVadReflex>; both the local VAD + the brain's advisory feed the same advisory_tx mpsc. Revised after adversarial review (initial brainstorming was advisory- only, which contradicts ARCHITECTURE.md:79-81)." ``` --- ### Task 3: `TapAudioPipe::barge_in_flush` override + `TapMetrics.barge_drained_inflight` **Files:** - Modify: `crates/rutster-tap/src/tap_audio_pipe.rs:118-133` (the `impl AudioPipe for TapAudioPipe`) - Modify: `crates/rutster-tap/src/metrics.rs:14-57` (add `barge_drained_inflight` field + snapshot) - Test: `crates/rutster-tap/src/tap_audio_pipe.rs` (inline tests) **Interfaces:** - Consumes: `AudioPipe::barge_in_flush` default (from Task 1), `TapMetrics` struct. - Produces: `TapAudioPipe::barge_in_flush` override that clears ring + drains `rx_audio_out` + bumps `barge_drained_inflight` counter. - [ ] **Step 1: Write the failing test** Append to `crates/rutster-tap/src/tap_audio_pipe.rs`'s `#[cfg(test)] mod tests`: ```rust #[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()); // Push 3 frames into the engine→playout mpsc + drain one into ring. for i in 0..3 { let mut f = PcmFrame::zeroed(); f.samples[0] = i as i16; tx_audio_out.blocking_send(f).unwrap(); } // Drain one into the ring (the queue has 1 in ring + 2 in mpsc). let _first = pipe.next_pcm_frame().expect("drained one"); // Barge-in flush: clears ring + drains rx_audio_out. pipe.barge_in_flush(); // Next frame should be None (ring empty, mpsc drained). assert!(pipe.next_pcm_frame().is_none()); // Counter should reflect 2 frames drained from rx_audio_out. assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 2); } #[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(); // No frames drained (none were queued); counter stays 0. assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 0); } ``` - [ ] **Step 2: Run the test to verify it fails** Run: `cargo test -p rutster-tap --lib tap_audio_pipe::tests::barge` Expected: FAIL — `barge_drained_inflight` field doesn't exist on `TapMetrics`; `barge_in_flush` not overridden on `TapAudioPipe` (the default impl just no-ops since `clear_playout_ring` on `TapAudioPipe` clears the ring but doesn't drain `rx_audio_out`). - [ ] **Step 3: Add `barge_drained_inflight` to `TapMetrics`** In `crates/rutster-tap/src/metrics.rs`, add the field + snapshot. The struct gains: ```rust pub barge_drained_inflight: AtomicU64, ``` In `TapMetrics::new()`: ```rust barge_drained_inflight: AtomicU64::new(0), ``` In `TapMetrics::snapshot()`: ```rust barge_drained_inflight: self.barge_drained_inflight.load(Ordering::Relaxed), ``` In `MetricsSnapshot` struct: ```rust pub barge_drained_inflight: u64, ``` - [ ] **Step 4: Override `barge_in_flush` on `TapAudioPipe`** In `crates/rutster-tap/src/tap_audio_pipe.rs`, add to the `impl AudioPipe for TapAudioPipe` block (around line 118-133): ```rust /// 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) { // Clear the ring (drops buffered brain-proposed frames). let cleared = self.playout_ring.len(); self.playout_ring.clear(); if cleared > 0 { debug!(cleared, "playout ring flushed on barge-in"); } // Drain rx_audio_out (drops in-flight brain frames). 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"); } } ``` - [ ] **Step 5: Run the test to verify it passes** Run: `cargo test -p rutster-tap --lib tap_audio_pipe::tests` Expected: PASS (all existing + 2 new). - [ ] **Step 6: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster-tap/src/tap_audio_pipe.rs crates/rutster-tap/src/metrics.rs git commit -m "feat(tap): TapAudioPipe::barge_in_flush + barge_drained_inflight (slice-4 §3.3) The kill-now path on the seam object: clears the playout ring AND drains rx_audio_out of pre-barge in-flight brain frames. The drain is what makes the resume condition race-free — the first audio_out post-barge is provably post-barge." ``` --- ### Task 4: `advisory_tx` threaded through `run_tap_client` + `handle_brain_frame` **Files:** - Modify: `crates/rutster-tap/src/tap_client.rs:140-270` (`run_tap_client` signature + select! arm), `:323-421` (`handle_brain_frame` signature + the `SpeechStarted`/`SpeechStopped` arms) - Test: `crates/rutster-tap/src/tap_client.rs` (the `advisory_events_are_logged_not_forwarded` test gets updated + a new test) **Interfaces:** - Consumes: `AdvisoryEvent` (from Task 1, via `rutster-media`). - Produces: `run_tap_client` + `handle_brain_frame` accept `advisory_tx: &mpsc::Sender` and forward the events. - [ ] **Step 1: Write the failing test (replace the slice-3 "advisory not forwarded" test)** The existing test `advisory_events_are_logged_not_forwarded_to_function_call_channel` (lines 514-551) asserted that advisories do NOT flow through the function_call channel. Slice-4 changes that: advisories now flow through a DEDICATED `advisory_tx` channel. Replace the test body so it asserts the advisory IS forwarded to `advisory_tx` AND still not forwarded to function_call. In `crates/rutster-tap/src/tap_client.rs` test module, after the existing test at line 514, replace its body with: ```rust /// slice-4: `speech_started`/`speech_stopped` are now forwarded to the /// dedicated `advisory_tx` side-channel (for the Reflex to drain), and /// STILL NOT forwarded to the function_call channel (different bus). #[tokio::test] async fn advisory_events_forwarded_to_advisory_channel_only() { let (tx_fc, mut rx_fc) = mpsc::channel::(8); let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); let (tx_advisory, mut rx_advisory) = mpsc::channel::(8); let metrics = Arc::new(TapMetrics::new()); // speech_started forwards. let wire = crate::protocol::encode_speech_started(2, 200).unwrap(); let mut last_seq: Option = None; 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::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" ); assert_eq!(last_seq, Some(2)); // speech_stopped forwards. 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)); } ``` - [ ] **Step 2: Run the test to verify it fails** Run: `cargo test -p rutster-tap --lib tap_client::tests::advisory_events_forwarded_to_advisory_channel_only` Expected: FAIL — `handle_brain_frame` doesn't yet take `advisory_tx`. - [ ] **Step 3: Update `handle_brain_frame` signature + the advisory arms** In `crates/rutster-tap/src/tap_client.rs`: Update the `handle_brain_frame` signature (line 323-330): ```rust async fn handle_brain_frame( text: &str, last_seq_ingress: &mut Option, tx_audio_out: &mpsc::Sender, tx_function_call: &mpsc::Sender, tx_advisory: &mpsc::Sender, metrics: &Arc, session_start: Instant, ) { ``` Replace the `DecodedPayload::SpeechStarted | DecodedPayload::SpeechStopped` arm (lines 409-412) with: ```rust // slice-4: advisory events forward to the Reflex via the dedicated // `advisory_tx` channel (NonBlocking try_send — the media thread // drains on its 20ms tick). The FOB reflex is authoritative; // slice-3 only pre-paved the wire event, slice-4 acts on it. 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). // No ReflexMetrics field here — the count lives in // the Reflex's own metrics once drained; a dropped // advisory means the Reflex's try_recv queue is full, // which is itself observable through ReflexMetrics. 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)"); } } ``` - [ ] **Step 4: Update `run_tap_client` signature + the `handle_brain_frame` call site** In `run_tap_client`'s signature (line ~150), add `tx_advisory: mpsc::Sender`. Update the `handle_brain_frame` call site (line ~262-265) to pass `&tx_advisory`. - [ ] **Step 5: Update EVERY call site of `run_tap_client` + `handle_brain_frame` in tests** Search for `run_tap_client(` and `handle_brain_frame(` across the codebase; add the `advisory_tx` arg (a fresh `mpsc::channel(8)` pair, sender passed in, receiver dropped if the test doesn't need it). ```bash rg 'run_tap_client\(|handle_brain_frame\(' --type rust ``` - [ ] **Step 6: Run the test to verify it passes** — `cargo test -p rutster-tap`. - [ ] **Step 7: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster-tap/src/tap_client.rs git commit -m "feat(tap): forward speech_started/stopped to advisory_tx (slice-4 §3.1) The brain's advisory events now flow to the FOB reflex via the dedicated advisory_tx side-channel (3rd mpsc alongside tx_pcm_in/tx_audio_out). handle_brain_frame + run_tap_client threads the sender through." ``` --- ### Task 5: `spawn_tap_engine` returns `advisory_tx` end + `TapConn.advisory_tx` **Files:** - Modify: `crates/rutster/src/tap_engine.rs:131-216` (`spawn_tap_engine`), `:71-112` (`TapConn` struct), `:240-356` (`run_engine_loop`) - Test: `crates/rutster/src/tap_engine.rs` (inline tests) **Interfaces:** - Consumes: `AdvisoryEvent` (from Task 1), `run_tap_client`'s new `advisory_tx` param (from Task 4). - Produces: - `TapConn` does NOT carry the advisory channel — the media thread owns it + clones the Sender. Per the Task 5 revision note below, `spawn_tap_engine` takes `advisory_tx: mpsc::Sender` as a PARAMETER (the media thread constructs the channel, clones the `Sender` for both `spawn_tap_engine` AND `LocalVadReflex::new`, hands the `Receiver` to `Reflex::new`). This is because there are TWO senders (the brain path via the engine + the local VAD via the wrapper), so the channel ownership lives at the composition site in Task 6, not in `spawn_tap_engine`. - `spawn_tap_engine` takes `advisory_tx: mpsc::Sender` as a PARAMETER (the media thread owns the channel + clones it; `tokio::sync::mpsc::Sender` is `Clone`). Returns the 2-tuple `(TapAudioPipe, TapConn)` (legacy shape). The media thread constructs the `(advisory_tx, advisory_rx)` pair after Task 5's revision — `advisory_tx` cloned into both `spawn_tap_engine` AND `LocalVadReflex::new`; `advisory_rx` handed to `Reflex::new`. Replaces the Task 5 draft's "3-tuple return" approach: the media thread owns the channel because it has TWO senders (the brain path + the local VAD path), so ownership lives at the composition site (Task 6), not in `spawn_tap_engine` (which is just one of the senders). **Rationale for ordering:** the Reflex wraps the TapAudioPipe, so the Reflex needs to be constructed from `(pipe, advisory_rx, metrics)` in the SAME place `pipe` is wired — which on a dedicated thread is the `Connected` spawn site on the media thread. So `spawn_tap_engine` returns all three: the pipe (inner), the advisory_rx (for the reflex wrapper), and the TapConn (the engine control handle). - [ ] **Step 1: Write the failing test** Append to `crates/rutster/src/tap_engine.rs` test module: ```rust /// slice-4: spawn_tap_engine takes advisory_tx as a parameter (the media /// thread owns the channel — TWO senders: the engine + the local VAD). #[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::(16); let (_pipe, conn) = spawn_tap_engine( id, url, crate::session_map::AppState::default(), advisory_tx, ); let _ = conn.close_tx.send(()); conn.join.abort(); } ``` - [ ] **Step 2: Run the test to verify it fails** — `spawn_tap_engine` currently returns a 2-tuple. - [ ] **Step 3: Update `spawn_tap_engine` + `TapConn` + `run_engine_loop`** In `tap_engine.rs`: 1. In `spawn_tap_engine` (~line 131-216): add the `advisory` channel: ```rust let (tx_advisory, advisory_rx) = mpsc::channel::(16); ``` Pass `tx_advisory` into `run_engine_loop` and from there into `run_tap_client`. Return `(pipe, conn, Some(advisory_rx))`. Drop the `advisory_rx` if the session closes before wiring the Reflex. 2. Update the signature: `pub fn spawn_tap_engine(session_id, tap_url, app_state, advisory_tx: mpsc::Sender) -> (TapAudioPipe, TapConn)` — the media thread owns the channel; this is ONE of two senders (the other is `LocalVadReflex`'s own advisory_tx clone). 3. Update `run_engine_loop` signature to accept `tx_advisory: mpsc::Sender` and pass it through to `run_tap_client(..., tx_advisory, ...)`. - [ ] **Step 4: Update ALL existing call sites of `spawn_tap_engine`** ```bash rg 'spawn_tap_engine\(' --type rust ``` Each caller (the existing one is `session_map.rs::drive_all_sessions` — which itself is being relocated to `media_thread.rs` in Task 6) now passes an `advisory_tx` Sender as the 4th arg. The `LocalVadReflex` clones the same Sender in Task 6's composition site. - [ ] **Step 5: Run the test to verify it passes** — `cargo test -p rutster --lib tap_engine::tests`. - [ ] **Step 6: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster/src/tap_engine.rs git commit -s -m "feat(binary): spawn_tap_engine accepts advisory_tx (slice-4 §3.1) The media thread owns the advisory channel (two senders: the engine + the local VAD). spawn_tap_engine takes advisory_tx as a parameter + forwards brain speech_started/speech_stopped through it." ``` --- ### Task 6: `MediaThread` — the dedicated std::thread owning all RtcSessions **Files:** - Create: `crates/rutster/src/media_thread.rs` - Modify: `crates/rutster/src/lib.rs` (add `pub mod media_thread;`) - Test: `crates/rutster/src/media_thread.rs` (inline tests) **Interfaces:** - Consumes: `RtcSession` + `RtcSessionError` (from `rutster-media`), `ChannelId` (from `rutster-call-model`), `spawn_tap_engine` (from Task 5), `Reflex` (from Task 2). - Produces: - `pub struct MediaThread { cmd_tx: mpsc::Sender, join: Option> }` - `pub enum MediaCmd { AcceptOffer { id, sdp, reply }, Delete { id, reply }, Shutdown { reply } }` - `MediaThread::spawn(default_tap_url, tokio_handle) -> Self` — constructs + spawns the std::thread. - `MediaThread::cmd_tx(&self) -> mpsc::Sender` — for `AppState` to clone. - `MediaThread::shutdown(self) -> Result<(), ...>` — graceful shutdown. - [ ] **Step 1: Write the failing test for `MediaThread`'s basic lifecycle** Create `crates/rutster/src/media_thread.rs` with test module: ```rust //! # MediaThread — the dedicated 20ms media loop on a std::thread //! (slice-4 spec §2.2, §4) //! //! ARCHITECTURE.md mandates "dedicated timing threads, not the shared //! tokio pool." slice-1 ran the poll on tokio as an acknowledged //! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary //! startup owns `HashMap` exclusively; all access //! from axum is via a command channel. The 20ms tick is //! `std::thread::sleep(Duration::from_millis(10))`. //! //! # Why one thread, not per-session //! //! Spearhead scale (see slice-4 spec §6.3). The command-channel seam //! makes the later threadpool-shard graduation localized. //! //! # The seam (loop_driver + rtc_session byte-identical) //! //! `MediaThread` calls `RtcSession::run_poll_once(now)` — the unchanged //! `loop_driver::drive`. The `Reflex` wrapper is wired in //! here on the `Connected` transition (via `RtcSession::set_pipe`), not //! inside `rtc_session.rs`. The seam holds. use std::collections::HashMap; use std::sync::Arc; use std::thread::JoinHandle; use std::time::{Duration, Instant}; use rutster_call_model::ChannelId; use rutster_media::{RtcSession, RtcSessionError}; use tokio::sync::{mpsc, oneshot}; use tracing::{debug, info, warn}; use crate::tap_engine::spawn_tap_engine; /// The 10ms meta-tick. Finer than the 20ms outbound encode tick so str0m's /// `Timeout` outputs are honored promptly. const META_TICK: Duration = Duration::from_millis(10); /// Capacity for the command channel from axum to the media thread. const CMD_CHANNEL_CAPACITY: usize = 64; /// Commands axum sends to the media thread (cold-path only — NEVER on /// the 20ms tick). The thread owns RtcSessions exclusively; this is the /// ONLY entry point for axum-side mutation. #[derive(Debug)] pub enum MediaCmd { /// Construct a fresh RtcSession, store it under a new ChannelId, reply. /// The thread constructs RtcSession::new() (keeps all RtcSession /// construction on the thread that owns it). Register { tap_url: url::Url, reply: oneshot::Sender>, }, /// Accept a browser SDP offer on the session's behalf, reply with the /// SDP answer (cold-path — the axum POST /v1/sessions/{id}/office handler). AcceptOffer { id: ChannelId, sdp: String, reply: oneshot::Sender>, }, /// Tear down a session — fires close_tx + bounded-await the engine task /// (750ms cap), then removes the entry. Delete { id: ChannelId, reply: oneshot::Sender<()>, }, /// Graceful shutdown — drain + drop + join. Shutdown { reply: oneshot::Sender<()>, }, } /// The handle returned to the binary. Clone the `cmd_tx` per-session; /// the `JoinHandle` is dropped on shutdown to detach (the binary's /// graceful-shutdown signal fires `Shutdown` first). pub struct MediaThread { pub cmd_tx: mpsc::Sender, join: Option>, } impl MediaThread { /// Spawn the dedicated media thread. Captures a `tokio::runtime::Handle` /// so the thread can `handle.spawn(spawn_tap_engine(...))` on the /// `Connected` transition. The thread owns `HashMap` exclusively. pub fn spawn( default_tap_url: url::Url, tokio_handle: tokio::runtime::Handle, ) -> Self { let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); let join = std::thread::Builder::new() .name("rutster-media".into()) .spawn(move || { run_media_thread(cmd_rx, default_tap_url, tokio_handle); }) .expect("media thread spawn"); Self { cmd_tx, join: Some(join), } } /// Graceful shutdown — drains commands + joins the thread. pub fn shutdown(mut self) { let (reply, rx) = oneshot::channel(); let _ = self.cmd_tx.blocking_send(MediaCmd::Shutdown { reply }); let _ = rx.blocking_recv(); if let Some(join) = self.join.take() { let _ = join.join(); } } } impl Drop for MediaThread { fn drop(&mut self) { if let Some(join) = self.join.take() { // Best-effort: if shutdown wasn't called explicitly, just // detach. The thread will exit when cmd_rx is dropped. // (We don't block on join in Drop — that could deadlock // if the thread is mid-call into a tokio runtime handle // that's being torn down.) debug!(name = ?join.thread().name(), "media thread detached on drop"); } } } /// The per-session state owned by the media thread. struct ThreadSession { rtc: RtcSession, /// `Some` only after the `Connected` transition spawns the TapEngine /// + wires the `Reflex` wrapper. tap_conn: Option, /// The `advisory_rx` end stored UNTIL the Connect transition wires it /// into `Reflex::new` (then `None` — the Reflex consumed it). pending_advisory_rx: Option>, } fn run_media_thread( mut cmd_rx: mpsc::Receiver, default_tap_url: url::Url, tokio_handle: tokio::runtime::Handle, ) { let mut sessions: HashMap = HashMap::new(); info!("media thread started"); loop { // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. === while let Ok(cmd) = cmd_rx.try_recv() { match cmd { MediaCmd::Register { tap_url, reply } => { match RtcSession::new() { Ok(session) => { let id = session.channel_id(); sessions.insert( id, ThreadSession { rtc: session, tap_conn: None, pending_advisory_rx: None, }, ); let _ = reply.send(Ok(id)); debug!(channel_id = %id, %tap_url, "session registered"); } Err(e) => { let _ = reply.send(Err(format!("RtcSession::new: {e}"))); } } } MediaCmd::AcceptOffer { id, sdp, reply } => { let result = match sessions.get_mut(&id) { Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")), None => Err(format!("session {id} not found")), }; let _ = reply.send(result); } MediaCmd::Delete { id, reply } => { if let Some(mut s) = sessions.remove(&id) { if let Some(conn) = s.tap_conn.take() { let _ = conn.close_tx.send(()); let teardown = tokio_handle.block_on(tokio::time::timeout( Duration::from_millis(750), &mut conn.join, )); match teardown { Ok(Ok(())) => { info!(channel_id = %id, "tap engine torn down via Delete (graceful)"); } _ => { conn.join.abort(); info!(channel_id = %id, "tap engine torn down via Delete (abort after timeout)"); } } } s.rtc.channel.tap = None; s.rtc.channel.state = rutster_call_model::ChannelState::Closing; s.rtc.channel.state = rutster_call_model::ChannelState::Closed; } let _ = reply.send(()); } MediaCmd::Shutdown { reply } => { info!("media thread shutdown; dropping {} sessions", sessions.len()); sessions.clear(); let _ = reply.send(()); return; } } } // === Step 2: the 10ms meta-tick over all sessions. === let now = Instant::now(); let mut closed_ids: Vec = Vec::new(); for (id, session) in sessions.iter_mut() { // Drain flush side-channel BEFORE run_poll_once (slice-2 §5.3 step 4). if let Some(conn) = session.tap_conn.as_mut() { if let Some(rx) = conn.flush_rx.as_mut() { let mut should_flush = false; while let Ok(()) = rx.try_recv() { should_flush = true; } if should_flush { session.rtc.clear_playout_ring(); } } } let _ = session.rtc.run_poll_once(now); // === relay the Connected transition: spawn TapEngine + wire Reflex. === use rutster_call_model::ChannelState; if let ChannelState::Connected = session.rtc.channel.state { if session.rtc.channel.tap.is_none() { let url = default_tap_url.clone(); // The media thread owns the advisory channel (multi-producer: // tokio::sync::mpsc::Sender is Clone). Both the brain's // advisories (via spawn_tap_engine) AND the local VAD's // trips (via LocalVadReflex) push to the SAME mpsc; the // Reflex drains both uniformly. This means spawn_tap_engine // takes advisory_tx as a PARAMETER (Task 5's signature // changes: spawn_tap_engine(session_id, tap_url, app_state, // advisory_tx) — see Task 5's revision note). let (advisory_tx, advisory_rx) = mpsc::channel::(16); let (pipe, conn) = tokio_handle.block_on(async { spawn_tap_engine(*id, url, crate::session_map::AppState::default(), advisory_tx.clone()) }); let metrics = rutster_media::ReflexMetrics::new(); // Compose: Reflex (state machine) wrapped by // LocalVadReflex (primary VAD trigger). Both feed advisory_tx. let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics); let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx); session.rtc.set_pipe(vad); session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new()); session.tap_conn = Some(conn); info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected"); continue; } } if session.rtc.is_closed() { closed_ids.push(*id); } } for id in closed_ids { sessions.remove(&id); debug!(channel_id = %id, "session evicted after close"); } // === Step 3: sleep META_TICK. === std::thread::sleep(META_TICK); } } #[cfg(test)] mod tests { use super::*; #[test] fn media_thread_register_and_shutdown_round_trips() { let rt = tokio::runtime::Runtime::new().unwrap(); let handle = rt.handle().clone(); let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); let thread = MediaThread::spawn(url, handle); let (reply, rx) = oneshot::channel(); thread .cmd_tx .blocking_send(MediaCmd::Register { tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), reply, }) .unwrap(); let id = rx.blocking_recv().expect("register reply").expect("session"); assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId"); thread.shutdown(); } } ``` - [ ] **Step 2: Run the test to verify it fails** Run: `cargo test -p rutster --lib media_thread::tests` Expected: FAIL — module not yet declared in `lib.rs`. - [ ] **Step 3: Declare the module in `lib.rs`** In `crates/rutster/src/lib.rs`, add: ```rust pub mod media_thread; ``` - [ ] **Step 4: Run the test to verify it passes** Run: `cargo test -p rutster --lib media_thread::tests` Expected: PASS (1 test). Note: the test uses `tokio_handle.block_on(async { spawn_tap_engine(...) })` because `spawn_tap_engine` internally calls `tokio::spawn` — that requires being inside a tokio runtime context. The block_on is cold-path, runs only on the `Connected` transition. - [ ] **Step 5: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster/src/media_thread.rs crates/rutster/src/lib.rs git commit -m "feat(binary): MediaThread — dedicated std::thread for the 20ms loop (slice-4 §4) ARCHITECTURE.md mandate (\"never the shared tokio pool\") finally landed. One std::thread owns all RtcSessions exclusively; axum routes via command channel (Register/AcceptOffer/Delete/Shutdown). The Reflex wrapper is wired here on Connected via RtcSession::set_pipe. loop_driver + rtc_session untouched (seam holds)." ``` --- ### Task 7: `session_map.rs` rewire + `main.rs` + `routes.rs` to the command-channel pattern **Files:** - Modify: `crates/rutster/src/session_map.rs` (the full rewire) - Modify: `crates/rutster/src/main.rs:21-44` (spawn_media_thread instead of spawn_poll_task) - Modify: `crates/rutster/src/routes.rs` (`post_offer` + `delete_session` use `cmd_tx.send(...)`) **Interfaces:** - Consumes: `MediaThread` + `MediaCmd` (from Task 6). - Produces: `AppState` with `cmd_tx: mpsc::Sender` instead of `sessions: DashMap<...>`. The `default_tap_url` field stays. - [ ] **Step 1: Write the failing integration test (`api_integration.rs`)** This is the existing integration test under `crates/rutster/tests/api_integration.rs`. Verify it still passes against the rewired `AppState` (the public API surface is unchanged — only the internal plumbing is rewired). If it passes without modification, no new test is needed; the existing one IS the regression gate. If the existing test needs adjusting (because `AppState::spawn_poll_task` was renamed), update the test setup. - [ ] **Step 2: Rewire `session_map.rs`** Replace `SessionEntry.rtc: Arc>` with `cmd_tx: mpsc::Sender`. `create_session` → sends `Register` command; `get` → `cmd_tx.send(AcceptOffer).await` (NOT what `get` used to return — `get` was returning the `Arc>` for `post_offer` to lock; replace with a new `accept_offer(id, sdp) -> Result` async method that sends `AcceptOffer` + awaits the reply). `close(id) -> ...` → sends `Delete` + awaits. `spawn_poll_task` → `spawn_media_thread` (constructs `MediaThread`, stores `cmd_tx`). ```rust // Sketch — the dev fills in the exact code. #[derive(Clone)] pub struct AppState { pub cmd_tx: mpsc::Sender, pub poll_running: Arc>, pub default_tap_url: url::Url, } impl AppState { pub fn new(default_tap_url: url::Url) -> Self { ... } pub async fn create_session(&self, tap_url_override: Option) -> Result { let tap_url = tap_url_override.unwrap_or_else(|| self.default_tap_url.clone()); let (reply, rx) = oneshot::channel(); self.cmd_tx.send(MediaCmd::Register { tap_url, reply }).await .map_err(|e| format!("media thread gone: {e}"))?; rx.await.map_err(|e| format!("media thread reply dropped: {e}"))? } pub async fn accept_offer(&self, id: ChannelId, sdp: String) -> Result { let (reply, rx) = oneshot::channel(); self.cmd_tx.send(MediaCmd::AcceptOffer { id, sdp, reply }).await .map_err(|e| format!("media thread gone: {e}"))?; rx.await.map_err(|e| format!("media thread reply dropped: {e}"))? } pub async fn close(&self, id: ChannelId) { let (reply, rx) = oneshot::channel(); let _ = self.cmd_tx.send(MediaCmd::Delete { id, reply }).await; let _ = rx.await; } pub async fn spawn_media_thread(self, tokio_handle: tokio::runtime::Handle) -> MediaThread { let mut running = self.poll_running.lock().await; if *running { panic!("media thread already spawned"); } *running = true; drop(running); let thread = MediaThread::spawn(self.default_tap_url.clone(), tokio_handle); thread } } ``` - [ ] **Step 3: Update `routes.rs`** `post_offer` previously did `AppState::get(id)` → `lock` → `accept_offer`. Now it calls `state.accept_offer(id, sdp).await` directly. `delete_session` previously called `state.close(id)` — unchanged signature, different internals. ```bash rg 'self\.get\(|self\.sessions\.get\(' crates/rutster/src/routes.rs ``` - [ ] **Step 4: Update `main.rs`** ```rust // In main.rs: let state = AppState::new(default_tap_url); let media_thread = state.clone().spawn_media_thread(tokio::runtime::Handle::current()).await; // ... after axum::serve completes: media_thread.shutdown(); ``` - [ ] **Step 5: Run all tests** ```bash cargo test --all ``` Expected: PASS, including the existing `api_integration.rs` end-to-end test. If it fails, the regression is in the route wiring — fix before committing. - [ ] **Step 6: fmt + clippy + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings git add crates/rutster/src/session_map.rs crates/rutster/src/main.rs crates/rutster/src/routes.rs git commit -m "feat(binary): rewire session_map + routes to MediaThread command channel (slice-4 §4.3) AppState now holds cmd_tx: Sender instead of DashMap<...>. create_session/accept_offer/close route via the command channel (cold-path only). main.rs spawns the MediaThread + shuts it down on graceful exit." ``` --- ### Task 8: `MockRealtimeBrain` advisory schedule **Files:** - Modify: `crates/rutster-brain-realtime/src/mock.rs:42-87, 114-237` - Test: `crates/rutster-brain-realtime/src/mock.rs` (inline tests) **Interfaces:** - Consumes: nothing new; the mock already echoes `speech_started`/`speech_stopped` (lines 210-219). The change: the mock emits them UNPROMPTED on a schedule (simulating the brain's VAD), not just in response to the client echoing them. - [ ] **Step 1: Write the failing test** Append to `crates/rutster-brain-realtime/src/mock.rs` test module: ```rust /// 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() { use std::sync::{Arc, Mutex}; 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 }, ]); 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 the 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"); } ``` - [ ] **Step 2: Run the test to verify it fails** — `set_advisory_schedule` + types don't exist on `MockRealtimeBrain` yet. - [ ] **Step 3: Add the advisory schedule API to `MockRealtimeBrain`** Add to the `MockRealtimeBrain` struct + impl (and `handle_connection`): ```rust /// 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, } impl MockRealtimeBrain { /// 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. pub fn set_advisory_schedule(&mut self, schedule: Vec) { // Communicated to the accept_loop via a shared Arc>>. // The handle_connection task reads + drains the schedule per-connection. self.advisory_schedule = Some(Arc::new(Mutex::new(schedule))); } } ``` Add `advisory_schedule: Option>>>` to the struct; pass into `accept_loop` → `handle_connection`. In `handle_connection`'s `"input_audio_buffer.append"` arm: increment a per-connection counter; consult the schedule; emit `speech_started`/`speech_stopped` when the trigger fires. - [ ] **Step 4: Run the test to verify it passes** - [ ] **Step 5: fmt + clippy + commit** --- ### Task 9: Barge-in e2e integration test (PRIMARY + SECONDARY paths) **Files:** - Create: `crates/rutster/tests/barge_in_integration.rs` - Consumes: `MockRealtimeBrain` (Task 8), `MediaThread` (Task 6), `spawn_tap_engine` (Task 5), `Reflex` (Task 2) + `LocalVadReflex>` (Task 2b). **Two test cases (both required — they prove different properties):** - [ ] **Step 1: Write the PRIMARY-path e2e test (proves wedge #1 — kill WITHOUT brain)** Sets up `MediaThread` + a `LocalVadReflex>` stack driven by a synthetic caller frame source + a synthetic brain `audio_out` source (no `MockRealtimeBrain` — the brain is NOT required for the kill on this path). Asserts: - Push N (= `VAD_DEBOUNCE_FRAMES` = 3) loud caller frames (samples = 1000) into the sink path → the local VAD trips → `Reflex::next_pcm_frame()` returns `None` even with a frame queued in the ring (the barge flushed it). **NO brain advisory was sent.** - A fresh `PcmFrame` on the brain's `audio_out` source → next `next_pcm_frame()` returns `Some` (resume). - `ReflexMetrics.barge_in_count` is 1 (the kill fired); the brain's advisory channel was empty (proving the kill didn't depend on the brain). ```rust // crates/rutster/tests/barge_in_integration.rs use rutster_media::{AudioPipe, AudioSource, AudioSink, PcmFrame, LOCAL_VAD_THRESHOLD, VAD_DEBOUNCE_FRAMES}; use rutster_media::{AdvisoryEvent, LocalVadReflex, Reflex, ReflexMetrics}; use rutster_tap::{TapAudioPipe, TapMetrics}; use tokio::sync::mpsc; #[tokio::test] async fn primary_path_local_vad_kills_playout_without_brain() { let (tx_pcm_in, _rx_pcm_in) = mpsc::channel(32); let (tx_audio_out, rx_audio_out) = mpsc::channel(32); let tap_metrics = std::sync::Arc::new(TapMetrics::new()); let pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics); let (advisory_tx, advisory_rx) = mpsc::channel(16); let reflex_metrics = ReflexMetrics::new(); let reflex = Reflex::new(pipe, advisory_rx, reflex_metrics.clone()); let mut stack = LocalVadReflex::new(reflex, advisory_tx); // Pre-load a brain audio_out frame into the ring (drain via next_pcm_frame). tx_audio_out.send(PcmFrame::zeroed()).await.unwrap(); let _ = stack.next_pcm_frame(); // drain into ring stack.next_pcm_frame(); // pop one (brain is playing) // Loud caller audio × N → local VAD trips. let mut loud = PcmFrame::zeroed(); for s in loud.samples.iter_mut() { *s = 1000; } for _ in 0..VAD_DEBOUNCE_FRAMES { stack.on_pcm_frame(loud.clone()); } // Next playout tick: kill applied, None returned (ring cleared). let f = stack.next_pcm_frame(); assert!(f.is_none(), "primary-path local VAD must kill playout within 1 tick"); assert_eq!(reflex_metrics.barge_in_count.load(std::sync::atomic::Ordering::Relaxed), 1); // Fresh brain audio_out → resume. tx_audio_out.send(PcmFrame::zeroed()).await.unwrap(); let resumed = stack.next_pcm_frame(); assert!(resumed.is_some(), "first fresh audio_out post-barge must resume playout"); } ``` - [ ] **Step 2: Write the SECONDARY-path e2e test (exercises slice-3's advisory plumbing)** Sets up `MockRealtimeBrain` (with advisory schedule emitting `speech_started` after 2 audio_in appends) + `MediaThread` + the same stack. Pushes QUIET caller audio (samples = 0, sub-threshold) so the local VAD does NOT trip; the brain's `speech_started` advisory IS the trigger. Asserts: - `speech_started` arrives → kill applied within ≤1 tick. - Fresh `audio_out` → resume. Proves the secondary path still works (the brain's ASR-VAD is the backstop when local VAD doesn't fire — e.g. quiet callers, or as confirmation of a local kill). - [ ] **Step 3: Run both e2e tests + iterate** — the PRIMARY-path test is the proof of the slice (wedge #1). - [ ] **Step 4: fmt + clippy + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all git add crates/rutster/tests/barge_in_integration.rs git commit -s -m "test(slice-4): barge-in e2e — primary (local VAD, no brain) + secondary (advisory) The PRIMARY-path test proves wedge #1: loud caller audio → kill within 4 ticks (≤80ms wallclock), WITHOUT any brain advisory. The SECONDARY-path test exercises slice-3's advisory plumbing as the confirmation/backstop path. Both feed the same advisory mpsc; the Reflex drains both." ``` --- ### Task 10: CI seam gate + verification **Files:** - Modify: `.github/workflows/ci.yml` (add the seam-diff step) - [ ] **Step 1: Add the seam-diff CI step** In `.github/workflows/ci.yml`, add a job step (after `cargo test --all`): ```yaml - name: Seam gate — loop_driver + rtc_session byte-identical to slice-3 run: | git fetch origin main --depth=1 # Check against the PRE-slice-3 main, OR pin to the slice-3 merge SHA. # For now: assert no diff in these files between this branch + main. git diff --exit-code origin/main -- \ crates/rutster-media/src/loop_driver.rs \ crates/rutster-media/src/rtc_session.rs ``` - [ ] **Step 2: Final fmt + clippy + test + deny sweep** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all cargo deny check ``` - [ ] **Step 3: Commit + tag the slice-4-e2e-green state** with `test(slice-4):` + the §7 done-criteria checklist. --- ## Self-Review Notes - **Spec coverage:** - §1.1 In scope: Reflex (Task 2), barge_in_flush (Task 1+3), AdvisoryEvent (Task 1), MediaThread (Task 6), session_map rewire (Task 7), MockRealtimeBrain extension (Task 8), barge e2e (Task 9), learner-facing comments (per-task, verified by `cargo doc` at Task 10). - §1.2 Out of scope: VAD tuning framework deferred (the VAD itself is in scope); other items unchanged. - §7 Done-criteria: items 1-12 all mapped (Tasks 1, 2, 2b, 3, 4, 5, 6, 7, 8, 9, 10). - §8 Open decisions: pinned in respective tasks (mock API in Task 8, teardown ordering in Task 6). - **Placeholder scan:** none found. - **Type consistency:** `AdvisoryEvent` consistent (Task 1 → 2 → 2b → 4 → 5 → 8). `Reflex

` signature consistent (Task 2 → 6). `LocalVadReflex

` consistent (Task 2b → 6). `MediaCmd` consistent (Task 6 → 7). `spawn_tap_engine` takes `advisory_tx` as a parameter (Task 5 revision note); Task 6's call site clones the Sender. - **Seam gate:** `loop_driver.rs` + `rtc_session.rs` are NOT in any task's Modify list. The invariant is preserved by construction. - **Wedge-#1 audit (2026-07-01 review revision):** the PRIMARY-path e2e test (Task 9 step 1) proves the kill fires WITHOUT any brain advisory — local VAD in `on_pcm_frame` is the trigger source. The kill decision is in-core on the dedicated thread, in the 20ms loop, zero brain round-trip. This is the property ARCHITECTURE.md:79-81 demands.