rutster/docs/superpowers/plans/2026-07-01-slice-4-barge-in.md

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<P: AudioPipe> 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 RtcSessions 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<P> 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<ChannelId, RtcSession> 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<AdvisoryEvent> 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<Mutex<RtcSession>> → cmd_tx: mpsc::Sender<MediaCmd>. 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<AdvisoryTrigger>) 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<P> 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<Self> + 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:

//! # Reflex — the FOB barge-in reflex (spec §3.1, §3.2; slice-4)
//!
//! `Reflex<P: AudioPipe>` 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<P>` composes outside the advisory
//! `Reflex<P>`, the same way `Reflex<TapAudioPipe>` 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<Self> {
        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):

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::...:

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.

# 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

cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all

Expected: all green.

• Step 6: Commit

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<P> 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<AdvisoryEvent>.

• Produces:

  • pub struct Reflex<P: AudioPipe> { inner: P, advisory_rx: mpsc::Receiver<AdvisoryEvent>, muted: bool, barge_epoch: u64, metrics: Arc<ReflexMetrics> }
  • impl<P: AudioPipe> Reflex<P> with pub fn new(inner: P, advisory_rx: mpsc::Receiver<AdvisoryEvent>, metrics: Arc<ReflexMetrics>) -> Self
  • impl<P: AudioPipe> AudioPipe for Reflex<P> (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:

    /// 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<PcmFrame>,
        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<PcmFrame> {
            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<MockPipe>, mpsc::Sender<AdvisoryEvent>, Arc<ReflexMetrics>) {
        let (tx, rx) = mpsc::channel::<AdvisoryEvent>(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<P> + the AudioPipe impl

Append (above the #[cfg(test)] mod tests) to crates/rutster-media/src/reflex.rs:

/// 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<dyn AudioPipe>`)
///
/// 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<dyn AudioPipe>` 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<P: AudioPipe> {
    pub(crate) inner: P,
    pub(crate) advisory_rx: mpsc::Receiver<AdvisoryEvent>,
    pub(crate) muted: bool,
    pub(crate) barge_epoch: u64,
    pub(crate) metrics: Arc<ReflexMetrics>,
}

impl<P: AudioPipe> Reflex<P> {
    pub fn new(
        inner: P,
        advisory_rx: mpsc::Receiver<AdvisoryEvent>,
        metrics: Arc<ReflexMetrics>,
    ) -> 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<P: AudioPipe> AudioSource for Reflex<P> {
    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
        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<P: AudioPipe> AudioSink for Reflex<P> {
    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<P: AudioPipe> AudioPipe for Reflex<P> {
    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

cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all

• Step 6: Commit

git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/lib.rs
git commit -m "feat(media): Reflex<P> 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:

    #[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:

pub barge_drained_inflight: AtomicU64,

In TapMetrics::new():

barge_drained_inflight: AtomicU64::new(0),

In TapMetrics::snapshot():

barge_drained_inflight: self.barge_drained_inflight.load(Ordering::Relaxed),

In MetricsSnapshot struct:

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):

    /// 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

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<AdvisoryEvent> 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:

    /// 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::<FunctionCallEvent>(8);
        let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
        let (tx_advisory, mut rx_advisory) =
            mpsc::channel::<rutster_media::AdvisoryEvent>(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<u64> = 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):

async fn handle_brain_frame(
    text: &str,
    last_seq_ingress: &mut Option<u64>,
    tx_audio_out: &mpsc::Sender<PcmFrame>,
    tx_function_call: &mpsc::Sender<FunctionCallEvent>,
    tx_advisory: &mpsc::Sender<rutster_media::AdvisoryEvent>,
    metrics: &Arc<TapMetrics>,
    session_start: Instant,
) {

Replace the DecodedPayload::SpeechStarted | DecodedPayload::SpeechStopped arm (lines 409-412) with:

        // 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<rutster_media::AdvisoryEvent>. 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).

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

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<mpsc::Sender<rutster_media::AdvisoryEvent>> (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<mpsc::Receiver<AdvisoryEvent>> 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<AdvisoryEvent>). 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:

    /// 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:

   let (tx_advisory, advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(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<mpsc::Receiver<rutster_media::AdvisoryEvent>>)

3. Update run_engine_loop signature to accept tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent> and pass it through to run_tap_client(..., tx_advisory, ...).

• Step 4: Update ALL existing call sites of spawn_tap_engine

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

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<TapAudioPipe> (from Task 2).

• Produces:

  • pub struct MediaThread { cmd_tx: mpsc::Sender<MediaCmd>, join: Option<std::thread::JoinHandle<()>> }
  • 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<MediaCmd> — 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:

//! # 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<ChannelId, RtcSession>` 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<TapAudioPipe>` 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<Result<ChannelId, String>>,
    },
    /// 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<Result<String, String>>,
    },
    /// 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<MediaCmd>,
    join: Option<JoinHandle<()>>,
}

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<ChannelId,
    /// ThreadSession>` 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<TapAudioPipe>` wrapper.
    tap_conn: Option<crate::tap_engine::TapConn>,
    /// The `advisory_rx` end stored UNTIL the Connect transition wires it
    /// into `Reflex::new` (then `None` — the Reflex consumed it).
    pending_advisory_rx: Option<mpsc::Receiver<rutster_media::AdvisoryEvent>>,
}

fn run_media_thread(
    mut cmd_rx: mpsc::Receiver<MediaCmd>,
    default_tap_url: url::Url,
    tokio_handle: tokio::runtime::Handle,
) {
    let mut sessions: HashMap<ChannelId, ThreadSession> = 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<ChannelId> = 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:

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

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<TapAudioPipe>
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<MediaCmd> 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<Mutex<RtcSession>> with cmd_tx: mpsc::Sender<MediaCmd>. create_session → sends Register command; get → cmd_tx.send(AcceptOffer).await (NOT what get used to return — get was returning the Arc<Mutex<RtcSession>> for post_offer to lock; replace with a new accept_offer(id, sdp) -> Result<String, String> async method that sends AcceptOffer + awaits the reply). close(id) -> ... → sends Delete + awaits. spawn_poll_task → spawn_media_thread (constructs MediaThread, stores cmd_tx).

// Sketch — the dev fills in the exact code.
#[derive(Clone)]
pub struct AppState {
    pub cmd_tx: mpsc::Sender<MediaCmd>,
    pub poll_running: Arc<Mutex<bool>>,
    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<url::Url>) -> Result<ChannelId, String> {
        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<String, String> {
        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.

rg 'self\.get\(|self\.sessions\.get\(' crates/rutster/src/routes.rs

• Step 4: Update main.rs

// 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

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

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<MediaCmd> 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:

    /// 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):

/// 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<AdvisoryTrigger>) {
        // Communicated to the accept_loop via a shared Arc<Mutex<Vec<...>>>.
        // The handle_connection task reads + drains the schedule per-connection.
        self.advisory_schedule = Some(Arc::new(Mutex::new(schedule)));
    }
}

Add advisory_schedule: Option<Arc<Mutex<Vec<AdvisoryTrigger>>>> 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<TapAudioPipe> (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<TapAudioPipe> 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):

      - 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

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<P> 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.