1930 lines
85 KiB
Markdown
1930 lines
85 KiB
Markdown
# 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.
|