diff --git a/docs/superpowers/plans/2026-07-01-slice-4-barge-in.md b/docs/superpowers/plans/2026-07-01-slice-4-barge-in.md new file mode 100644 index 0000000..a0f1886 --- /dev/null +++ b/docs/superpowers/plans/2026-07-01-slice-4-barge-in.md @@ -0,0 +1,1601 @@ +# 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 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. +- **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+5. `MediaThread` — the binary-side dedicated std::thread; owns RtcSessions + spawns TapEngine. +- **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`. +- **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 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` gains `pub advisory_tx: Option>` (the Reflex holds the paired `advisory_rx`, so this is `None` — the engine task owns the SENDER side via run_tap_client). Wait — actually the engine task CALLS `run_tap_client(... tx_advisory ...)`, which means the engine OWNS the sender; the Reflex holds the receiver. So `TapConn` should carry the `advisory_rx: Option>` so the media thread can wire it into the `Reflex::new(inner, advisory_rx, metrics)` call on `Connected`. + - `spawn_tap_engine` returns a 3-tuple now: `(TapAudioPipe, TapConn, mpsc::Receiver)`. The `advisory_rx` lives separately so the media thread can hand it to `Reflex::new`. + +**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 returns the advisory_rx end so the media + /// thread can wire it into Reflex::new on the Connected transition. + #[tokio::test] + async fn spawn_returns_advisory_rx_end() { + let id = ChannelId::new(); + let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); + let (_pipe, conn, advisory_rx) = + spawn_tap_engine(id, url, crate::session_map::AppState::default()); + // The advisory_rx is a live Receiver (not yet closed by the + // engine task's failure). We push nothing; subsequent recv() + // would yield None only after the engine task drops its sender. + let rx = advisory_rx.expect("spawn_tap_engine must return Some(advisory_rx)"); + let _ = conn.close_tx.send(()); + conn.join.abort(); + drop(rx); // clean drop + } +``` + +- [ ] **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(...) -> (TapAudioPipe, TapConn, Option>)` + +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 (e.g. `session_map.rs::drive_all_sessions` around line 301) destructures the new 3-tuple pattern. + +- [ ] **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 -m "feat(binary): spawn_tap_engine returns advisory_rx for Reflex (slice-4 §3.1) + +The 3rd channel end (alongside tx_pcm_in/rx_audio_out) — the media +thread wires it into Reflex::new on the Connected transition." +``` + +--- + +### 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(); + let (pipe, conn, advisory_rx) = + tokio_handle.block_on(async { spawn_tap_engine(*id, url, crate::session_map::AppState::default()) }); + let advisory_rx = advisory_rx.expect("spawn_tap_engine returns Some(advisory_rx)"); + let metrics = rutster_media::ReflexMetrics::new(); + let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics); + session.rtc.set_pipe(reflex); + session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new()); + session.tap_conn = Some(conn); + info!(channel_id = %id, "tap engine + reflex 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 + +**Files:** +- Create: `crates/rutster/tests/barge_in_integration.rs` +- Consumes: `MockRealtimeBrain` (Task 8), `MediaThread` (Task 6), `spawn_tap_engine` (Task 5), `Reflex` (Task 2). + +- [ ] **Step 1: Write the integration test** + +Extends slice-3's `realtime_integration.rs` pattern. Sets up: `MockRealtimeBrain` (with advisory schedule emitting `speech_started` after 2 audio_in appends + `speech_stopped` after 4) + a `MediaThread` + a synthetic WebRTC peer (or, simpler for the bare-barge assertion: directly drive `Reflex` against a `TapAudioPipe` whose `tx_audio_out` is test-controlled). Asserts: +- `speech_started` → next `Reflex::next_pcm_frame()` returns `None` even with a frame queued (the barge flushed it). +- A fresh `PcmFrame` on `tx_audio_out` → next `next_pcm_frame()` returns `Some` (resume). + +This is the §7 #7-style S4 assertion extended to the reflex. See slice-3's `realtime_integration.rs` for the harness pattern. + +- [ ] **Step 2: Run the test + iterate** — the barge e2e is the proof of the slice. + +- [ ] **Step 3: fmt + clippy + commit** with `test(slice-4):` prefix. + +--- + +### 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: not implemented, as expected. + - §7 Done-criteria: items 1-10 all mapped (Tasks 1-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 → 4 → 5 → 8). `Reflex

` signature consistent (Task 2 → 6). `MediaCmd` consistent (Task 6 → 7). `spawn_tap_engine` returns 3-tuple from Task 5 onward; Task 6's call site uses 3-tuple destructure. +- **Seam gate:** `loop_driver.rs` + `rtc_session.rs` are NOT in any task's Modify list. The invariant is preserved by construction.