spec(slice-4): barge-in / VAD-driven playout kill on dedicated media thread
Spearhead step 4 of 6. The FOB reflex loop: on brain speech_started advisory, kill playout from the core-authoritative buffer; resume on first fresh audio_out post-barge. Graduates the 20ms media loop off the tokio pool onto a dedicated std::thread (ARCHITECTURE.md mandate, deferred from slice-1 loop_driver.rs:18-23). Reflex<P: AudioPipe> wrapper in rutster-media (FOB) decorates the pipe — loop_driver.rs + rtc_session.rs stay byte-identical (the §8.5 #6 seam gate, restated). Decisions (brainstorming 2026-07-01): - Trigger: advisory-only (brain speech_started/stopped); local VAD deferred - Resume: first fresh audio_out; SpeechStopped observational only - Thread: single dedicated std::thread; per-session/threadpool deferred - Approach C: B's thread model + reflex.rs module (per user selection) Out-of-scope (§1.2): local VAD, per-session threads, min-mute floor, brain-side input_audio_buffer.interrupt, TLS, authn, spend cap. Depends on: slice-1 (media loop + seam), slice-2 (tap + playout buffer), slice-3 (brain + speech_started/stopped advisory, S4 turn-ownership lock).
This commit is contained in:
opencode controller
613
docs/superpowers/specs/2026-07-01-slice-4-barge-in-design.md
Normal file
613
docs/superpowers/specs/2026-07-01-slice-4-barge-in-design.md
Normal file
@@ -0,0 +1,613 @@
|
|||||||
|
# Rutster slice 4 — Barge-in: VAD-driven playout kill on a dedicated media thread
|
||||||
|
|
||||||
|
- **Status:** Draft (pending review)
|
||||||
|
- **Date:** 2026-07-01
|
||||||
|
- **Spearhead step:** 4 of 6 (vision-revision §10 / PORT_PLAN "Phasing")
|
||||||
|
- **Origin:** brainstorming session 2026-07-01
|
||||||
|
- **Depends on:** [slice 1 — WebRTC media loopback](2026-06-28-slice-1-webrtc-loopback-design.md),
|
||||||
|
[slice 2 — The agent tap](2026-06-28-slice-2-agent-tap-design.md), and slice-3's
|
||||||
|
OpenAI Realtime brain (merged as `c30a452` — `MockRealtimeBrain` + translator + the
|
||||||
|
`speech_started` / `speech_stopped` advisory events). All three must be landed and green.
|
||||||
|
- **Related:** [ADR-0002](../../adr/0002-north-star-and-fused-core.md) (fused vertical — the
|
||||||
|
hot-path hop invariant this slice re-affirms), [ADR-0008](../../adr/0008-fob-and-green-zone.md)
|
||||||
|
(FOB/green-zone doctrine — the reflex is a FOB member: hot-path, differentiating),
|
||||||
|
[ARCHITECTURE.md §"Biggest technical risk"](../../ARCHITECTURE.md) (the reflex loop *is*
|
||||||
|
the remaining long pole), [ARCHITECTURE.md §"Media plane"](../../ARCHITECTURE.md)
|
||||||
|
("Dedicated timing threads for the 20ms loop, **never the shared tokio pool**" — this
|
||||||
|
slice finally lands that mandate).
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## TL;DR
|
||||||
|
|
||||||
|
Stand up spearhead step 4: the **FOB reflex loop**. Slice 3 pre-paved the advisory signals
|
||||||
|
(`speech_started` / `speech_stopped` from the brain) and locked the turn-ownership decision
|
||||||
|
(OpenAI Realtime server-side VAD disabled; the FOB owns turn-taking). Slice 4 **acts** on
|
||||||
|
those advisories: on `speech_started`, the FOB kills playout from the core-authoritative
|
||||||
|
buffer; playout resumes when the brain's *first fresh `audio_out` after the barge* arrives,
|
||||||
|
proving the brain has yielded and started a new response. No brain round-trip gates the
|
||||||
|
kill — the decision lives in the 20 ms media loop.
|
||||||
|
|
||||||
|
Slice 4 also **graduates the media loop off the tokio pool**: a single dedicated `std::thread`
|
||||||
|
owns all `RtcSession`s exclusively and drives the 20 ms tick via `std::thread::sleep`. This
|
||||||
|
honors ARCHITECTURE.md's "never the shared tokio pool" mandate, which slice-1 explicitly
|
||||||
|
deferred to "step 4 (barge-in)" (`loop_driver.rs:18-23`). The graduation is load-bearing:
|
||||||
|
the reflex is the differentiator and the long pole, and its timing discipline demands a
|
||||||
|
thread that doesn't compete with the axum runtime for scheduling.
|
||||||
|
|
||||||
|
The **seam slice 1→3 preserved** (`loop_driver.rs` + `rtc_session.rs` byte-identical) holds
|
||||||
|
for slice-4 as well: the reflex is a `Reflex<P>` wrapper decorating the `AudioPipe` trait,
|
||||||
|
invisibly to `loop_driver::drive`. Only the binary-side wiring (`session_map.rs` →
|
||||||
|
`media_thread.rs`) changes shape; the media crate's hot path stays untouched.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 1. Scope
|
||||||
|
|
||||||
|
### 1.1 In scope
|
||||||
|
|
||||||
|
- Implementation of spearhead step 4: **barge-in / VAD-driven playout kill**, where "VAD"
|
||||||
|
for the MVP is the brain's `speech_started` / `speech_stopped` advisory (slice-3 pre-paved)
|
||||||
|
— no local DSP energy detector this slice (deferred).
|
||||||
|
- A **new `Reflex<P: AudioPipe>` wrapper** (`rutster-media/src/reflex.rs`) that decorates the
|
||||||
|
pipe the `RtcSession` holds. The reflex owns the mute state machine, the advisory channel
|
||||||
|
receiver, and the barge-in flush trigger. It is the concrete embodiment of ARCHITECTURE.md's
|
||||||
|
"local real-time reflexes" row for the barge-in case.
|
||||||
|
- A **new `barge_in_flush` method on the `AudioPipe` trait** (default impl delegates to
|
||||||
|
`clear_playout_ring`) — the seam object's "kill now" path: clear the playout ring AND drain
|
||||||
|
the brain-bound `rx_audio_out` channel of any frames queued before the barge so the first
|
||||||
|
`audio_out` observed post-barge is provably post-barge. `TapAudioPipe` overrides;
|
||||||
|
`EchoAudioPipe` uses the default.
|
||||||
|
- A **new `AdvisoryEvent` enum** (`SpeechStarted { at }`, `SpeechStopped { at }`) flowing
|
||||||
|
over a tokio mpsc from the TapEngine (tokio) to the Reflex (media thread). The engine
|
||||||
|
pushes the events it already decodes from the brain (slice-3 wired these as log+count;
|
||||||
|
slice-4 *forwards* them into the reflex).
|
||||||
|
- A **new dedicated media thread** (`rutster/src/media_thread.rs`) replacing the tokio
|
||||||
|
`spawn_poll_task`. One `std::thread::spawn` at binary startup owns
|
||||||
|
`HashMap<ChannelId, RtcSession>` exclusively; all access from axum is via a command
|
||||||
|
channel (`AcceptOffer`, `Delete`, `Shutdown`). The 20 ms tick is `std::thread::sleep`.
|
||||||
|
- **Rewired `session_map.rs`** (binary): `SessionEntry.rtc: Arc<Mutex<RtcSession>>` →
|
||||||
|
`cmd_tx: mpsc::Sender<MediaCmd>`. `create_session`, `post_offer`, `close`,
|
||||||
|
`spawn_poll_task` all route through the command channel. The async handlers are cold-path;
|
||||||
|
no cross-thread coordination happens on the 20 ms tick.
|
||||||
|
- **`MockRealtimeBrain` extension** (`rutster-brain-realtime/src/mock.rs`): gains the ability
|
||||||
|
to emit `speech_started` / `speech_stopped` on a programmable schedule (e.g. "after N
|
||||||
|
audio_in frames received, send `speech_started`; after M more, send `speech_stopped`").
|
||||||
|
- **Barge-in e2e integration test** (extends slice-3's
|
||||||
|
`crates/rutster/tests/realtime_integration.rs` harness): synthetic WebRTC peer →
|
||||||
|
MediaThread → TapEngine → MockRealtimeBrain; mock emits `speech_started`; assert playout
|
||||||
|
goes silent within ≤1 tick (20 ms); mock emits fresh `audio_out`; assert playout resumes.
|
||||||
|
- New `ReflexMetrics` (`barge_in_count`, `advisory_dropped`, `frames_suppressed`) mirroring
|
||||||
|
`TapMetrics` shape (atomics, snapshot fn). Threaded through the same `TapConn.metrics`
|
||||||
|
surface where reasonable, or a new side-car.
|
||||||
|
- Thorough learner-facing comments on the new std-thread / channel-bridge / wrapper-decorator
|
||||||
|
patterns (slice-1 §7 standard carries over).
|
||||||
|
|
||||||
|
### 1.2 Out of scope (with scheduled return)
|
||||||
|
|
||||||
|
| Deferred item | Returns in | Why deferred |
|
||||||
|
|---|---|---|
|
||||||
|
| Local VAD (energy/RMS detector in `on_pcm_frame`) | post-spearhead refinement | Advisory-only MVP per slice-4 brainstorming decision. Local VAD needs threshold tuning + DSP analysis worth its own slice; the `Reflex<P>` wrapper shape is designed so a local-VAD decorator composes as a second wrapper outside (or inside) the advisory one. |
|
||||||
|
| Per-session media threads / threadpool shard | later rung | Single thread covers spearhead scale (loopback dev + low-concurrency PSTN via slice-5). The command-channel seam between axum and the thread makes the graduation to a threadpool shard localized. |
|
||||||
|
| Trickle ICE | later | Unchanged from slice-1 deferral. |
|
||||||
|
| Min-mute floor / inter-word-gap debouncing | post-spearhead | `SpeechStopped` is a no-op for mute; a floor timer on resume would protect against brain-yield races (brain emits fresh `audio_out` before the caller's inter-word gap ends). Defer until observed in practice. |
|
||||||
|
| Brain-side `input_audio_buffer.interrupt` / `clear` on barge | slice-5 or brain-side | Whether the brain should clear its own input buffer on `speech_started` is a brain-UX decision, not a FOB one; the FOB only kills *playout* (its half-duplex gate). The advisory already tells the brain what happened; the brain's response is its own concern. |
|
||||||
|
| Half-duplex gating beyond playout kill | later rung | Barge-in is the first half-duplex reflex; full HD gating (mixing, jitter buffer interaction, multi-party) arrives with conferencing. |
|
||||||
|
| TLS on HTTP / WSS | slice-5 | Unchanged. |
|
||||||
|
| Authn / authz / multi-tenancy | slice-6 | Unchanged. |
|
||||||
|
| Spend cap / abuse gate | slice-6 | Unchanged. |
|
||||||
|
| Browser-based automated e2e (Playwright/Selenium) | post-spearhead | Unchanged. The synthetic-peer harness from slice-2/3 is the test vehicle. |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 2. Architecture delta
|
||||||
|
|
||||||
|
### 2.1 The reflex wrapper
|
||||||
|
|
||||||
|
`Reflex<P: AudioPipe>` is a zero-cost-style decorator around any `AudioPipe`. It sits
|
||||||
|
between `RtcSession.pipe` (which `loop_driver::drive` calls via `session.pipe.next_pcm_frame()`)
|
||||||
|
and the concrete pipe (`TapAudioPipe` in production, `EchoAudioPipe` in slice-1's unit tests).
|
||||||
|
`loop_driver` is oblivious to the wrapper: it still calls `session.pipe.next_pcm_frame()`,
|
||||||
|
the dynamic dispatch through `Box<dyn AudioPipe>` lands in `Reflex::next_pcm_frame`, which
|
||||||
|
applies the state machine and delegates to `inner.next_pcm_frame()` per the table in §3.2.
|
||||||
|
|
||||||
|
The reflex owns three pieces of state:
|
||||||
|
- `advisory_rx: mpsc::Receiver<AdvisoryEvent>` — drained sync-non-blocking via `try_recv`
|
||||||
|
on the 20 ms tick before delegating to `inner`. Fed by the TapEngine task over tokio mpsc.
|
||||||
|
- `muted: bool` — the kill state. `next_pcm_frame` returns `None` while muted, *unless* the
|
||||||
|
inner returns `Some` (the resume condition — the first fresh `audio_out` clears mute).
|
||||||
|
- `barge_epoch: u64` — incremented on every `SpeechStarted`. Not strictly required for the
|
||||||
|
advisory-only MVP (the flush + drain makes the resume race-free), but it's the seam for a
|
||||||
|
future local-VAD wrapper that could race the advisory. Documented as forward-compatible.
|
||||||
|
|
||||||
|
### 2.2 The dedicated media thread
|
||||||
|
|
||||||
|
A single `std::thread::spawn` replaces the tokio `spawn_poll_task`. The thread owns
|
||||||
|
`HashMap<ChannelId, RtcSession>` **exclusively** — no `Arc<Mutex<RtcSession>>` shared with
|
||||||
|
axum. All access from the axum handlers is via a command channel:
|
||||||
|
|
||||||
|
```rust
|
||||||
|
enum MediaCmd {
|
||||||
|
AcceptOffer { id: ChannelId, sdp: String, reply: oneshot::Sender<Result<String, String>> },
|
||||||
|
Delete { id: ChannelId, reply: oneshot::Sender<()> },
|
||||||
|
Shutdown { reply: oneshot::Sender<()> },
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
The thread loop per 10 ms meta-tick:
|
||||||
|
1. Drain `cmd_rx` via `try_recv` loop — handle all pending commands before ticking.
|
||||||
|
2. For each session in the map: drain the per-session `flush_rx` side-channel (slice-2's
|
||||||
|
existing disconnect-flush signal) BEFORE `run_poll_once`, then call
|
||||||
|
`RtcSession::run_poll_once(now)` (the unchanged `loop_driver::drive`).
|
||||||
|
3. After `run_poll_once`, observe `channel.state`:
|
||||||
|
- `Connected && tap.is_none()` → spawn the TapEngine (tokio task via the
|
||||||
|
`tokio::runtime::Handle` captured at thread-start) + wire
|
||||||
|
`Reflex<TapAudioPipe>` as the session's pipe. Mirror of slice-2's spawn seam,
|
||||||
|
relocated from `session_map.rs::drive_all_sessions` to here.
|
||||||
|
- `Closed` → remove the entry + drop the session.
|
||||||
|
4. `std::thread::sleep(Duration::from_millis(10))` — 10 ms meta-tick. (Stable API:
|
||||||
|
`std::thread::sleep_until` is nightly-only; `sleep(dur)` is the stable path. The 20 ms
|
||||||
|
outbound encode tick is driven inside `loop_driver::drive` (unchanged); the 10 ms
|
||||||
|
meta-tick gives finer resolution so str0m's `Timeout` outputs are honored promptly.)
|
||||||
|
|
||||||
|
**The tokio ↔ std-thread bridge:** all channels are tokio mpsc/oneshot (constructable on
|
||||||
|
tokio, drainable via `try_recv`/`blocking_recv` from any thread). The `tokio::runtime::Handle`
|
||||||
|
captured at `MediaThread::spawn` time is used on the std thread to `handle.spawn(...)` the
|
||||||
|
TapEngine when the `Connected` transition fires. No async code runs on the std thread itself
|
||||||
|
— only sync channel ops + `RtcSession::run_poll_once`.
|
||||||
|
|
||||||
|
**Why a single thread, not per-session:** spearhead scale. One loopback peer at a time in
|
||||||
|
dev; even at low PSTN concurrency (slice-5) one thread drives dozens of sessions in 10 ms.
|
||||||
|
Per-session threads arrive when the threadpool shard model lands (deferred). The
|
||||||
|
command-channel seam between axum and the thread makes that graduation localized.
|
||||||
|
|
||||||
|
### 2.3 The hot-path audit (ADR-0002 honored)
|
||||||
|
|
||||||
|
ADR-0002's load-bearing rule: *"the control↔media gRPC hop on the per-call hot path is
|
||||||
|
removed."* Slice 4 does not re-introduce a hop:
|
||||||
|
|
||||||
|
- The reflex's kill decision happens **inside** `Reflex::next_pcm_frame` on the dedicated
|
||||||
|
thread — no channel send, no cross-thread coordination on the 20 ms tick. The advisory
|
||||||
|
arrives via a `try_recv` drain (sync, non-blocking).
|
||||||
|
- axum → media-thread is **cold-path only** (SDP accept, DELETE). None of it runs on the
|
||||||
|
20 ms tick.
|
||||||
|
- The brain WS ↔ TapEngine (tokio) path is unchanged from slice-3. The advisory channel
|
||||||
|
is a *third* mpsc alongside the existing `tx_pcm_in`/`rx_audio_out`/`flush_tx` — same
|
||||||
|
pattern, additive.
|
||||||
|
|
||||||
|
**The fused vertical stays fused.** ADR-0002 honored.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 3. Component design
|
||||||
|
|
||||||
|
### 3.1 `AdvisoryEvent` enum
|
||||||
|
|
||||||
|
```rust
|
||||||
|
// crates/rutster-media/src/reflex.rs
|
||||||
|
|
||||||
|
/// A turn-event advisory from the brain. The brain decodes its own
|
||||||
|
/// speech-to-text / VAD results and forwards these; the FOB *owns*
|
||||||
|
/// turn-taking and acts on them (slice-3 §4.3 — OpenAI Realtime
|
||||||
|
/// server-side VAD is DISABLED; the FOB's reflex is authoritative).
|
||||||
|
///
|
||||||
|
/// Carried over a tokio mpsc from the TapEngine (tokio task) to the
|
||||||
|
/// `Reflex` wrapper (media thread). Drained sync via `try_recv` on the
|
||||||
|
/// 20 ms tick — the kill decision lives in the loop, not in a handler.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum AdvisoryEvent {
|
||||||
|
/// The brain detected caller speech. Trigger barge-in: kill playout.
|
||||||
|
SpeechStarted { at: Instant },
|
||||||
|
/// The brain detected caller speech ended. Observed + counted; does
|
||||||
|
/// NOT toggle mute (the resume condition is "first fresh audio_out
|
||||||
|
/// after the barge", not "speech_stopped" — see §3.2 state table).
|
||||||
|
SpeechStopped { at: Instant },
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3.2 `Reflex<P>` state machine
|
||||||
|
|
||||||
|
| Current state | Event | Action | New state |
|
||||||
|
|---|---|---|---|
|
||||||
|
| Playing | `SpeechStarted` | `muted=true`; `epoch++`; `inner.barge_in_flush()` (clear ring + drain `rx_audio_out` so stale brain frames queued pre-barge are dropped); `metrics.barge_in_count++` | Muted |
|
||||||
|
| Muted | `SpeechStarted` (duplicate/re-barge) | `epoch++`; `barge_in_flush()` again (fresh barge resets the "fresh audio" clock); `barge_in_count++` | Muted |
|
||||||
|
| Muted | `SpeechStopped` | increment `advisory_observed_speech_stopped` counter; **no state change** | Muted |
|
||||||
|
| Playing | `SpeechStopped` | increment counter; **no state change** | Playing |
|
||||||
|
| Muted | inner `next_pcm_frame()` returns `Some(f)` (fresh brain audio arrived post-barge) | `muted=false`; return `Some(f)` | Playing |
|
||||||
|
| Muted | inner `next_pcm_frame()` returns `None` | return `None` (silence); `metrics.frames_suppressed++` | Muted |
|
||||||
|
|
||||||
|
**Why `SpeechStopped` is a no-op for mute:** per the resume-semantics decision (resume on
|
||||||
|
first fresh `audio_out`). The brain's `speech_stopped` is *observed* (counter) but doesn't
|
||||||
|
gate — this avoids the inter-word-gap problem (caller pauses, VAD fires stopped, brain
|
||||||
|
un-mutes too early, brain's audio overlaps caller's next word). The resume condition is
|
||||||
|
"the brain has yielded and started a new response," which is provably signaled by the first
|
||||||
|
`audio_out` frame after the barge — not by the caller's silence.
|
||||||
|
|
||||||
|
**Why `epoch`:** not strictly needed for advisory-only (MVP), but it's the seam for the
|
||||||
|
local-VAD backstop (deferred per §1.2). A future `LocalVadReflex` wrapper racing the
|
||||||
|
advisory would need to disambiguate "is this barge a re-barge of the same event or a new
|
||||||
|
one" — the epoch is the disambiguator. Forward-compatible.
|
||||||
|
|
||||||
|
### 3.3 `AudioPipe` trait extension
|
||||||
|
|
||||||
|
```rust
|
||||||
|
// crates/rutster-media/src/pcm.rs — additive method on `AudioPipe`
|
||||||
|
|
||||||
|
/// Barge-in flush: clear the playout ring AND drain the inbound brain
|
||||||
|
/// audio queue of any frames queued before the barge. Called by `Reflex`
|
||||||
|
/// on `SpeechStarted`. The drain of `rx_audio_out` is what makes the
|
||||||
|
/// resume condition race-free: the first `audio_out` observed post-barge
|
||||||
|
/// is provably post-barge (frames queued pre-barge are dropped here).
|
||||||
|
///
|
||||||
|
/// Default impl delegates to `clear_playout_ring` — sufficient for
|
||||||
|
/// pipes without an inbound queue to drain (like `EchoAudioPipe`).
|
||||||
|
fn barge_in_flush(&mut self) {
|
||||||
|
self.clear_playout_ring();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
`TapAudioPipe` overrides:
|
||||||
|
```rust
|
||||||
|
// crates/rutster-tap/src/tap_audio_pipe.rs
|
||||||
|
|
||||||
|
fn barge_in_flush(&mut self) {
|
||||||
|
// Clear the playout ring (drops buffered brain-proposed frames).
|
||||||
|
self.playout_ring.clear();
|
||||||
|
// Drain rx_audio_out of any frames the engine task queued before
|
||||||
|
// the barge. Without this, a stale frame in the mpsc would un-mute
|
||||||
|
// immediately on the next tick — defeating the "first fresh audio_out"
|
||||||
|
// resume condition. Hot-path: try_recv loop, bounded, no blocking.
|
||||||
|
while self.rx_audio_out.try_recv().is_ok() {
|
||||||
|
self.metrics.barge_drained_inflight.fetch_add(1, Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3.4 `Reflex<P>` struct + impl
|
||||||
|
|
||||||
|
```rust
|
||||||
|
// crates/rutster-media/src/reflex.rs
|
||||||
|
|
||||||
|
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> {
|
||||||
|
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 §3.2.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<P: AudioPipe> AudioPipe for Reflex<P> {
|
||||||
|
fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
|
||||||
|
self.drain_advisories();
|
||||||
|
if self.muted {
|
||||||
|
// Muted: pull from inner. Some(f) = fresh brain audio arrived
|
||||||
|
// post-barge → un-mute + return. None = silence, stay muted.
|
||||||
|
match self.inner.next_pcm_frame() {
|
||||||
|
Some(f) => {
|
||||||
|
self.muted = false;
|
||||||
|
Some(f)
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
self.metrics.frames_suppressed.fetch_add(1, Ordering::Relaxed);
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
self.inner.next_pcm_frame()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn on_pcm_frame(&mut self, frame: PcmFrame) {
|
||||||
|
// Inbound caller audio is NEVER gated by the reflex. The brain
|
||||||
|
// still hears the caller during barge — that's the point (the
|
||||||
|
// brain needs to know the caller interrupted; the FOB only kills
|
||||||
|
// its OWN playout, not the caller's path to the brain).
|
||||||
|
self.inner.on_pcm_frame(frame)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn clear_playout_ring(&mut self) {
|
||||||
|
// The reconnect-flush path (slice-2 §5.3) still works through the
|
||||||
|
// wrapper. If it fires during mute, the ring stays empty and mute
|
||||||
|
// clears on the next post-reconnect audio_out.
|
||||||
|
self.inner.clear_playout_ring()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn barge_in_flush(&mut self) {
|
||||||
|
// Allow outer wrappers (future local-VadReflex) to barge the inner.
|
||||||
|
self.inner.barge_in_flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3.5 `ReflexMetrics`
|
||||||
|
|
||||||
|
Mirror of `TapMetrics` shape (atomics + snapshot struct):
|
||||||
|
|
||||||
|
```rust
|
||||||
|
// crates/rutster-media/src/reflex.rs
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
pub struct ReflexMetrics {
|
||||||
|
pub barge_in_count: AtomicU64,
|
||||||
|
pub advisory_dropped: AtomicU64, // advisory channel full (e.g. 16-cap)
|
||||||
|
pub frames_suppressed: AtomicU64, // None returns while muted
|
||||||
|
pub advisory_observed_speech_stopped: AtomicU64,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct ReflexMetricsSnapshot {
|
||||||
|
pub barge_in_count: u64,
|
||||||
|
pub advisory_dropped: u64,
|
||||||
|
pub frames_suppressed: u64,
|
||||||
|
pub advisory_observed_speech_stopped: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
// `barge_drained_inflight` lives on `TapMetrics` (in `rutster-tap`), not
|
||||||
|
// `ReflexMetrics`, because the drain happens inside `TapAudioPipe::barge_in_flush`,
|
||||||
|
// not inside `Reflex`. The path: `Reflex::drain_advisories` calls
|
||||||
|
// `inner.barge_in_flush()` which is `TapAudioPipe::barge_in_flush`, which is
|
||||||
|
// where the `rx_audio_out` drain + the counter increment happen.
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 4. The dedicated media thread
|
||||||
|
|
||||||
|
### 4.1 `MediaThread`
|
||||||
|
|
||||||
|
```rust
|
||||||
|
// crates/rutster/src/media_thread.rs
|
||||||
|
|
||||||
|
pub struct MediaThread {
|
||||||
|
cmd_tx: mpsc::Sender<MediaCmd>,
|
||||||
|
join: Option<std::thread::JoinHandle<()>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
enum MediaCmd {
|
||||||
|
AcceptOffer { id: ChannelId, sdp: String, reply: oneshot::Sender<Result<String, String>> },
|
||||||
|
Delete { id: ChannelId, reply: oneshot::Sender<()> },
|
||||||
|
Shutdown { reply: oneshot::Sender<()> },
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
Spawned at binary startup (`main.rs`), before `axum::serve`. The thread captures a
|
||||||
|
`tokio::runtime::Handle` (to spawn TapEngine tasks when `Connected` transitions fire) and
|
||||||
|
owns `HashMap<ChannelId, RtcSession>` + (per-session, lazily) the TapConn / advisory_rx /
|
||||||
|
Reflex wrapper.
|
||||||
|
|
||||||
|
### 4.2 Thread loop (per 10 ms meta-tick)
|
||||||
|
|
||||||
|
1. `cmd_rx.try_recv()` loop — handle ALL pending commands before ticking. `AcceptOffer`
|
||||||
|
calls `RtcSession::accept_offer(sdp)` and replies via the oneshot. `Delete` fires
|
||||||
|
`close_tx` + bounded-await the engine task (750 ms cap via
|
||||||
|
`tokio::runtime::Handle::block_on(timeout(...))`) — the std thread briefly enters the
|
||||||
|
tokio runtime to await; cold-path, not the 20 ms tick. `Shutdown` drains + replies.
|
||||||
|
2. For each `RtcSession` in the map:
|
||||||
|
- Drain per-session `flush_rx` side-channel (slice-2's existing disconnect-flush) BEFORE
|
||||||
|
`run_poll_once`.
|
||||||
|
- Call `RtcSession::run_poll_once(now)` — the unchanged `loop_driver::drive`.
|
||||||
|
- Observe `channel.state`:
|
||||||
|
- `Connected && tap.is_none()` → `handle.spawn(spawn_tap_engine(...))` to bring up the
|
||||||
|
tokio task; construct `Reflex::new(TapAudioPipe::new(...), advisory_rx, metrics)`;
|
||||||
|
call `RtcSession::set_pipe(reflex)`. Mirror of slice-2's spawn seam.
|
||||||
|
- `Closed` → remove the entry (drops the `RtcSession` + its pipe + advisory ends).
|
||||||
|
3. `std::thread::sleep(Duration::from_millis(10))` — 10 ms meta-tick.
|
||||||
|
|
||||||
|
### 4.3 `session_map.rs` rewire
|
||||||
|
|
||||||
|
`SessionEntry` loses `rtc: Arc<Mutex<RtcSession>>`, gains `cmd_tx: mpsc::Sender<MediaCmd>`
|
||||||
|
(cloned per-entry; cheap). `tap_url` stays (the thread reads it when spawning the engine).
|
||||||
|
`tap_conn: Option<TapConn>` moves onto the media thread (the thread owns it after spawn).
|
||||||
|
|
||||||
|
- `AppState::create_session` → sends a `Register { tap_url, reply }` command to the media
|
||||||
|
thread; the **thread** constructs `RtcSession::new()` (saves a cross-thread move of the
|
||||||
|
struct + keeps all `RtcSession` construction on the thread that owns it). The thread
|
||||||
|
replies with `(id, cmd_tx_for_this_session)`; axum stores `SessionEntry { cmd_tx, tap_url,
|
||||||
|
tap_conn: None }`.
|
||||||
|
- `AppState::get(id)` (SDP path) → `cmd_tx.send(AcceptOffer { ... }).await` + `reply.await`.
|
||||||
|
Cold-path; the axum handler is async.
|
||||||
|
- `AppState::close(id)` → `cmd_tx.send(Delete { id, reply }).await` + `reply.await`. The
|
||||||
|
reply returns after the TapEngine teardown completes on the thread.
|
||||||
|
- `spawn_poll_task` → `spawn_media_thread`: constructs the channels, spawns the std thread,
|
||||||
|
stores `cmd_tx` + `join` in `AppState`. Same idempotent-guard pattern.
|
||||||
|
|
||||||
|
### 4.4 TapEngine extension
|
||||||
|
|
||||||
|
`spawn_tap_engine` returns a third channel end: `advisory_tx: mpsc::Sender<AdvisoryEvent>`.
|
||||||
|
The pump loop, on receiving `speech_started` / `speech_stopped` from the brain (slice-3
|
||||||
|
already decodes these in the tap protocol layer — `protocol_events.rs`), pushes the
|
||||||
|
corresponding `AdvisoryEvent` into `advisory_tx`. If the channel is full, drop + count
|
||||||
|
(hot-path "drop + observe" policy; an advisory is a hint, not a command). The `Reflex`
|
||||||
|
wrapper holds `advisory_rx`.
|
||||||
|
|
||||||
|
### 4.5 `MockRealtimeBrain` extension
|
||||||
|
|
||||||
|
`rutster-brain-realtime/src/mock.rs` gains a programmable advisory schedule: the test can
|
||||||
|
register "after N `audio_in` frames received, send `speech_started`" and "after M more,
|
||||||
|
send `speech_stopped`". The mock already asserts `turn_detection: null` on
|
||||||
|
`session.update` (slice-3's S4 lock); slice-4 keeps that assertion.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 5. Data flow
|
||||||
|
|
||||||
|
### 5.1 Barge-in (the kill)
|
||||||
|
|
||||||
|
```
|
||||||
|
1. caller speaks into mic → peer RTP → str0m decode → on_pcm_frame → tx_pcm_in → TapClient → audio_in (WS) → brain
|
||||||
|
2. brain's VAD fires → brain sends speech_started back over WS (slice-3 already decodes this)
|
||||||
|
3. TapEngine pump loop → push AdvisoryEvent::SpeechStarted → advisory_tx (tokio mpsc, 16-cap)
|
||||||
|
4. media thread 20 ms tick → Reflex::next_pcm_frame → drain_advisories → SpeechStarted seen
|
||||||
|
→ muted=true; epoch++; inner.barge_in_flush() (ring cleared + rx_audio_out drained)
|
||||||
|
→ returns None (silence) for this + subsequent ticks while muted
|
||||||
|
5. loop_driver::drive pulls None from pipe → encodes Opus silence → peer hears silence
|
||||||
|
(the brain's in-flight audio_out frames are dropped; no overlap with caller's speech)
|
||||||
|
```
|
||||||
|
|
||||||
|
### 5.2 Resume (the un-mute)
|
||||||
|
|
||||||
|
```
|
||||||
|
1. brain decides to yield/respond → sends a fresh audio_out frame
|
||||||
|
(provably post-barge: barge_in_flush drained rx_audio_out)
|
||||||
|
2. TapClient → audio_out (WS) → TapEngine → tx_audio_out → rx_audio_out → playout ring
|
||||||
|
3. media thread 20 ms tick → Reflex::next_pcm_frame → drain_advisories (empty)
|
||||||
|
→ muted=true → inner.next_pcm_frame() returns Some(f) (fresh brain audio)
|
||||||
|
→ muted=false; return Some(f)
|
||||||
|
4. loop_driver encodes + writes → peer hears the brain's new response
|
||||||
|
```
|
||||||
|
|
||||||
|
### 5.3 Cold-path (axum ↔ media thread)
|
||||||
|
|
||||||
|
```
|
||||||
|
- POST /v1/sessions → AppState::create_session → MediaCmd::Register → thread constructs RtcSession → reply(id)
|
||||||
|
- POST /v1/sessions/{id}/offer → AppState::get + cmd_tx.send(AcceptOffer) → thread.lock(session).accept_offer(sdp) → reply(answer)
|
||||||
|
- DELETE /v1/sessions/{id} → AppState::close → cmd_tx.send(Delete) → thread: fire close_tx, bounded-await engine task teardown → reply
|
||||||
|
- graceful shutdown → cmd_tx.send(Shutdown) → thread drains + drops → reply → join
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 6. Why these decisions
|
||||||
|
|
||||||
|
### 6.1 Why advisory-only (no local VAD) for the MVP
|
||||||
|
|
||||||
|
- Matches slice-3's S4 turn-ownership posture: OpenAI Realtime's server-side VAD is disabled;
|
||||||
|
the FOB owns turn-taking. The brain already runs VAD (it has to, to do STT); forwarding its
|
||||||
|
result is the cheapest path to a working barge-in.
|
||||||
|
- Local VAD (energy/RMS detector in `on_pcm_frame`) is DSP work + threshold tuning — worth
|
||||||
|
its own slice. The `Reflex<P>` wrapper shape is designed so a `LocalVadReflex` decorator
|
||||||
|
composes outside (or inside) the advisory one when it arrives.
|
||||||
|
- YAGNI: prove the advisory→reflex→kill path end-to-end first; add the backstop if the
|
||||||
|
brain's VAD latency proves insufficient in practice.
|
||||||
|
|
||||||
|
### 6.2 Why resume on first fresh `audio_out` (not `speech_stopped`)
|
||||||
|
|
||||||
|
- The "the brain has yielded and started a new response" condition is provably signaled by
|
||||||
|
the first `audio_out` frame after the barge — not by the caller's silence. `speech_stopped`
|
||||||
|
fires between words; resuming on it un-mutes too early (inter-word-gap overlap).
|
||||||
|
- The `barge_in_flush` drain of `rx_audio_out` makes the resume race-free: the first
|
||||||
|
`audio_out` observed post-barge is provably post-barge (frames queued pre-barge are dropped
|
||||||
|
in the flush).
|
||||||
|
|
||||||
|
### 6.3 Why a single dedicated thread (not per-session)
|
||||||
|
|
||||||
|
- Spearhead scale: one loopback peer in dev; even at low PSTN concurrency (slice-5), one
|
||||||
|
thread drives dozens of sessions in 10 ms.
|
||||||
|
- The command-channel seam between axum and the thread makes the graduation to a threadpool
|
||||||
|
shard localized — when per-CPU-shard threading arrives, it's a fan-out of the
|
||||||
|
`cmd_rx`/`HashMap` shape, not a redesign.
|
||||||
|
- Per-session threads arrive when load demands; the spearhead's "shortest blocking path"
|
||||||
|
rule dislikes spawning work per session that may not need it (pre-ICE-connected sessions
|
||||||
|
would redundantly spin).
|
||||||
|
|
||||||
|
### 6.4 Why `Reflex<P>` as a wrapper (not inline in `TapAudioPipe`)
|
||||||
|
|
||||||
|
- Composition: a future `LocalVadReflex<P>` composes outside the advisory `Reflex<P>`, the
|
||||||
|
same way `Reflex<TapAudioPipe>` composes today. The pattern (decorator over `AudioPipe`) is
|
||||||
|
forward-compatible without restructuring.
|
||||||
|
- The seam: `loop_driver.rs` byte-identical (still calls `pipe.next_pcm_frame()`). If the
|
||||||
|
reflex lived inline in `TapAudioPipe`, the binary-side wiring would still change but the
|
||||||
|
`TapAudioPipe` module itself would grow the reflex state — less isolated.
|
||||||
|
- YAGNI caveated: the wrapper is the right abstraction for advisory-only because there's
|
||||||
|
exactly one reflex. When local VAD arrives, the wrapper pattern pays off; the spec does
|
||||||
|
not pre-empt that by collapsing the wrapper now.
|
||||||
|
|
||||||
|
### 6.5 Why `barge_in_flush` on `AudioPipe` (not just `clear_playout_ring`)
|
||||||
|
|
||||||
|
- `clear_playout_ring` (slice-2) clears the *ring*. `barge_in_flush` clears the ring AND
|
||||||
|
drains the *inbound brain queue* (`rx_audio_out`). The distinction matters: on a brain
|
||||||
|
disconnect (slice-2's case), the brain is gone — `rx_audio_out` will drain itself on the
|
||||||
|
next `Disconnected` `try_recv`. On a barge-in, the brain is alive and may have queued
|
||||||
|
frames pre-barge that would un-mute immediately if not drained here. Two different
|
||||||
|
"clear the playout path" semantics, two methods.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 7. Done-criteria
|
||||||
|
|
||||||
|
1. `cargo test --all` passes (stable + 1.85, the CI matrix).
|
||||||
|
2. `cargo fmt --check` + `cargo clippy -- -D warnings` clean.
|
||||||
|
3. `loop_driver.rs` + `rtc_session.rs` **byte-identical** to slice-3 — CI-asserted via
|
||||||
|
`git diff --exit-code main -- crates/rutster-media/src/loop_driver.rs
|
||||||
|
crates/rutster-media/src/rtc_session.rs` (the §8.5 #6 seam gate, restated for slice-4).
|
||||||
|
4. Dedicated media thread drives sessions off the tokio pool; `MediaThread` integration test
|
||||||
|
passes (AcceptOffer / Delete / Shutdown).
|
||||||
|
5. `Reflex` state-machine unit tests all pass:
|
||||||
|
- `SpeechStarted` → next `next_pcm_frame` returns None even if ring has frames.
|
||||||
|
- `SpeechStarted` then `inner.next_pcm_frame()=Some` → un-mutes, returns the frame.
|
||||||
|
- `SpeechStopped` during Muted → stays Muted.
|
||||||
|
- `SpeechStopped` during Playing → no-op.
|
||||||
|
- Duplicate `SpeechStarted` re-flushes + stays Muted.
|
||||||
|
- Metrics counters (`barge_in_count`, `frames_suppressed`) increment correctly.
|
||||||
|
- `advisory_rx` full → `advisory_dropped` increments, no panic.
|
||||||
|
6. `barge_in_flush` unit tests pass (ring + `rx_audio_out` drain).
|
||||||
|
7. Barge-in e2e: `speech_started` → playout silent within ≤1 tick (20 ms); fresh `audio_out`
|
||||||
|
→ playout resumes. Extends slice-3's `realtime_integration.rs` harness.
|
||||||
|
8. S4 turn-ownership lock preserved: `MockRealtimeBrain` still asserts
|
||||||
|
`turn_detection: null` on `session.update` (slice-3's #7, unchanged).
|
||||||
|
9. `MockRealtimeBrain` extended to emit `speech_started`/`speech_stopped` on schedule.
|
||||||
|
10. `cargo doc --no-deps` renders the new `reflex.rs` + `media_thread.rs` module/item docs
|
||||||
|
cleanly (learner-facing comments present per AGENTS.md code style).
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 8. Open decisions
|
||||||
|
|
||||||
|
- ~~Trigger source~~ — decided: advisory-only (brain `speech_started`/`speech_stopped`).
|
||||||
|
- ~~Resume semantics~~ — decided: first fresh `audio_out` post-barge; `SpeechStopped`
|
||||||
|
observational only.
|
||||||
|
- ~~Thread model~~ — decided: single dedicated `std::thread`; per-session/threadpool deferred.
|
||||||
|
- **`MockRealtimeBrain` advisory schedule API shape** — landed in §4.5 as a programmable
|
||||||
|
"after N audio_in frames" schedule. Could alternatively be a free-form
|
||||||
|
`Vec<(trigger_frame_count, AdvisoryEvent)>` queue. The plan will pin the concrete API.
|
||||||
|
- **Thread shutdown ordering vs TapEngine teardown** — `Delete` command handler fires
|
||||||
|
`close_tx` + bounded-await the engine task (750 ms cap via
|
||||||
|
`tokio::runtime::Handle::block_on(timeout(...))`); the reply oneshot returns after
|
||||||
|
teardown. Cold-path, std thread briefly enters the tokio runtime to await. Documented as
|
||||||
|
an acceptable deviation (not the 20 ms tick).
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 9. Cross-references
|
||||||
|
|
||||||
|
- [slice-1 spec](2026-06-28-slice-1-webrtc-loopback-design.md) — the media loop + the seam
|
||||||
|
(`AudioSource`/`AudioSink` traits in `rutster-media`); slice-1 §8.5 #6 is the seam gate
|
||||||
|
this slice re-affirms.
|
||||||
|
- [slice-2 spec](2026-06-28-slice-2-agent-tap-design.md) — the tap interface, the
|
||||||
|
`TapAudioPipe`, the core-authoritative playout buffer (§4.1), the `flush_tx` side-channel
|
||||||
|
pattern that the `advisory_rx` mirrors.
|
||||||
|
- slice-3 (merged `c30a452`) — `MockRealtimeBrain`, the translator, the
|
||||||
|
`speech_started`/`speech_stopped` protocol events, the S4 turn-ownership lock.
|
||||||
|
- [ADR-0002](../../adr/0002-north-star-and-fused-core.md) — fused vertical; the hot-path
|
||||||
|
hop invariant this slice re-affirms (§2.3 audit).
|
||||||
|
- [ADR-0008](../../adr/0008-fob-and-green-zone.md) — FOB/green-zone doctrine; the reflex is
|
||||||
|
a FOB member (hot-path, security-constitutive for turn-taking, differentiating).
|
||||||
|
- [ARCHITECTURE.md](../../ARCHITECTURE.md) — §"Media plane" ("Dedicated timing threads
|
||||||
|
for the 20ms loop, never the shared tokio pool" — this slice lands it); §"Biggest
|
||||||
|
technical risk" (the reflex loop *is* the remaining long pole).
|
||||||
|
- [PORT_PLAN.md](../../PORT_PLAN.md) — §Phasing, step 4 = barge-in.
|
||||||
Reference in New Issue
Block a user