# Rutster slice 4 — Barge-in: VAD-driven playout kill on a dedicated media thread - **Status:** Draft (pending review) - **Date:** 2026-07-01 - **Spearhead step:** 4 of 6 (vision-revision §10 / PORT_PLAN "Phasing") - **Origin:** brainstorming session 2026-07-01 - **Depends on:** [slice 1 — WebRTC media loopback](2026-06-28-slice-1-webrtc-loopback-design.md), [slice 2 — The agent tap](2026-06-28-slice-2-agent-tap-design.md), and slice-3's OpenAI Realtime brain (merged as `c30a452` — `MockRealtimeBrain` + translator + the `speech_started` / `speech_stopped` advisory events). All three must be landed and green. - **Related:** [ADR-0002](../../adr/0002-north-star-and-fused-core.md) (fused vertical — the hot-path hop invariant this slice re-affirms), [ADR-0008](../../adr/0008-fob-and-green-zone.md) (FOB/green-zone doctrine — the reflex is a FOB member: hot-path, differentiating), [ARCHITECTURE.md §"Biggest technical risk"](../../ARCHITECTURE.md) (the reflex loop *is* the remaining long pole), [ARCHITECTURE.md §"Media plane"](../../ARCHITECTURE.md) ("Dedicated timing threads for the 20ms loop, **never the shared tokio pool**" — this slice finally lands that mandate). --- ## TL;DR Stand up spearhead step 4: the **FOB reflex loop**. Slice 3 pre-paved the advisory signals (`speech_started` / `speech_stopped` from the brain) and locked the turn-ownership decision (OpenAI Realtime server-side VAD disabled; the FOB owns turn-taking). Slice 4 **acts** on the caller's speech with a **local in-core VAD** as the primary trigger — an RMS/energy detector running in `on_pcm_frame` on the dedicated thread, in the 20 ms loop, with **zero brain round-trip** between caller speech and playout kill. This is the property ARCHITECTURE.md:79-81 demands ("Local real-time reflexes... live in-core because the brain round-trip is too slow to enforce them") and the proof wedge #1 rests on ("VAD killing TTS the instant the caller speaks, without the brain" — README:98-100). Slice-3's `speech_started`/`speech_stopped` advisory becomes the **secondary/confirmation** signal — the brain's ASR-quality VAD confirms the local kill slightly later, but the kill itself fires from the FOB's own inspection of caller audio. Slice 4 also **graduates the media loop off the tokio pool**: a single dedicated `std::thread` owns all `RtcSession`s exclusively and drives the 20 ms tick via `Instant::sleep_until`. This honors ARCHITECTURE.md's "never the shared tokio pool" mandate, which slice-1 explicitly deferred to "step 4 (barge-in)" (`loop_driver.rs:18-23`). The graduation is load-bearing: the reflex is the differentiator and the long pole, and its timing discipline demands a thread that doesn't compete with the axum runtime for scheduling. The **seam slice 1→3 preserved** (`loop_driver.rs` + `rtc_session.rs` byte-identical) holds for slice-4 as well: the reflex is a pair of composing `AudioPipe` decorators (`Reflex

` + `LocalVadReflex

`) in `rutster-media`, invisibly to `loop_driver::drive`. Only the binary-side wiring (`session_map.rs` → `media_thread.rs`) changes shape; the media crate's hot path stays untouched. --- ## 1. Scope ### 1.1 In scope - Implementation of spearhead step 4: **barge-in / VAD-driven playout kill**, driven by a **local in-core VAD** (RMS/energy detector in `on_pcm_frame` — the primary trigger, zero brain round-trip) with the brain's `speech_started`/`speech_stopped` advisory as the secondary/confirmation signal. This proves wedge #1 ("VAD killing TTS the instant the caller speaks, without the brain" — README:98-100, ARCHITECTURE.md:79-81). - A **new `Reflex` wrapper** (`rutster-media/src/reflex.rs`) that decorates the pipe the `RtcSession` holds. The reflex owns the mute state machine, the advisory channel receiver, and the barge-in flush trigger. It is the concrete embodiment of ARCHITECTURE.md's "local real-time reflexes" row for the barge-in case. Fed by BOTH the local VAD (via the outer `LocalVadReflex` wrapper that injects `AdvisoryEvent`s into the same mpsc) AND the brain's advisories. - A **new `LocalVadReflex` wrapper** (`rutster-media/src/reflex.rs`) — the PRIMARY trigger. Decorates `AudioPipe`'s `on_pcm_frame` path: inspects the caller's decoded PCM samples (computes RMS, compares to a `const` threshold, debounces N consecutive above-threshold frames), fires `AdvisoryEvent::SpeechStarted { at: Instant::now() }` into the `Reflex`'s advisory mpsc when local VAD trips. Wraps `Reflex` (composition: `LocalVadReflex>` — the decorator pattern from §6.4 pays off here). Forward-compatible: the threshold is a `const` for the MVP (no tuning framework — that's the only piece deferred, per §1.2). - A **new `barge_in_flush` method on the `AudioPipe` trait** (default impl delegates to `clear_playout_ring`) — the seam object's "kill now" path: clear the playout ring AND drain the brain-bound `rx_audio_out` channel of any frames queued before the barge so the first `audio_out` observed post-barge is provably post-barge. `TapAudioPipe` overrides; `EchoAudioPipe` uses the default. - A **new `AdvisoryEvent` enum** (`SpeechStarted { at }`, `SpeechStopped { at }`) flowing over a tokio mpsc from the TapEngine (tokio) to the Reflex (media thread). The engine pushes the events it already decodes from the brain (slice-3 wired these as log+count; slice-4 *forwards* them into the reflex). - A **new dedicated media thread** (`rutster/src/media_thread.rs`) replacing the tokio `spawn_poll_task`. One `std::thread::spawn` at binary startup owns `HashMap` exclusively; all access from axum is via a command channel (`AcceptOffer`, `Delete`, `Shutdown`). The 20 ms tick is `std::thread::sleep`. - **Rewired `session_map.rs`** (binary): `SessionEntry.rtc: Arc>` → `cmd_tx: mpsc::Sender`. `create_session`, `post_offer`, `close`, `spawn_poll_task` all route through the command channel. The async handlers are cold-path; no cross-thread coordination happens on the 20 ms tick. - **`MockRealtimeBrain` extension** (`rutster-brain-realtime/src/mock.rs`): gains the ability to emit `speech_started` / `speech_stopped` on a programmable schedule (e.g. "after N audio_in frames received, send `speech_started`; after M more, send `speech_stopped`"). - **Barge-in e2e integration test** (extends slice-3's `crates/rutster/tests/realtime_integration.rs` harness): synthetic WebRTC peer → MediaThread → TapEngine → MockRealtimeBrain; mock emits `speech_started`; assert playout goes silent within ≤1 tick (20 ms); mock emits fresh `audio_out`; assert playout resumes. - New `ReflexMetrics` (`barge_in_count`, `advisory_dropped`, `frames_suppressed`) mirroring `TapMetrics` shape (atomics, snapshot fn). Threaded through the same `TapConn.metrics` surface where reasonable, or a new side-car. - Thorough learner-facing comments on the new std-thread / channel-bridge / wrapper-decorator patterns (slice-1 §7 standard carries over). ### 1.2 Out of scope (with scheduled return) | Deferred item | Returns in | Why deferred | |---|---|---| | Local VAD **tuning framework** (configurable thresholds, per-environment calibration, adaptive noise floor) | post-spearhead refinement | The VAD itself (RMS/energy detector + debounce) IS in scope for slice-4 (the primary barge-in trigger, proving wedge #1). Only the *tuning* framework — configurable thresholds, calibration UI, adaptive noise floors — is deferred. The MVP ships with a single `const` threshold + N-frame debounce, exercised by the e2e test with a synthetic loud signal. Tuning to real-world noise conditions is post-spearhead. | | Per-session media threads / threadpool shard | later rung | Single thread covers spearhead scale (loopback dev + low-concurrency PSTN via slice-5). The command-channel seam between axum and the thread makes the graduation to a threadpool shard localized. | | Trickle ICE | later | Unchanged from slice-1 deferral. | | Min-mute floor / inter-word-gap debouncing | post-spearhead | `SpeechStopped` is a no-op for mute; a floor timer on resume would protect against brain-yield races (brain emits fresh `audio_out` before the caller's inter-word gap ends). Defer until observed in practice. | | Brain-side `input_audio_buffer.interrupt` / `clear` on barge | slice-5 or brain-side | Whether the brain should clear its own input buffer on `speech_started` is a brain-UX decision, not a FOB one; the FOB only kills *playout* (its half-duplex gate). The advisory already tells the brain what happened; the brain's response is its own concern. | | Half-duplex gating beyond playout kill | later rung | Barge-in is the first half-duplex reflex; full HD gating (mixing, jitter buffer interaction, multi-party) arrives with conferencing. | | TLS on HTTP / WSS | slice-5 | Unchanged. | | Authn / authz / multi-tenancy | slice-6 | Unchanged. | | Spend cap / abuse gate | slice-6 | Unchanged. | | Browser-based automated e2e (Playwright/Selenium) | post-spearhead | Unchanged. The synthetic-peer harness from slice-2/3 is the test vehicle. | --- ## 2. Architecture delta ### 2.1 The reflex wrapper `Reflex` is a zero-cost-style decorator around any `AudioPipe`. It sits between `RtcSession.pipe` (which `loop_driver::drive` calls via `session.pipe.next_pcm_frame()`) and the concrete pipe (`TapAudioPipe` in production, `EchoAudioPipe` in slice-1's unit tests). `loop_driver` is oblivious to the wrapper: it still calls `session.pipe.next_pcm_frame()`, the dynamic dispatch through `Box` lands in `Reflex::next_pcm_frame`, which applies the state machine and delegates to `inner.next_pcm_frame()` per the table in §3.2. The reflex owns three pieces of state: - `advisory_rx: mpsc::Receiver` — drained sync-non-blocking via `try_recv` on the 20 ms tick before delegating to `inner`. Fed by TWO senders on the same channel: the outer `LocalVadReflex` (local VAD, the primary trigger) and the TapEngine task (the brain's advisory, secondary/confirmation) — both over tokio mpsc. - `muted: bool` — the kill state. `next_pcm_frame` returns `None` while muted, *unless* the inner returns `Some` (the resume condition — the first fresh `audio_out` clears mute). - `barge_epoch: u64` — incremented on every `SpeechStarted`. Load-bearing this slice: the local VAD (primary) and the brain's advisory (secondary) can both fire on the same barge, and the epoch distinguishes a genuine re-barge from the slower confirmation landing on the same event (§3.2). The flush + drain keeps resume race-free regardless. ### 2.2 The dedicated media thread A single `std::thread::spawn` replaces the tokio `spawn_poll_task`. The thread owns `HashMap` **exclusively** — no `Arc>` shared with axum. All access from the axum handlers is via a command channel: ```rust enum MediaCmd { AcceptOffer { id: ChannelId, sdp: String, reply: oneshot::Sender> }, Delete { id: ChannelId, reply: oneshot::Sender<()> }, Shutdown { reply: oneshot::Sender<()> }, } ``` The thread loop per 10 ms meta-tick: 1. Drain `cmd_rx` via `try_recv` loop — handle all pending commands before ticking. 2. For each session in the map: drain the per-session `flush_rx` side-channel (slice-2's existing disconnect-flush signal) BEFORE `run_poll_once`, then call `RtcSession::run_poll_once(now)` (the unchanged `loop_driver::drive`). 3. After `run_poll_once`, observe `channel.state`: - `Connected && tap.is_none()` → spawn the TapEngine (tokio task via the `tokio::runtime::Handle` captured at thread-start) + wire `Reflex` as the session's pipe. Mirror of slice-2's spawn seam, relocated from `session_map.rs::drive_all_sessions` to here. - `Closed` → remove the entry + drop the session. 4. `std::thread::sleep(Duration::from_millis(10))` — 10 ms meta-tick. (Stable API: `std::thread::sleep_until` is nightly-only; `sleep(dur)` is the stable path. The 20 ms outbound encode tick is driven inside `loop_driver::drive` (unchanged); the 10 ms meta-tick gives finer resolution so str0m's `Timeout` outputs are honored promptly.) **The tokio ↔ std-thread bridge:** all channels are tokio mpsc/oneshot (constructable on tokio, drainable via `try_recv`/`blocking_recv` from any thread). The `tokio::runtime::Handle` captured at `MediaThread::spawn` time is used on the std thread to `handle.spawn(...)` the TapEngine when the `Connected` transition fires. No async code runs on the std thread itself — only sync channel ops + `RtcSession::run_poll_once`. **Why a single thread, not per-session:** spearhead scale. One loopback peer at a time in dev; even at low PSTN concurrency (slice-5) one thread drives dozens of sessions in 10 ms. Per-session threads arrive when the threadpool shard model lands (deferred). The command-channel seam between axum and the thread makes that graduation localized. ### 2.3 The hot-path audit (ADR-0002 honored) ADR-0002's load-bearing rule: *"the control↔media gRPC hop on the per-call hot path is removed."* Slice 4 does not re-introduce a hop: - The reflex's kill decision happens **inside** `Reflex::next_pcm_frame` on the dedicated thread — no channel send, no cross-thread coordination on the 20 ms tick. The advisory arrives via a `try_recv` drain (sync, non-blocking). - axum → media-thread is **cold-path only** (SDP accept, DELETE). None of it runs on the 20 ms tick. - The brain WS ↔ TapEngine (tokio) path is unchanged from slice-3. The advisory channel is a *third* mpsc alongside the existing `tx_pcm_in`/`rx_audio_out`/`flush_tx` — same pattern, additive. **The fused vertical stays fused.** ADR-0002 honored. --- ## 3. Component design ### 3.1 `AdvisoryEvent` enum ```rust // crates/rutster-media/src/reflex.rs /// 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 §3.2 state table). SpeechStopped { at: Instant }, } ``` ### 3.2 `Reflex

` state machine | Current state | Event | Action | New state | |---|---|---|---| | Playing | `SpeechStarted` | `muted=true`; `epoch++`; `inner.barge_in_flush()` (clear ring + drain `rx_audio_out` so stale brain frames queued pre-barge are dropped); `metrics.barge_in_count++` | Muted | | Muted | `SpeechStarted` (duplicate/re-barge) | `epoch++`; `barge_in_flush()` again (fresh barge resets the "fresh audio" clock); `barge_in_count++` | Muted | | Muted | `SpeechStopped` | increment `advisory_observed_speech_stopped` counter; **no state change** | Muted | | Playing | `SpeechStopped` | increment counter; **no state change** | Playing | | Muted | inner `next_pcm_frame()` returns `Some(f)` (fresh brain audio arrived post-barge) | `muted=false`; return `Some(f)` | Playing | | Muted | inner `next_pcm_frame()` returns `None` | return `None` (silence); `metrics.frames_suppressed++` | Muted | **Why `SpeechStopped` is a no-op for mute:** per the resume-semantics decision (resume on first fresh `audio_out`). The brain's `speech_stopped` is *observed* (counter) but doesn't gate — this avoids the inter-word-gap problem (caller pauses, VAD fires stopped, brain un-mutes too early, brain's audio overlaps caller's next word). The resume condition is "the brain has yielded and started a new response," which is provably signaled by the first `audio_out` frame after the barge — not by the caller's silence. **Why `epoch`:** with two concurrent trigger sources this slice — the local VAD (primary) and the brain's advisory (secondary/confirmation, §6.1) — the epoch disambiguates "is this barge a re-barge of the same event, or a new one" when both race into the same advisory channel. The local VAD trips first; the brain's slower ASR-grade advisory lands ~300 ms later on the *same* event and must not be counted as a fresh barge. The epoch is that disambiguator — load-bearing now, not a forward-compat seam. ### 3.3 `AudioPipe` trait extension ```rust // crates/rutster-media/src/pcm.rs — additive method on `AudioPipe` /// Barge-in flush: clear the playout ring AND drain the inbound brain /// audio queue of any frames queued before the barge. 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(); } ``` `TapAudioPipe` overrides: ```rust // crates/rutster-tap/src/tap_audio_pipe.rs fn barge_in_flush(&mut self) { // Clear the playout ring (drops buffered brain-proposed frames). self.playout_ring.clear(); // Drain rx_audio_out of any frames the engine task queued before // the barge. Without this, a stale 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. while self.rx_audio_out.try_recv().is_ok() { self.metrics.barge_drained_inflight.fetch_add(1, Ordering::Relaxed); } } ``` ### 3.4 `LocalVadReflex

` — the primary trigger (FOB-local, zero brain round-trip) The decorator that proves wedge #1. Wraps any `AudioPipe`; inspects the caller's decoded PCM in `on_pcm_frame`, computes RMS energy, fires `AdvisoryEvent::SpeechStarted` into the inner `Reflex`'s advisory channel when local VAD trips. Composes as `LocalVadReflex>` — the outer wrapper does local-VAD; the inner wrapper applies the mute state machine to the advisory stream (which now has TWO sources: the local VAD + the brain's advisory, both feeding the same mpsc). ```rust // crates/rutster-media/src/reflex.rs /// RMS energy threshold for caller-speech detection. 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. The synthetic-peer e2e test sends a frame with samples /// well above this threshold so the trip is deterministic. /// /// i16 max is 32767; a |sample| average of ~500 (~1.5% of full scale) is /// a quiet-but-unmistakable signal; room-tone background sits well below. /// The debounce (N consecutive frames above threshold) filters transient /// spikes (clicks, nose-breaths) without slowing the trip materially. pub const VAD_RMS_THRESHOLD: f64 = 500.0; /// Number of consecutive above-threshold frames required before the VAD /// trips. At 20 ms/frame, N=3 = 60 ms of above-threshold audio — well /// below the brain's ~300 ms ASR-VAD latency, comfortably instant for /// the wedge-#1 demonstration. Tunable in a later slice; const for MVP. pub const VAD_DEBOUNCE_FRAMES: u32 = 3; pub struct LocalVadReflex { inner: P, /// The advisory_tx the LocalVadReflex pushes into when local VAD /// trips. This is the SAME channel the brain's advisories arrive /// on (the Reflex holds the paired rx) — so the Reflex's mute state /// machine sees both sources uniformly via `drain_advisories`. advisory_tx: mpsc::Sender, /// Consecutive above-threshold frame count (the debounce counter). above_threshold_streak: u32, /// True once the VAD has tripped for the current caller-speech /// burst. Reset to false when the streak breaks (caller stopped) /// — so a new burst trips a fresh `SpeechStarted`. 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, // armed on construction } } /// Compute RMS energy of a PCM frame. The frame is 480 i16 samples /// @ 24 kHz; RMS = sqrt(mean(sample²)). Cheap: ~480 multiplications, /// one division, one sqrt — well under the 20 ms tick budget. 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. /// Called from `on_pcm_frame` (the sink path). Returns true if the /// VAD tripped THIS call (so the caller can push the advisory). 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; // disarm until caller stops return true; // trip! } } else { // Caller went quiet → re-arm for the next speech burst. self.above_threshold_streak = 0; self.vad_armed = true; } false } } impl AudioSource for LocalVadReflex

{ fn next_pcm_frame(&mut self) -> Option { // Pure delegation — the VAD only observes the SINK path // (inbound caller audio); playout is the inner Reflex's concern. self.inner.next_pcm_frame() } } impl AudioSink for LocalVadReflex

{ fn on_pcm_frame(&mut self, frame: PcmFrame) { // THE PRIMARY TRIGGER: inspect the caller's audio BEFORE delegating. // If the local VAD trips, push an advisory into the same channel the // brain's advisories arrive on — the inner Reflex drains it on the // next next_pcm_frame call + applies the kill. Zero brain round-trip. 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; a missed // local-VAD trip here is not catastrophic — the brain will fire // its ASR-VAD ~300 ms later. } // Delegate to inner — the caller's audio still reaches the brain. 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() } } ``` **Why the VAD trips `on_pcm_frame` (sink), not `next_pcm_frame` (source):** the caller's audio is what the VAD inspects — and caller audio arrives via the sink path (decoded from the peer's RTP by `loop_driver::drive` → `session.pipe.on_pcm_frame(pcm)`). The source path is the brain's `audio_out` — the playout being killed, not the signal being detected. **Why the SAME advisory channel as the brain:** the inner `Reflex` drains all advisories uniformly — it doesn't care whether the source is local VAD or the brain's ASR. The brain's advisory arriving ~300 ms later either finds the kill already applied (no-op, the `muted` flag is already true) or confirms it. This is the composition pattern §6.4 anticipated; the `Reflex

` wrapper shape was designed for exactly this layering. ### 3.5 `Reflex

` struct + impl ```rust // crates/rutster-media/src/reflex.rs pub struct Reflex { inner: P, advisory_rx: mpsc::Receiver, muted: bool, barge_epoch: u64, 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 §3.2. } } } } } impl AudioPipe for Reflex

{ fn next_pcm_frame(&mut self) -> Option { self.drain_advisories(); if self.muted { // Muted: pull from inner. Some(f) = fresh brain audio arrived // post-barge → un-mute + return. None = silence, stay 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() } } 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) } fn clear_playout_ring(&mut self) { // The reconnect-flush path (slice-2 §5.3) still works through the // wrapper. If it fires during mute, the ring stays empty and mute // clears on the next post-reconnect audio_out. self.inner.clear_playout_ring() } fn barge_in_flush(&mut self) { // Allow the outer `LocalVadReflex` (primary trigger) to barge the inner. self.inner.barge_in_flush() } } ``` ### 3.6 `ReflexMetrics` Mirror of `TapMetrics` shape (atomics + snapshot struct): ```rust // crates/rutster-media/src/reflex.rs #[derive(Default)] pub struct ReflexMetrics { pub barge_in_count: AtomicU64, pub advisory_dropped: AtomicU64, // advisory channel full (e.g. 16-cap) pub frames_suppressed: AtomicU64, // None returns while muted pub advisory_observed_speech_stopped: AtomicU64, } pub struct ReflexMetricsSnapshot { pub barge_in_count: u64, pub advisory_dropped: u64, pub frames_suppressed: u64, pub advisory_observed_speech_stopped: u64, } // `barge_drained_inflight` lives on `TapMetrics` (in `rutster-tap`), not // `ReflexMetrics`, because the drain happens inside `TapAudioPipe::barge_in_flush`, // not inside `Reflex`. The path: `Reflex::drain_advisories` calls // `inner.barge_in_flush()` which is `TapAudioPipe::barge_in_flush`, which is // where the `rx_audio_out` drain + the counter increment happen. ``` --- ## 4. The dedicated media thread ### 4.1 `MediaThread` ```rust // crates/rutster/src/media_thread.rs pub struct MediaThread { cmd_tx: mpsc::Sender, join: Option>, } enum MediaCmd { AcceptOffer { id: ChannelId, sdp: String, reply: oneshot::Sender> }, Delete { id: ChannelId, reply: oneshot::Sender<()> }, Shutdown { reply: oneshot::Sender<()> }, } ``` Spawned at binary startup (`main.rs`), before `axum::serve`. The thread captures a `tokio::runtime::Handle` (to spawn TapEngine tasks when `Connected` transitions fire) and owns `HashMap` + (per-session, lazily) the TapConn / advisory_rx / Reflex wrapper. ### 4.2 Thread loop (per 10 ms meta-tick) 1. `cmd_rx.try_recv()` loop — handle ALL pending commands before ticking. `AcceptOffer` calls `RtcSession::accept_offer(sdp)` and replies via the oneshot. `Delete` fires `close_tx` + bounded-await the engine task (750 ms cap via `tokio::runtime::Handle::block_on(timeout(...))`) — the std thread briefly enters the tokio runtime to await; cold-path, not the 20 ms tick. `Shutdown` drains + replies. 2. For each `RtcSession` in the map: - Drain per-session `flush_rx` side-channel (slice-2's existing disconnect-flush) BEFORE `run_poll_once`. - Call `RtcSession::run_poll_once(now)` — the unchanged `loop_driver::drive`. - Observe `channel.state`: - `Connected && tap.is_none()` → `handle.spawn(spawn_tap_engine(...))` to bring up the tokio task; construct `Reflex::new(TapAudioPipe::new(...), advisory_rx, metrics)`; call `RtcSession::set_pipe(reflex)`. Mirror of slice-2's spawn seam. - `Closed` → remove the entry (drops the `RtcSession` + its pipe + advisory ends). 3. `std::thread::sleep(Duration::from_millis(10))` — 10 ms meta-tick. ### 4.3 `session_map.rs` rewire `SessionEntry` loses `rtc: Arc>`, gains `cmd_tx: mpsc::Sender` (cloned per-entry; cheap). `tap_url` stays (the thread reads it when spawning the engine). `tap_conn: Option` moves onto the media thread (the thread owns it after spawn). - `AppState::create_session` → sends a `Register { tap_url, reply }` command to the media thread; the **thread** constructs `RtcSession::new()` (saves a cross-thread move of the struct + keeps all `RtcSession` construction on the thread that owns it). The thread replies with `(id, cmd_tx_for_this_session)`; axum stores `SessionEntry { cmd_tx, tap_url, tap_conn: None }`. - `AppState::get(id)` (SDP path) → `cmd_tx.send(AcceptOffer { ... }).await` + `reply.await`. Cold-path; the axum handler is async. - `AppState::close(id)` → `cmd_tx.send(Delete { id, reply }).await` + `reply.await`. The reply returns after the TapEngine teardown completes on the thread. - `spawn_poll_task` → `spawn_media_thread`: constructs the channels, spawns the std thread, stores `cmd_tx` + `join` in `AppState`. Same idempotent-guard pattern. ### 4.4 TapEngine extension `spawn_tap_engine` returns a third channel end: `advisory_tx: mpsc::Sender`. The pump loop, on receiving `speech_started` / `speech_stopped` from the brain (slice-3 already decodes these in the tap protocol layer — `protocol_events.rs`), pushes the corresponding `AdvisoryEvent` into `advisory_tx`. If the channel is full, drop + count (hot-path "drop + observe" policy; an advisory is a hint, not a command). The `Reflex` wrapper holds `advisory_rx`. ### 4.5 `MockRealtimeBrain` extension `rutster-brain-realtime/src/mock.rs` gains a programmable advisory schedule: the test can register "after N `audio_in` frames received, send `speech_started`" and "after M more, send `speech_stopped`". The mock already asserts `turn_detection: null` on `session.update` (slice-3's S4 lock); slice-4 keeps that assertion. --- ## 5. Data flow ### 5.1 Barge-in (the kill) — primary path: local VAD, zero brain round-trip ``` 1. caller speaks into mic → peer RTP → str0m decode → on_pcm_frame → LocalVadReflex::on_pcm_frame INSPECTS the frame FIRST: • rms(frame) computed (~480 muls, < 1 µs) • if rms ≥ VAD_RMS_THRESHOLD: streak++; if streak ≥ VAD_DEBOUNCE_FRAMES (3 = 60 ms) → TRIP • on trip: try_send(AdvisoryEvent::SpeechStarted) into advisory_tx (same channel as brain's advisories) → disarm until caller goes quiet → THEN delegates to inner.on_pcm_frame(frame) → tx_pcm_in → TapClient → audio_in (WS) → brain 2. [THREE HOPS HAPPEN HERE, ON THE BRAIN SIDE — NOT on the kill path]: - brain's ASR-VAD fires (~300 ms later, slower but more accurate) - brain sends speech_started back over WS - TapEngine → advisory_tx (the SAME mpsc; both sources feed one channel) 3. media thread next 20 ms tick → Reflex::next_pcm_frame → drain_advisories: • The LOCAL VAD's SpeechStarted arrives FIRST (it was pushed in step 1, same thread, no WS hop) → muted=true; epoch++; inner.barge_in_flush() (ring cleared + rx_audio_out drained) • The brain's SpeechStarted arrives ~300 ms LATER → Reflex sees muted=true already → re-barge (epoch++, barge_in_flush again — harmless, mute stays) → returns None (silence) for this + subsequent ticks while muted 4. loop_driver::drive pulls None from pipe → encodes Opus silence → peer hears silence (the brain's in-flight audio_out frames are dropped; no overlap with caller's speech) ``` **Latency budget:** caller speaks → kill fires in ≤ 60 ms (3 debounce frames × 20 ms tick) + one tick to drain the advisory + apply the kill = ≤ 80 ms wallclock. Zero brain round-trip on the primary path. (The brain's ASR-VAD advisory arrives ~300 ms later — it confirms the kill but doesn't gate it.) This is what ARCHITECTURE.md:80 demands ("the brain round-trip is too slow to enforce them") and the proof wedge #1 rests on. ### 5.2 Resume (the un-mute) ``` 1. brain decides to yield/respond → sends a fresh audio_out frame (provably post-barge: barge_in_flush drained rx_audio_out) 2. TapClient → audio_out (WS) → TapEngine → tx_audio_out → rx_audio_out → playout ring 3. media thread 20 ms tick → Reflex::next_pcm_frame → drain_advisories (empty) → muted=true → inner.next_pcm_frame() returns Some(f) (fresh brain audio) → muted=false; return Some(f) 4. loop_driver encodes + writes → peer hears the brain's new response ``` ### 5.3 Cold-path (axum ↔ media thread) ``` - POST /v1/sessions → AppState::create_session → MediaCmd::Register → thread constructs RtcSession → reply(id) - POST /v1/sessions/{id}/offer → AppState::get + cmd_tx.send(AcceptOffer) → thread.lock(session).accept_offer(sdp) → reply(answer) - DELETE /v1/sessions/{id} → AppState::close → cmd_tx.send(Delete) → thread: fire close_tx, bounded-await engine task teardown → reply - graceful shutdown → cmd_tx.send(Shutdown) → thread drains + drops → reply → join ``` --- ## 6. Why these decisions ### 6.1 Why both, local VAD primary (revised after adversarial review) The initial brainstorming landed on advisory-only for MVP (cheapest path to a working barge-in; the brain's VAD already runs for STT). The 2026-07-01 adversarial review surfaced the load-bearing problem with that choice: README:98-100 + ARCHITECTURE.md:79-81 rest the wedge on "local reflexes that don't need the brain — VAD killing TTS the instant the caller speaks." Advisory-only puts the brain round-trip in the trigger path (brain VAD → WS → TapEngine → mpsc → Reflex); the kill DECISION is in-core but the TRIGGER SOURCE crossed the brain. The spearhead's step 4 would prove a reflex that depends on the brain — the opposite of the property steps 1-4 exist to prove. The revision: **local VAD in `on_pcm_frame` is the primary trigger** (RMS/energy detector on the dedicated thread, in the 20 ms loop, zero brain round-trip — the actual wedge-#1 proof). Slice-3's advisory becomes the **secondary/confirmation** signal — the brain's ASR-quality VAD confirms the local kill ~300 ms later (slower but more accurate — it knows words, not just energy). The two sources feed the SAME advisory mpsc; the `Reflex` wrapper drains them uniformly. The composition is `LocalVadReflex>` — the decorator pattern from §6.4, which was originally specced as deferred to "when local VAD arrives." It arrived. - The VAD itself (~25 lines: RMS, threshold, debounce) is in scope. The **tuning framework** (configurable thresholds, per-environment calibration, adaptive noise floor) is deferred — the MVP ships a single `const` threshold + N-frame debounce, exercised by the e2e test with a synthetic loud signal. - Slice-3's advisory plumbing + `MockRealtimeBrain` schedule aren't wasted — they become the secondary path, still exercised by the e2e test (the brain's advisory fires SLIGHTLY after the local VAD; both feeds land in the Reflex's drain). - The `Reflex

` wrapper shape already supported composition — no structural rework, just a new `LocalVadReflex

` decorator landing in scope (Task 2b in the plan). ### 6.2 Why resume on first fresh `audio_out` (not `speech_stopped`) - The "the brain has yielded and started a new response" condition is provably signaled by the first `audio_out` frame after the barge — not by the caller's silence. `speech_stopped` fires between words; resuming on it un-mutes too early (inter-word-gap overlap). - The `barge_in_flush` drain of `rx_audio_out` makes the resume race-free: the first `audio_out` observed post-barge is provably post-barge (frames queued pre-barge are dropped in the flush). ### 6.3 Why a single dedicated thread (not per-session) - Spearhead scale: one loopback peer in dev; even at low PSTN concurrency (slice-5), one thread drives dozens of sessions in 10 ms. - The command-channel seam between axum and the thread makes the graduation to a threadpool shard localized — when per-CPU-shard threading arrives, it's a fan-out of the `cmd_rx`/`HashMap` shape, not a redesign. - Per-session threads arrive when load demands; the spearhead's "shortest blocking path" rule dislikes spawning work per session that may not need it (pre-ICE-connected sessions would redundantly spin). ### 6.4 Why `Reflex

` as a wrapper (not inline in `TapAudioPipe`) - Composition: `LocalVadReflex

` composes outside the advisory `Reflex

` (§6.1), the same way `Reflex` composes. The pattern (decorator over `AudioPipe`) stacks the two trigger sources without restructuring either. - The seam: `loop_driver.rs` byte-identical (still calls `pipe.next_pcm_frame()`). If the reflex lived inline in `TapAudioPipe`, the binary-side wiring would still change but the `TapAudioPipe` module itself would grow the reflex state — less isolated. - The payoff is realized this slice, not deferred: two stacked reflexes (local-VAD primary + advisory secondary) live as independent, separately-testable decorators rather than one module's commingled state. Keeping them as wrappers is what makes that separation free. ### 6.5 Why `barge_in_flush` on `AudioPipe` (not just `clear_playout_ring`) - `clear_playout_ring` (slice-2) clears the *ring*. `barge_in_flush` clears the ring AND drains the *inbound brain queue* (`rx_audio_out`). The distinction matters: on a brain disconnect (slice-2's case), the brain is gone — `rx_audio_out` will drain itself on the next `Disconnected` `try_recv`. On a barge-in, the brain is alive and may have queued frames pre-barge that would un-mute immediately if not drained here. Two different "clear the playout path" semantics, two methods. --- ## 7. Done-criteria 1. `cargo test --all` passes (stable + 1.85, the CI matrix). 2. `cargo fmt --check` + `cargo clippy -- -D warnings` clean. 3. `loop_driver.rs` + `rtc_session.rs` **byte-identical** to slice-3 — CI-asserted via `git diff --exit-code main -- crates/rutster-media/src/loop_driver.rs crates/rutster-media/src/rtc_session.rs` (the §8.5 #6 seam gate, restated for slice-4). 4. Dedicated media thread drives sessions off the tokio pool; `MediaThread` integration test passes (AcceptOffer / Delete / Shutdown). 5. `Reflex` state-machine unit tests all pass: - `SpeechStarted` → next `next_pcm_frame` returns None even if ring has frames. - `SpeechStarted` then `inner.next_pcm_frame()=Some` → un-mutes, returns the frame. - `SpeechStopped` during Muted → stays Muted. - `SpeechStopped` during Playing → no-op. - Duplicate `SpeechStarted` re-flushes + stays Muted. - Metrics counters (`barge_in_count`, `frames_suppressed`) increment correctly. - `advisory_rx` full → `advisory_dropped` increments, no panic. 6. `LocalVadReflex` unit tests pass: - RMS computation verified against a known-loud + known-quiet frame. - Debounce: N-1 above-threshold frames do NOT trip; the Nth does. - Re-arm: above-threshold → trip; below-threshold → re-arm; next streak trips again. - `on_pcm_frame` ALWAYS delegates to inner (caller audio reaches the brain even during barge). 7. `barge_in_flush` unit tests pass (ring + `rx_audio_out` drain). 8. Barge-in e2e (PRIMARY PATH, proves wedge #1): synthetic loud caller audio → playout killed within ≤4 ticks (≤80 ms wallclock: 3 debounce + 1 drain+apply) **WITHOUT any brain advisory**. The brain's `speech_started` advisory arrives later + is a no-op (mute already applied). 9. Barge-in e2e (SECONDARY PATH, exercises slice-3 advisory plumbing): `MockRealtimeBrain` emits `speech_started` on schedule; local VAD is NOT tripped (quiet synthetic caller audio); advisory → kill → fresh `audio_out` → resume. Proves the advisory→reflex→kill path still works. 10. S4 turn-ownership lock preserved: `MockRealtimeBrain` still asserts `turn_detection: null` on `session.update` (slice-3's #7, unchanged). 11. `MockRealtimeBrain` extended to emit `speech_started`/`speech_stopped` on schedule. 12. `cargo doc --no-deps` renders the new `reflex.rs` + `media_thread.rs` module/item docs cleanly (learner-facing comments present per AGENTS.md code style). --- ## 8. Open decisions - ~~Trigger source~~ — decided: **both, local VAD primary + advisory secondary** (revised after 2026-07-01 adversarial review; initial brainstorming landed on advisory-only, the review surfaced that advisory-only contradicts wedge #1's "VAD killing TTS the instant the caller speaks, without the brain"). Local VAD in `on_pcm_frame` is the primary trigger; the brain's `speech_started`/`speech_stopped` advisory is the secondary/confirmation. - ~~Resume semantics~~ — decided: first fresh `audio_out` post-barge; `SpeechStopped` observational only. - ~~Thread model~~ — decided: single dedicated `std::thread`; per-session/threadpool deferred. - **`MockRealtimeBrain` advisory schedule API shape** — landed in §4.5 as a programmable "after N audio_in frames" schedule. Could alternatively be a free-form `Vec<(trigger_frame_count, AdvisoryEvent)>` queue. The plan will pin the concrete API. - **Thread shutdown ordering vs TapEngine teardown** — `Delete` command handler fires `close_tx` + bounded-await the engine task (750 ms cap via `tokio::runtime::Handle::block_on(timeout(...))`); the reply oneshot returns after teardown. Cold-path, std thread briefly enters the tokio runtime to await. Documented as an acceptable deviation (not the 20 ms tick). --- ## 9. Cross-references - [slice-1 spec](2026-06-28-slice-1-webrtc-loopback-design.md) — the media loop + the seam (`AudioSource`/`AudioSink` traits in `rutster-media`); slice-1 §8.5 #6 is the seam gate this slice re-affirms. - [slice-2 spec](2026-06-28-slice-2-agent-tap-design.md) — the tap interface, the `TapAudioPipe`, the core-authoritative playout buffer (§4.1), the `flush_tx` side-channel pattern that the `advisory_rx` mirrors. - slice-3 (merged `c30a452`) — `MockRealtimeBrain`, the translator, the `speech_started`/`speech_stopped` protocol events, the S4 turn-ownership lock. - [ADR-0002](../../adr/0002-north-star-and-fused-core.md) — fused vertical; the hot-path hop invariant this slice re-affirms (§2.3 audit). - [ADR-0008](../../adr/0008-fob-and-green-zone.md) — FOB/green-zone doctrine; the reflex is a FOB member (hot-path, security-constitutive for turn-taking, differentiating). - [ARCHITECTURE.md](../../ARCHITECTURE.md) — §"Media plane" ("Dedicated timing threads for the 20ms loop, never the shared tokio pool" — this slice lands it); §"Biggest technical risk" (the reflex loop *is* the remaining long pole). - [PORT_PLAN.md](../../PORT_PLAN.md) — §Phasing, step 4 = barge-in.