# 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 `RtcSession`s exclusively, driven by a command channel from axum (cold-path only). `loop_driver.rs` + `rtc_session.rs` stay byte-identical (the §8.5 #6 seam gate).

**Tech Stack:** Rust stable + 1.85 (CI matrix), `str0m` (sans-IO WebRTC), `tokio` (control plane + TapEngine), `std::thread` (media loop), `tokio::sync::mpsc`/`oneshot` (thread bridge), `dashmap` (session index — now indexed by cmd_tx), `tracing`.

## Global Constraints

- **License:** GPL-3.0-or-later on every crate manifest (ADR-0004).
- **Seam gate:** `loop_driver.rs` + `rtc_session.rs` byte-identical to slice-3 (CI `git diff --exit-code`).
- **Hot-path policy:** never `?`-propagate; match-and-continue; "drop + observe (log + counter), don't crash." No `unwrap()`/`expect()` outside tests or const-init.
- **Code style:** `cargo fmt` is the single whitespace source of truth. `clippy -D warnings` is the lint bar. Newtype wrappers over primitives (e.g. `ChannelId(Uuid)`).
- **Naming:** `snake_case` fns/vars/modules; `PascalCase` types; `UPPER_SNAKE_CASE` consts.
- **Learner-facing comments:** `//!` module docs, `///` item docs, `//` inline "why" — per AGENTS.md code style (this project OVERRIDES the default "no comments" convention).
- **Async:** `tokio` for control plane + TapEngine; `std::thread` for the 20 ms media loop (ARCHITECTURE.md mandate, landed this slice).
- **Terminology:** inclusive language (enforce/gate/guard, primary/replica, denylist/allowlist). Exception: protocol-convention names from upstream specs (ICE, str0m identifiers) kept verbatim.
- **Error handling:** cold path = `thiserror` + `?`; hot path = match-and-continue, no `?`.
- **CI gates:** `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all` (stable + 1.85), `cargo deny check`.

## File Structure

### New files

| Path | Responsibility |
|---|---|
| `crates/rutster-media/src/reflex.rs` | `AdvisoryEvent` enum, `Reflex<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 2b** — depends on Tasks 1+2. `LocalVadReflex<P>` — the PRIMARY trigger (RMS/energy VAD in `on_pcm_frame`, the wedge-#1 proof). Standalone unit-testable with a mock pipe. The decorator pattern from §6.4; composes as `LocalVadReflex<Reflex<TapAudioPipe>>` in Task 6's spawn site.
- **Task 3** — depends on Task 1. `TapAudioPipe::barge_in_flush` override + `TapMetrics.barge_drained_inflight`. Standalone.
- **Task 4** — depends on Tasks 1+3. `advisory_tx` threaded through `run_tap_client` + `handle_brain_frame`. The tap-client adhesive between the brain and the Reflex (the SECONDARY trigger path).
- **Task 5** — depends on Tasks 1+3+4. `spawn_tap_engine` returns `advisory_tx` end; TapConn gains `advisory_tx`.
- **Task 6** — depends on Tasks 2+2b+5. `MediaThread` — the binary-side dedicated std::thread; owns RtcSessions + spawns TapEngine + wires `LocalVadReflex<Reflex<TapAudioPipe>>` on `Connected`.
- **Task 7** — depends on Task 6. `session_map.rs` rewire + `main.rs` + `routes.rs` to the command-channel pattern.
- **Task 8** — depends on Task 5. `MockRealtimeBrain` advisory schedule.
- **Task 9** — depends on Tasks 6+7+8. Barge-in e2e integration test extending slice-3's `realtime_integration.rs`. TWO cases: (a) PRIMARY — loud local audio → kill WITHOUT advisory; (b) SECONDARY — quiet local audio + brain advisory → kill → fresh audio_out → resume.
- **Task 10** — depends on Task 7. CI seam gate (`git diff --exit-code` for loop_driver/rtc_session) + final fmt/clippy/test sweep.

Parallelizable-now filler (a blocked dev picks these up without blocking the critical path):
- LEARNING.md pointers to the new `reflex.rs` + `media_thread.rs` (after Task 2 + Task 6 land).
- README dev-loop updates (after Task 7 lands).
- `cargo doc` rendering checks (after Task 2 + Task 6).

---

### Task 1: `AdvisoryEvent` enum + `ReflexMetrics` + `barge_in_flush` trait method — the critical-path foundation

**Files:**
- Create: `crates/rutster-media/src/reflex.rs`
- Modify: `crates/rutster-media/src/pcm.rs:102-115` (the `AudioPipe` trait)
- Modify: `crates/rutster-media/src/lib.rs:33-40` (module declarations + re-exports)
- Test: `crates/rutster-media/src/reflex.rs` (inline `#[cfg(test)] mod tests`)

**Interfaces:**
- Consumes: `PcmFrame` (from `pcm.rs`), `AudioPipe` trait (from `pcm.rs`).
- Produces:
  - `pub enum AdvisoryEvent { SpeechStarted { at: Instant }, SpeechStopped { at: Instant } }`
  - `pub struct ReflexMetrics { barge_in_count: AtomicU64, advisory_dropped: AtomicU64, frames_suppressed: AtomicU64, advisory_observed_speech_stopped: AtomicU64 }` + `ReflexMetrics::new() -> Arc<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:

```rust
//! # 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`):

```rust
pub trait AudioPipe: AudioSource + AudioSink {
    /// Clear any buffered playout frames (slice-2 spec §5.3 step 4).
    fn clear_playout_ring(&mut self) {}

    /// Barge-in flush: clear the playout ring AND drain the inbound brain
    /// audio queue of any frames queued before the barge (slice-4 spec §3.3).
    /// Called by `Reflex` on `SpeechStarted`. The drain of `rx_audio_out`
    /// is what makes the resume condition race-free: the first `audio_out`
    /// observed post-barge is provably post-barge (frames queued pre-barge
    /// are dropped here).
    ///
    /// Default impl delegates to `clear_playout_ring` — sufficient for
    /// pipes without an inbound queue to drain (like `EchoAudioPipe`).
    fn barge_in_flush(&mut self) {
        self.clear_playout_ring();
    }
}
```

- [ ] **Step 3: Wire the module + re-exports in `lib.rs`**

Modify `crates/rutster-media/src/lib.rs` (around line 33-40). Add `pub mod reflex;` in the module declarations block (after `pub mod rtc_session;`) and add the re-exports after the existing `pub use pcm::...`:

```rust
pub mod loop_driver;
pub mod opus_codec;
pub mod pcm;
pub mod reflex;
pub mod rtc_session;

pub use opus_codec::{OpusDecoder, OpusEncoder};
pub use pcm::{AudioPipe, AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME};
pub use reflex::{AdvisoryEvent, Reflex, ReflexMetrics, ReflexMetricsSnapshot};
pub use rtc_session::{RtcSession, RtcSessionError};
```

(Leave the rest of `lib.rs` — `MediaError` etc. — unchanged.)

- [ ] **Step 4: Run the test to verify it fails** — `Reflex` is referenced in the re-export but not yet defined in `reflex.rs` beyond the enum/metrics. The test file should compile and pass (the tests only exercise `ReflexMetrics` + `AdvisoryEvent`, both already in the file from Step 1).

Run: `cargo test -p rutster-media --lib reflex::tests`
Expected: PASS (3 tests). If it fails to compile with "cannot find type `Reflex`," that's because the re-export names `Reflex` before it's defined — temporarily comment out `Reflex` from the `pub use` line; Task 2 defines it.

```bash
# Temporary: comment Reflex from the re-export until Task 2 lands
# pub use reflex::{AdvisoryEvent, ReflexMetrics, ReflexMetricsSnapshot};
```

- [ ] **Step 5: Run the full workspace test + fmt + clippy**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
```
Expected: all green.

- [ ] **Step 6: Commit**

```bash
git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/pcm.rs crates/rutster-media/src/lib.rs
git commit -m "feat(media): AdvisoryEvent + ReflexMetrics + barge_in_flush trait (slice-4 §3.1, §3.3)

The critical-path foundation for the barge-in reflex. AdvisoryEvent is
the enum carried over a tokio mpsc from TapEngine to Reflex (brain →
FOB). ReflexMetrics is the observable surface. barge_in_flush is the
new AudioPipe trait method (default delegates to clear_playout_ring) —
the kill-now path that clears the ring AND drains rx_audio_out.

Task 1 of the slice-4 plan. Everything else depends on this landing.""
```

---

### Task 2: `Reflex<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`:

```rust
    /// A minimal mock pipe for unit-testing Reflex. Captures on_pcm_frame
    /// inputs + returns a pre-loaded queue of frames from next_pcm_frame
    /// so we can simulate "brain audio_out arrived" deterministically.
    struct MockPipe {
        queued: std::collections::VecDeque<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`:

```rust
/// The FOB reflex decorator (slice-4 spec §3.2). Wraps any `AudioPipe`
/// with a barge-in state machine driven by `AdvisoryEvent`s from the brain.
///
/// # Why `P: AudioPipe` generic (not `Box<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**

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

- [ ] **Step 6: Commit**

```bash
git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/lib.rs
git commit -m "feat(media): Reflex<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 2b: `LocalVadReflex<P>` — the primary trigger (local VAD, zero brain round-trip)

**Files:**
- Modify: `crates/rutster-media/src/reflex.rs` (add the struct + impl)
- Modify: `crates/rutster-media/src/lib.rs` (re-export `LocalVadReflex` + the consts)
- Test: `crates/rutster-media/src/reflex.rs` (inline tests)

**Interfaces:**
- Consumes: `AdvisoryEvent`, `Reflex<P>` (from Task 2 — actually consumes the same `AdvisoryEvent` enum + `mpsc::Sender` it pushes into), `AudioPipe` trait + `PcmFrame`, tokio `mpsc::Sender<AdvisoryEvent>`.
- Produces:
  - `pub const VAD_RMS_THRESHOLD: f64 = 500.0;`
  - `pub const VAD_DEBOUNCE_FRAMES: u32 = 3;`
  - `pub struct LocalVadReflex<P: AudioPipe> { inner: P, advisory_tx: mpsc::Sender<AdvisoryEvent>, above_threshold_streak: u32, vad_armed: bool }`
  - `impl<P: AudioPipe> LocalVadReflex<P>` with `pub fn new(inner: P, advisory_tx: mpsc::Sender<AdvisoryEvent>) -> Self` + `fn rms(frame: &PcmFrame) -> f64` + `fn observe(&mut self, frame: &PcmFrame) -> bool`
  - `impl<P: AudioPipe> AudioSource for LocalVadReflex<P>` (pure delegation)
  - `impl<P: AudioPipe> AudioSink for LocalVadReflex<P>` (THE PRIMARY TRIGGER — inspects + delegates)
  - `impl<P: AudioPipe> AudioPipe for LocalVadReflex<P>` (pure delegation)

- [ ] **Step 1: Write the failing tests for the VAD state machine + RMS**

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

```rust
    /// RMS of a zeroed frame is 0.0 (perfect silence).
    #[test]
    fn rms_of_silence_is_zero() {
        let frame = PcmFrame::zeroed();
        assert_eq!(LocalVadReflex::<MockPipe>::rms(&frame), 0.0);
    }

    /// RMS of a loud frame is well above the threshold.
    #[test]
    fn rms_of_loud_frame_exceeds_threshold() {
        let mut frame = PcmFrame::zeroed();
        for s in frame.samples.iter_mut() {
            *s = 1000; // well above VAD_RMS_THRESHOLD (500.0)
        }
        assert!(LocalVadReflex::<MockPipe>::rms(&frame) >= VAD_RMS_THRESHOLD);
    }

    /// Debounce: N-1 above-threshold frames do NOT trip; the Nth does.
    #[tokio::test]
    async fn debounce_requires_n_consecutive_above_threshold_frames() {
        let (tx, mut rx) = mpsc::channel::<AdvisoryEvent>(16);
        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
        let mut loud = PcmFrame::zeroed();
        for s in loud.samples.iter_mut() { *s = 1000; }

        // VAD_DEBOUNCE_FRAMES - 1 frames: no trip.
        for _ in 0..(VAD_DEBOUNCE_FRAMES - 1) {
            vad.on_pcm_frame(loud.clone());
            assert!(rx.try_recv().is_err(), "no advisory before debounce threshold");
        }
        // Nth frame: trip!
        vad.on_pcm_frame(loud.clone());
        let ev = rx.try_recv().expect("advisory after debounce threshold");
        assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. }));
    }

    /// Re-arm: a below-threshold frame resets the streak + re-arms.
    #[tokio::test]
    async fn below_threshold_re_arms_vad() {
        let (tx, mut rx) = mpsc::channel::<AdvisoryEvent>(16);
        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
        let mut loud = PcmFrame::zeroed();
        for s in loud.samples.iter_mut() { *s = 1000; }
        let quiet = PcmFrame::zeroed();

        // Trip the VAD.
        for _ in 0..VAD_DEBOUNCE_FRAMES {
            vad.on_pcm_frame(loud.clone());
        }
        let _ = rx.try_recv().expect("first trip");

        // Caller goes quiet — re-arm.
        vad.on_pcm_frame(quiet);

        // Next streak trips again.
        for _ in 0..VAD_DEBOUNCE_FRAMES {
            vad.on_pcm_frame(loud.clone());
        }
        let ev = rx.try_recv().expect("second trip after re-arm");
        assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. }));
    }

    /// on_pcm_frame ALWAYS delegates to inner (caller audio reaches the brain
    /// even during barge — the FOB only kills playout, not the caller's path).
    #[tokio::test]
    async fn on_pcm_frame_always_delegates_to_inner() {
        let (tx, _rx) = mpsc::channel::<AdvisoryEvent>(16);
        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
        let frame = PcmFrame::zeroed();
        vad.on_pcm_frame(frame.clone());
        // The inner MockPipe captured it — verified by the lack of panic
        // + the MockPipe's on_pcm_frame being called (push_back_bounded
        // on the underlying queue, which we don't observe here directly;
        // the absence of a drop is the assertion).
    }

    /// next_pcm_frame is pure delegation — the VAD only observes the SINK path.
    #[tokio::test]
    async fn next_pcm_frame_delegates_to_inner() {
        let (tx, _rx) = mpsc::channel::<AdvisoryEvent>(16);
        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
        // Inner has no frames queued → None.
        assert!(vad.next_pcm_frame().is_none());
        // Queue a frame on the inner directly + verify it comes through.
        vad.inner.push_frame(PcmFrame::zeroed());
        assert!(vad.next_pcm_frame().is_some());
    }
```

- [ ] **Step 2: Run the test to verify it fails**

Run: `cargo test -p rutster-media --lib reflex::tests`
Expected: compile error — `LocalVadReflex` + `VAD_RMS_THRESHOLD` + `VAD_DEBOUNCE_FRAMES` don't exist yet.

- [ ] **Step 3: Implement `LocalVadReflex<P>`**

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

```rust
/// RMS energy threshold for caller-speech detection (slice-4 spec §3.4).
/// The MVP ships with a single tuned-for-synthetic-loud-signal const;
/// the tuning framework (per-environment calibration, adaptive noise
/// floor) is deferred per slice-4 §1.2.
pub const VAD_RMS_THRESHOLD: f64 = 500.0;

/// Number of consecutive above-threshold frames required before the VAD
/// trips (slice-4 spec §3.4). At 20 ms/frame, N=3 = 60 ms of above-
/// threshold audio — well below the brain's ~300 ms ASR-VAD latency.
pub const VAD_DEBOUNCE_FRAMES: u32 = 3;

/// The PRIMARY barge-in trigger (slice-4 spec §3.4): a local in-core
/// RMS/energy VAD running in `on_pcm_frame` on the dedicated thread, in
/// the 20 ms loop, with ZERO brain round-trip. Proves wedge #1 ("VAD
/// killing TTS the instant the caller speaks, without the brain" —
/// README:98-100, ARCHITECTURE.md:79-81). Composes as
/// `LocalVadReflex<Reflex<TapAudioPipe>>` — the outer wrapper does local
/// VAD; the inner wrapper applies the mute state machine to the advisory
/// stream (which has TWO sources: local VAD + brain advisory, both
/// feeding the same mpsc).
pub struct LocalVadReflex<P: AudioPipe> {
    pub(crate) inner: P,
    pub(crate) advisory_tx: mpsc::Sender<AdvisoryEvent>,
    pub(crate) above_threshold_streak: u32,
    pub(crate) vad_armed: bool,
}

impl<P: AudioPipe> LocalVadReflex<P> {
    pub fn new(inner: P, advisory_tx: mpsc::Sender<AdvisoryEvent>) -> Self {
        Self {
            inner,
            advisory_tx,
            above_threshold_streak: 0,
            vad_armed: true,
        }
    }

    /// Compute RMS energy of a PCM frame. ~480 multiplications + one
    /// sqrt — well under the 20 ms tick budget. Hot-path, no allocations.
    fn rms(frame: &PcmFrame) -> f64 {
        let sum_sq: u64 = frame.samples.iter()
            .map(|&s| (s as i64 * s as i64) as u64)
            .sum();
        (sum_sq as f64 / frame.samples.len() as f64).sqrt()
    }

    /// Inspect a caller PCM frame + apply the debounce state machine.
    /// Returns true if the VAD tripped THIS call (so on_pcm_frame can
    /// push the advisory). Called from `on_pcm_frame` (the sink path).
    fn observe(&mut self, frame: &PcmFrame) -> bool {
        let energy = Self::rms(frame);
        if energy >= VAD_RMS_THRESHOLD {
            self.above_threshold_streak += 1;
            if self.above_threshold_streak >= VAD_DEBOUNCE_FRAMES && self.vad_armed {
                self.vad_armed = false;
                return true;
            }
        } else {
            self.above_threshold_streak = 0;
            self.vad_armed = true;
        }
        false
    }
}

impl<P: AudioPipe> AudioSource for LocalVadReflex<P> {
    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
        self.inner.next_pcm_frame()
    }
}

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

impl<P: AudioPipe> AudioPipe for LocalVadReflex<P> {
    fn clear_playout_ring(&mut self) { self.inner.clear_playout_ring() }
    fn barge_in_flush(&mut self) { self.inner.barge_in_flush() }
}
```

Also: add `LocalVadReflex`, `VAD_RMS_THRESHOLD`, `VAD_DEBOUNCE_FRAMES` to `lib.rs`'s re-export.

- [ ] **Step 4: Run the test to verify it passes**

Run: `cargo test -p rutster-media --lib reflex::tests`
Expected: PASS (all prior tests + 6 new Task 2b tests).

- [ ] **Step 5: fmt + clippy + full test + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster-media/src/reflex.rs crates/rutster-media/src/lib.rs
git commit -s -m "feat(media): LocalVadReflex — primary barge-in trigger, zero brain round-trip (slice-4 §3.4)

The wedge-#1 proof. RMS/energy VAD in on_pcm_frame on the dedicated
thread, in the 20ms loop — caller speech trips SpeechStarted locally,
without any brain round-trip. Debounce (N=3 frames = 60ms) filters
transients. Composes as LocalVadReflex<Reflex<TapAudioPipe>>; both
the local VAD + the brain's advisory feed the same advisory_tx mpsc.
Revised after adversarial review (initial brainstorming was advisory-
only, which contradicts ARCHITECTURE.md:79-81)."
```

---

### Task 3: `TapAudioPipe::barge_in_flush` override + `TapMetrics.barge_drained_inflight`

**Files:**
- Modify: `crates/rutster-tap/src/tap_audio_pipe.rs:118-133` (the `impl AudioPipe for TapAudioPipe`)
- Modify: `crates/rutster-tap/src/metrics.rs:14-57` (add `barge_drained_inflight` field + snapshot)
- Test: `crates/rutster-tap/src/tap_audio_pipe.rs` (inline tests)

**Interfaces:**
- Consumes: `AudioPipe::barge_in_flush` default (from Task 1), `TapMetrics` struct.
- Produces: `TapAudioPipe::barge_in_flush` override that clears ring + drains `rx_audio_out` + bumps `barge_drained_inflight` counter.

- [ ] **Step 1: Write the failing test**

Append to `crates/rutster-tap/src/tap_audio_pipe.rs`'s `#[cfg(test)] mod tests`:

```rust
    #[test]
    fn barge_in_flush_clears_ring_and_drains_rx_audio_out() {
        let (_tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
        // Push 3 frames into the engine→playout mpsc + drain one into ring.
        for i in 0..3 {
            let mut f = PcmFrame::zeroed();
            f.samples[0] = i as i16;
            tx_audio_out.blocking_send(f).unwrap();
        }
        // Drain one into the ring (the queue has 1 in ring + 2 in mpsc).
        let _first = pipe.next_pcm_frame().expect("drained one");
        // Barge-in flush: clears ring + drains rx_audio_out.
        pipe.barge_in_flush();
        // Next frame should be None (ring empty, mpsc drained).
        assert!(pipe.next_pcm_frame().is_none());
        // Counter should reflect 2 frames drained from rx_audio_out.
        assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 2);
    }

    #[test]
    fn barge_in_flush_when_already_empty_is_noop() {
        let (_tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(_tx_pcm_in, rx_audio_out, metrics.clone());
        pipe.barge_in_flush();
        // No frames drained (none were queued); counter stays 0.
        assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 0);
    }
```

- [ ] **Step 2: Run the test to verify it fails**

Run: `cargo test -p rutster-tap --lib tap_audio_pipe::tests::barge`
Expected: FAIL — `barge_drained_inflight` field doesn't exist on `TapMetrics`; `barge_in_flush` not overridden on `TapAudioPipe` (the default impl just no-ops since `clear_playout_ring` on `TapAudioPipe` clears the ring but doesn't drain `rx_audio_out`).

- [ ] **Step 3: Add `barge_drained_inflight` to `TapMetrics`**

In `crates/rutster-tap/src/metrics.rs`, add the field + snapshot. The struct gains:

```rust
pub barge_drained_inflight: AtomicU64,
```

In `TapMetrics::new()`:

```rust
barge_drained_inflight: AtomicU64::new(0),
```

In `TapMetrics::snapshot()`:

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

In `MetricsSnapshot` struct:

```rust
pub barge_drained_inflight: u64,
```

- [ ] **Step 4: Override `barge_in_flush` on `TapAudioPipe`**

In `crates/rutster-tap/src/tap_audio_pipe.rs`, add to the `impl AudioPipe for TapAudioPipe` block (around line 118-133):

```rust
    /// slice-4 spec §3.3 — barge-in flush: clear the playout ring AND
    /// drain `rx_audio_out` of any frames queued before the barge. Without
    /// this drain, a stale brain frame in the mpsc would un-mute
    /// immediately on the next tick — defeating the "first fresh audio_out"
    /// resume condition. Hot-path: try_recv loop, bounded, no blocking.
    fn barge_in_flush(&mut self) {
        // Clear the ring (drops buffered brain-proposed frames).
        let cleared = self.playout_ring.len();
        self.playout_ring.clear();
        if cleared > 0 {
            debug!(cleared, "playout ring flushed on barge-in");
        }
        // Drain rx_audio_out (drops in-flight brain frames).
        let mut drained = 0usize;
        while self.rx_audio_out.try_recv().is_ok() {
            drained += 1;
        }
        if drained > 0 {
            self.metrics
                .barge_drained_inflight
                .fetch_add(drained as u64, Ordering::Relaxed);
            debug!(drained, "in-flight brain frames drained on barge-in");
        }
    }
```

- [ ] **Step 5: Run the test to verify it passes**

Run: `cargo test -p rutster-tap --lib tap_audio_pipe::tests`
Expected: PASS (all existing + 2 new).

- [ ] **Step 6: fmt + clippy + full test + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster-tap/src/tap_audio_pipe.rs crates/rutster-tap/src/metrics.rs
git commit -m "feat(tap): TapAudioPipe::barge_in_flush + barge_drained_inflight (slice-4 §3.3)

The kill-now path on the seam object: clears the playout ring AND drains
rx_audio_out of pre-barge in-flight brain frames. The drain is what makes
the resume condition race-free — the first audio_out post-barge is
provably post-barge."
```

---

### Task 4: `advisory_tx` threaded through `run_tap_client` + `handle_brain_frame`

**Files:**
- Modify: `crates/rutster-tap/src/tap_client.rs:140-270` (`run_tap_client` signature + select! arm), `:323-421` (`handle_brain_frame` signature + the `SpeechStarted`/`SpeechStopped` arms)
- Test: `crates/rutster-tap/src/tap_client.rs` (the `advisory_events_are_logged_not_forwarded` test gets updated + a new test)

**Interfaces:**
- Consumes: `AdvisoryEvent` (from Task 1, via `rutster-media`).
- Produces: `run_tap_client` + `handle_brain_frame` accept `advisory_tx: &mpsc::Sender<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:

```rust
    /// slice-4: `speech_started`/`speech_stopped` are now forwarded to the
    /// dedicated `advisory_tx` side-channel (for the Reflex to drain), and
    /// STILL NOT forwarded to the function_call channel (different bus).
    #[tokio::test]
    async fn advisory_events_forwarded_to_advisory_channel_only() {
        let (tx_fc, mut rx_fc) = mpsc::channel::<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):

```rust
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:

```rust
        // slice-4: advisory events forward to the Reflex via the dedicated
        // `advisory_tx` channel (NonBlocking try_send — the media thread
        // drains on its 20ms tick). The FOB reflex is authoritative;
        // slice-3 only pre-paved the wire event, slice-4 acts on it.
        DecodedPayload::SpeechStarted => {
            let ev = rutster_media::AdvisoryEvent::SpeechStarted { at: Instant::now() };
            if tx_advisory.try_send(ev).is_err() {
                // Channel full → drop + observe (hot-path policy).
                // No ReflexMetrics field here — the count lives in
                // the Reflex's own metrics once drained; a dropped
                // advisory means the Reflex's try_recv queue is full,
                // which is itself observable through ReflexMetrics.
                metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed);
                warn!("advisory SpeechStarted dropped (advisory_tx full)");
            }
        }
        DecodedPayload::SpeechStopped => {
            let ev = rutster_media::AdvisoryEvent::SpeechStopped { at: Instant::now() };
            if tx_advisory.try_send(ev).is_err() {
                metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed);
                warn!("advisory SpeechStopped dropped (advisory_tx full)");
            }
        }
```

- [ ] **Step 4: Update `run_tap_client` signature + the `handle_brain_frame` call site**

In `run_tap_client`'s signature (line ~150), add `tx_advisory: mpsc::Sender<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).

```bash
rg 'run_tap_client\(|handle_brain_frame\(' --type rust
```

- [ ] **Step 6: Run the test to verify it passes** — `cargo test -p rutster-tap`.

- [ ] **Step 7: fmt + clippy + full test + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster-tap/src/tap_client.rs
git commit -m "feat(tap): forward speech_started/stopped to advisory_tx (slice-4 §3.1)

The brain's advisory events now flow to the FOB reflex via the dedicated
advisory_tx side-channel (3rd mpsc alongside tx_pcm_in/tx_audio_out).
handle_brain_frame + run_tap_client threads the sender through."
```

---

### Task 5: `spawn_tap_engine` returns `advisory_tx` end + `TapConn.advisory_tx`

**Files:**
- Modify: `crates/rutster/src/tap_engine.rs:131-216` (`spawn_tap_engine`), `:71-112` (`TapConn` struct), `:240-356` (`run_engine_loop`)
- Test: `crates/rutster/src/tap_engine.rs` (inline tests)

**Interfaces:**
- Consumes: `AdvisoryEvent` (from Task 1), `run_tap_client`'s new `advisory_tx` param (from Task 4).
- Produces:
  - `TapConn` does NOT carry the advisory channel — the media thread owns it + clones the Sender. Per the Task 5 revision note below, `spawn_tap_engine` takes `advisory_tx: mpsc::Sender<AdvisoryEvent>` as a PARAMETER (the media thread constructs the channel, clones the `Sender` for both `spawn_tap_engine` AND `LocalVadReflex::new`, hands the `Receiver` to `Reflex::new`). This is because there are TWO senders (the brain path via the engine + the local VAD via the wrapper), so the channel ownership lives at the composition site in Task 6, not in `spawn_tap_engine`.
  - `spawn_tap_engine` takes `advisory_tx: mpsc::Sender<AdvisoryEvent>` as a PARAMETER (the media thread owns the channel + clones it; `tokio::sync::mpsc::Sender` is `Clone`). Returns the 2-tuple `(TapAudioPipe, TapConn)` (legacy shape). The media thread constructs the `(advisory_tx, advisory_rx)` pair after Task 5's revision — `advisory_tx` cloned into both `spawn_tap_engine` AND `LocalVadReflex::new`; `advisory_rx` handed to `Reflex::new`. Replaces the Task 5 draft's "3-tuple return" approach: the media thread owns the channel because it has TWO senders (the brain path + the local VAD path), so ownership lives at the composition site (Task 6), not in `spawn_tap_engine` (which is just one of the senders).

**Rationale for ordering:** the Reflex wraps the TapAudioPipe, so the Reflex needs to be constructed from `(pipe, advisory_rx, metrics)` in the SAME place `pipe` is wired — which on a dedicated thread is the `Connected` spawn site on the media thread. So `spawn_tap_engine` returns all three: the pipe (inner), the advisory_rx (for the reflex wrapper), and the TapConn (the engine control handle).

- [ ] **Step 1: Write the failing test**

Append to `crates/rutster/src/tap_engine.rs` test module:

```rust
    /// slice-4: spawn_tap_engine takes advisory_tx as a parameter (the media
    /// thread owns the channel — TWO senders: the engine + the local VAD).
    #[tokio::test]
    async fn spawn_accepts_advisory_tx_parameter() {
        let id = ChannelId::new();
        let url = Url::parse("ws://127.0.0.1:1/echo").unwrap();
        let (advisory_tx, _advisory_rx) =
            mpsc::channel::<rutster_media::AdvisoryEvent>(16);
        let (_pipe, conn) = spawn_tap_engine(
            id, url, crate::session_map::AppState::default(), advisory_tx,
        );
        let _ = conn.close_tx.send(());
        conn.join.abort();
    }
```

- [ ] **Step 2: Run the test to verify it fails** — `spawn_tap_engine` currently returns a 2-tuple.

- [ ] **Step 3: Update `spawn_tap_engine` + `TapConn` + `run_engine_loop`**

In `tap_engine.rs`:

1. In `spawn_tap_engine` (~line 131-216): add the `advisory` channel:
   ```rust
   let (tx_advisory, advisory_rx) = mpsc::channel::<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(session_id, tap_url, app_state, advisory_tx: mpsc::Sender<rutster_media::AdvisoryEvent>) -> (TapAudioPipe, TapConn)` — the media thread owns the channel; this is ONE of two senders (the other is `LocalVadReflex`'s own advisory_tx clone).

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

- [ ] **Step 4: Update ALL existing call sites of `spawn_tap_engine`**

```bash
rg 'spawn_tap_engine\(' --type rust
```

Each caller (the existing one is `session_map.rs::drive_all_sessions` — which itself is being relocated to `media_thread.rs` in Task 6) now passes an `advisory_tx` Sender as the 4th arg. The `LocalVadReflex` clones the same Sender in Task 6's composition site.

- [ ] **Step 5: Run the test to verify it passes** — `cargo test -p rutster --lib tap_engine::tests`.

- [ ] **Step 6: fmt + clippy + full test + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster/src/tap_engine.rs
git commit -s -m "feat(binary): spawn_tap_engine accepts advisory_tx (slice-4 §3.1)

The media thread owns the advisory channel (two senders: the engine +
the local VAD). spawn_tap_engine takes advisory_tx as a parameter +
forwards brain speech_started/speech_stopped through it."
```

---

### Task 6: `MediaThread` — the dedicated std::thread owning all RtcSessions

**Files:**
- Create: `crates/rutster/src/media_thread.rs`
- Modify: `crates/rutster/src/lib.rs` (add `pub mod media_thread;`)
- Test: `crates/rutster/src/media_thread.rs` (inline tests)

**Interfaces:**
- Consumes: `RtcSession` + `RtcSessionError` (from `rutster-media`), `ChannelId` (from `rutster-call-model`), `spawn_tap_engine` (from Task 5), `Reflex<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:

```rust
//! # MediaThread — the dedicated 20ms media loop on a std::thread
//! (slice-4 spec §2.2, §4)
//!
//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
//! tokio pool." slice-1 ran the poll on tokio as an acknowledged
//! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary
//! startup owns `HashMap<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();
                    // The media thread owns the advisory channel (multi-producer:
                    // tokio::sync::mpsc::Sender is Clone). Both the brain's
                    // advisories (via spawn_tap_engine) AND the local VAD's
                    // trips (via LocalVadReflex) push to the SAME mpsc; the
                    // Reflex drains both uniformly. This means spawn_tap_engine
                    // takes advisory_tx as a PARAMETER (Task 5's signature
                    // changes: spawn_tap_engine(session_id, tap_url, app_state,
                    // advisory_tx) — see Task 5's revision note).
                    let (advisory_tx, advisory_rx) =
                        mpsc::channel::<rutster_media::AdvisoryEvent>(16);
                    let (pipe, conn) = tokio_handle.block_on(async {
                        spawn_tap_engine(*id, url, crate::session_map::AppState::default(), advisory_tx.clone())
                    });
                    let metrics = rutster_media::ReflexMetrics::new();
                    // Compose: Reflex<TapAudioPipe> (state machine) wrapped by
                    // LocalVadReflex (primary VAD trigger). Both feed advisory_tx.
                    let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics);
                    let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx);
                    session.rtc.set_pipe(vad);
                    session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
                    session.tap_conn = Some(conn);
                    info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
                    continue;
                }
            }

            if session.rtc.is_closed() {
                closed_ids.push(*id);
            }
        }
        for id in closed_ids {
            sessions.remove(&id);
            debug!(channel_id = %id, "session evicted after close");
        }

        // === Step 3: sleep META_TICK. ===
        std::thread::sleep(META_TICK);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn media_thread_register_and_shutdown_round_trips() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let thread = MediaThread::spawn(url, handle);
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let id = rx.blocking_recv().expect("register reply").expect("session");
        assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId");
        thread.shutdown();
    }
}
```

- [ ] **Step 2: Run the test to verify it fails**

Run: `cargo test -p rutster --lib media_thread::tests`
Expected: FAIL — module not yet declared in `lib.rs`.

- [ ] **Step 3: Declare the module in `lib.rs`**

In `crates/rutster/src/lib.rs`, add:

```rust
pub mod media_thread;
```

- [ ] **Step 4: Run the test to verify it passes**

Run: `cargo test -p rutster --lib media_thread::tests`
Expected: PASS (1 test). Note: the test uses `tokio_handle.block_on(async { spawn_tap_engine(...) })` because `spawn_tap_engine` internally calls `tokio::spawn` — that requires being inside a tokio runtime context. The block_on is cold-path, runs only on the `Connected` transition.

- [ ] **Step 5: fmt + clippy + full test + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster/src/media_thread.rs crates/rutster/src/lib.rs
git commit -m "feat(binary): MediaThread — dedicated std::thread for the 20ms loop (slice-4 §4)

ARCHITECTURE.md mandate (\"never the shared tokio pool\") finally landed.
One std::thread owns all RtcSessions exclusively; axum routes via command
channel (Register/AcceptOffer/Delete/Shutdown). The Reflex<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`).

```rust
// 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.

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

- [ ] **Step 4: Update `main.rs`**

```rust
// In main.rs:
let state = AppState::new(default_tap_url);
let media_thread = state.clone().spawn_media_thread(tokio::runtime::Handle::current()).await;
// ... after axum::serve completes:
media_thread.shutdown();
```

- [ ] **Step 5: Run all tests**

```bash
cargo test --all
```
Expected: PASS, including the existing `api_integration.rs` end-to-end test. If it fails, the regression is in the route wiring — fix before committing.

- [ ] **Step 6: fmt + clippy + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
git add crates/rutster/src/session_map.rs crates/rutster/src/main.rs crates/rutster/src/routes.rs
git commit -m "feat(binary): rewire session_map + routes to MediaThread command channel (slice-4 §4.3)

AppState now holds cmd_tx: Sender<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:

```rust
    /// slice-4: MockRealtimeBrain can emit `speech_started`/`speech_stopped`
    /// on a programmable schedule, simulating the brain's VAD firing. This
    /// is what the slice-4 barge-in e2e test drives.
    #[tokio::test]
    async fn emits_speech_started_on_schedule_after_n_audio_in_frames() {
        use std::sync::{Arc, Mutex};
        let mut mock = MockRealtimeBrain::start().await.unwrap();
        mock.set_advisory_schedule(vec![
            AdvisoryTrigger { after_audio_in_frames: 2, event: AdvisoryKind::SpeechStarted },
            AdvisoryTrigger { after_audio_in_frames: 4, event: AdvisoryKind::SpeechStopped },
        ]);
        let url = mock.url();
        let req = url.as_str().into_client_request().unwrap();
        let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();
        // Send session.update first (the mock's contract).
        let session_update = json!({
            "type": "session.update",
            "session": { "turn_detection": null }
        });
        ws.send(Message::Text(session_update.to_string())).await.unwrap();
        // Send 2 audio_in appends → expect a speech_started.
        for _ in 0..2 {
            let append = json!({ "type": "input_audio_buffer.append", "audio": "AAAA" });
            ws.send(Message::Text(append.to_string())).await.unwrap();
        }
        // Skip the canned response.audio.delta replies; wait for the speech_started.
        let mut saw_started = false;
        for _ in 0..10 {
            let msg = tokio::time::timeout(Duration::from_millis(500), ws.next())
                .await
                .expect("event within 500ms")
                .unwrap()
                .unwrap();
            let text = msg.into_text().unwrap();
            if text.contains("speech_started") {
                saw_started = true;
                break;
            }
        }
        assert!(saw_started, "mock must emit speech_started after N appends");
    }
```

- [ ] **Step 2: Run the test to verify it fails** — `set_advisory_schedule` + types don't exist on `MockRealtimeBrain` yet.

- [ ] **Step 3: Add the advisory schedule API to `MockRealtimeBrain`**

Add to the `MockRealtimeBrain` struct + impl (and `handle_connection`):

```rust
/// A trigger for the advisory schedule. The mock counts
/// `input_audio_buffer.append` events; when the count reaches
/// `after_audio_in_frames`, it emits `event` unprompted (simulating
/// the brain's VAD firing).
#[derive(Debug, Clone)]
pub struct AdvisoryTrigger {
    pub after_audio_in_frames: u32,
    pub event: AdvisoryKind,
}

#[derive(Debug, Clone, Copy)]
pub enum AdvisoryKind {
    SpeechStarted,
    SpeechStopped,
}

impl MockRealtimeBrain {
    /// Set a schedule of advisory events the mock emits UNPROMPTED after
    /// observing N `input_audio_buffer.append` events. Used by the slice-4
    /// barge-in e2e test to drive the reflex.
    pub fn set_advisory_schedule(&mut self, schedule: Vec<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 (PRIMARY + SECONDARY paths)

**Files:**
- Create: `crates/rutster/tests/barge_in_integration.rs`
- Consumes: `MockRealtimeBrain` (Task 8), `MediaThread` (Task 6), `spawn_tap_engine` (Task 5), `Reflex<TapAudioPipe>` (Task 2) + `LocalVadReflex<Reflex<TapAudioPipe>>` (Task 2b).

**Two test cases (both required — they prove different properties):**

- [ ] **Step 1: Write the PRIMARY-path e2e test (proves wedge #1 — kill WITHOUT brain)**

Sets up `MediaThread` + a `LocalVadReflex<Reflex<TapAudioPipe>>` stack driven by a synthetic caller
frame source + a synthetic brain `audio_out` source (no `MockRealtimeBrain` — the brain is NOT
required for the kill on this path). Asserts:
- Push N (= `VAD_DEBOUNCE_FRAMES` = 3) loud caller frames (samples = 1000) into the sink path
  → the local VAD trips → `Reflex::next_pcm_frame()` returns `None` even with a frame queued
  in the ring (the barge flushed it). **NO brain advisory was sent.**
- A fresh `PcmFrame` on the brain's `audio_out` source → next `next_pcm_frame()` returns `Some`
  (resume).
- `ReflexMetrics.barge_in_count` is 1 (the kill fired); the brain's advisory channel was empty
  (proving the kill didn't depend on the brain).

```rust
// crates/rutster/tests/barge_in_integration.rs
use rutster_media::{AudioPipe, AudioSource, AudioSink, PcmFrame, LOCAL_VAD_THRESHOLD, VAD_DEBOUNCE_FRAMES};
use rutster_media::{AdvisoryEvent, LocalVadReflex, Reflex, ReflexMetrics};
use rutster_tap::{TapAudioPipe, TapMetrics};
use tokio::sync::mpsc;

#[tokio::test]
async fn primary_path_local_vad_kills_playout_without_brain() {
    let (tx_pcm_in, _rx_pcm_in) = mpsc::channel(32);
    let (tx_audio_out, rx_audio_out) = mpsc::channel(32);
    let tap_metrics = std::sync::Arc::new(TapMetrics::new());
    let pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);

    let (advisory_tx, advisory_rx) = mpsc::channel(16);
    let reflex_metrics = ReflexMetrics::new();
    let reflex = Reflex::new(pipe, advisory_rx, reflex_metrics.clone());
    let mut stack = LocalVadReflex::new(reflex, advisory_tx);

    // Pre-load a brain audio_out frame into the ring (drain via next_pcm_frame).
    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
    let _ = stack.next_pcm_frame(); // drain into ring
    stack.next_pcm_frame(); // pop one (brain is playing)

    // Loud caller audio × N → local VAD trips.
    let mut loud = PcmFrame::zeroed();
    for s in loud.samples.iter_mut() { *s = 1000; }
    for _ in 0..VAD_DEBOUNCE_FRAMES {
        stack.on_pcm_frame(loud.clone());
    }

    // Next playout tick: kill applied, None returned (ring cleared).
    let f = stack.next_pcm_frame();
    assert!(f.is_none(), "primary-path local VAD must kill playout within 1 tick");
    assert_eq!(reflex_metrics.barge_in_count.load(std::sync::atomic::Ordering::Relaxed), 1);

    // Fresh brain audio_out → resume.
    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
    let resumed = stack.next_pcm_frame();
    assert!(resumed.is_some(), "first fresh audio_out post-barge must resume playout");
}
```

- [ ] **Step 2: Write the SECONDARY-path e2e test (exercises slice-3's advisory plumbing)**

Sets up `MockRealtimeBrain` (with advisory schedule emitting `speech_started` after 2 audio_in
appends) + `MediaThread` + the same stack. Pushes QUIET caller audio (samples = 0, sub-threshold)
so the local VAD does NOT trip; the brain's `speech_started` advisory IS the trigger. Asserts:
- `speech_started` arrives → kill applied within ≤1 tick.
- Fresh `audio_out` → resume.

Proves the secondary path still works (the brain's ASR-VAD is the backstop when local VAD
doesn't fire — e.g. quiet callers, or as confirmation of a local kill).

- [ ] **Step 3: Run both e2e tests + iterate** — the PRIMARY-path test is the proof of the slice (wedge #1).

- [ ] **Step 4: fmt + clippy + commit**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
git add crates/rutster/tests/barge_in_integration.rs
git commit -s -m "test(slice-4): barge-in e2e — primary (local VAD, no brain) + secondary (advisory)

The PRIMARY-path test proves wedge #1: loud caller audio → kill within 4
ticks (≤80ms wallclock), WITHOUT any brain advisory. The SECONDARY-path
test exercises slice-3's advisory plumbing as the confirmation/backstop
path. Both feed the same advisory mpsc; the Reflex drains both."
```

---

### Task 10: CI seam gate + verification

**Files:**
- Modify: `.github/workflows/ci.yml` (add the seam-diff step)

- [ ] **Step 1: Add the seam-diff CI step**

In `.github/workflows/ci.yml`, add a job step (after `cargo test --all`):

```yaml
      - name: Seam gate — loop_driver + rtc_session byte-identical to slice-3
        run: |
          git fetch origin main --depth=1
          # Check against the PRE-slice-3 main, OR pin to the slice-3 merge SHA.
          # For now: assert no diff in these files between this branch + main.
          git diff --exit-code origin/main -- \
            crates/rutster-media/src/loop_driver.rs \
            crates/rutster-media/src/rtc_session.rs
```

- [ ] **Step 2: Final fmt + clippy + test + deny sweep**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo test --all
cargo deny check
```

- [ ] **Step 3: Commit + tag the slice-4-e2e-green state** with `test(slice-4):` + the §7 done-criteria checklist.

---

## Self-Review Notes

- **Spec coverage:**
  - §1.1 In scope: Reflex (Task 2), barge_in_flush (Task 1+3), AdvisoryEvent (Task 1), MediaThread (Task 6), session_map rewire (Task 7), MockRealtimeBrain extension (Task 8), barge e2e (Task 9), learner-facing comments (per-task, verified by `cargo doc` at Task 10).
  - §1.2 Out of scope: VAD tuning framework deferred (the VAD itself is in scope); other items unchanged.
  - §7 Done-criteria: items 1-12 all mapped (Tasks 1, 2, 2b, 3, 4, 5, 6, 7, 8, 9, 10).
  - §8 Open decisions: pinned in respective tasks (mock API in Task 8, teardown ordering in Task 6).
- **Placeholder scan:** none found.
- **Type consistency:** `AdvisoryEvent` consistent (Task 1 → 2 → 2b → 4 → 5 → 8). `Reflex<P>` signature consistent (Task 2 → 6). `LocalVadReflex<P>` consistent (Task 2b → 6). `MediaCmd` consistent (Task 6 → 7). `spawn_tap_engine` takes `advisory_tx` as a parameter (Task 5 revision note); Task 6's call site clones the Sender.
- **Seam gate:** `loop_driver.rs` + `rtc_session.rs` are NOT in any task's Modify list. The invariant is preserved by construction.
- **Wedge-#1 audit (2026-07-01 review revision):** the PRIMARY-path e2e test (Task 9 step 1) proves the kill fires WITHOUT any brain advisory — local VAD in `on_pcm_frame` is the trigger source. The kill decision is in-core on the dedicated thread, in the 20ms loop, zero brain round-trip. This is the property ARCHITECTURE.md:79-81 demands.
