From 488b6ab4237f2ad336ef4e7400d51a22e5dae7a7 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Thu, 2 Jul 2026 23:46:09 -0400 Subject: [PATCH] =?UTF-8?q?feat(media):=20Reflex

=20barge-in=20state=20m?= =?UTF-8?q?achine=20(slice-4=20=C2=A73.2,=20=C2=A73.4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The decorator that instruments any AudioPipe with turn-taking reflexes. SpeechStarted -> muted=true + barge_in_flush + epoch bump. First fresh audio_out -> un-mute. SpeechStopped is observational (no toggle). Inbound audio (on_pcm_frame) is NEVER gated — the brain still hears the caller during barge; the FOB only kills its OWN playout. loop_driver + rtc_session byte-identical (seam holds). barge_epoch is load-bearing THIS slice (spec §6.1, commit 86b7460) — the local VAD (Task 2b) fires ~0 ms after caller speech; the brain's slower ASR advisory fires ~300 ms later on the SAME barge; the epoch disambiguates a fresh re-barge from the late confirmation. Signed-off-by: Aaron D. Lee --- crates/rutster-media/src/lib.rs | 2 +- crates/rutster-media/src/reflex.rs | 292 +++++++++++++++++++++++++++++ 2 files changed, 293 insertions(+), 1 deletion(-) diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index de717dc..f091883 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -40,7 +40,7 @@ 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, ReflexMetrics, ReflexMetricsSnapshot}; // Reflex re-export re-enabled in Task 2 +pub use reflex::{AdvisoryEvent, Reflex, ReflexMetrics, ReflexMetricsSnapshot}; pub use rtc_session::{RtcSession, RtcSessionError}; use thiserror::Error; diff --git a/crates/rutster-media/src/reflex.rs b/crates/rutster-media/src/reflex.rs index 3b79f49..337e115 100644 --- a/crates/rutster-media/src/reflex.rs +++ b/crates/rutster-media/src/reflex.rs @@ -22,6 +22,15 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; +// `Reflex

` consumes advisories from a tokio mpsc drained on the 20 ms +// tick (try_recv, never blocking) + delegates the `AudioPipe` seam to +// `P`. `mpsc` lives in the production type signature, not just tests, so +// the import is module-level — `tokio` stays a runtime dep of the binary +// (which constructs the channel + spawns the TapEngine); rutster-media +// only names the channel's `Receiver` type, no tokio runtime calls. +use crate::pcm::{AudioPipe, AudioSink, AudioSource, PcmFrame}; +use tokio::sync::mpsc; + // Task 2 will reintroduce these when `Reflex

` lands: `Reflex` holds the // `mpsc::Receiver` drained on each 20 ms tick, and the // `P: AudioPipe` generic bound names `AudioPipe`/`AudioSource`/`AudioSink` @@ -86,9 +95,120 @@ pub struct ReflexMetricsSnapshot { pub advisory_observed_speech_stopped: u64, } +/// 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`) +/// +/// 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` 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 { + pub(crate) inner: P, + pub(crate) advisory_rx: mpsc::Receiver, + pub(crate) muted: bool, + // `barge_epoch` is load-bearing THIS slice, not a forward-compat seam: + // the local VAD (Task 2b) fires ~0 ms after caller speech; the brain's + // slower ASR advisory fires ~300 ms later on the SAME barge. The epoch + // disambiguates "a fresh re-barge" from "the late confirmation of the + // barge already in flight" — see slice-4 spec §6.1, commit 86b7460. + pub(crate) barge_epoch: u64, + pub(crate) metrics: Arc, +} + +impl Reflex

{ + pub fn new( + inner: P, + advisory_rx: mpsc::Receiver, + metrics: Arc, + ) -> Self { + Self { + inner, + advisory_rx, + muted: false, + barge_epoch: 0, + metrics, + } + } + + /// Drain all pending advisories + apply the state table. Called at + /// the top of `next_pcm_frame`. Hot-path: try_recv loop, bounded. + fn drain_advisories(&mut self) { + while let Ok(ev) = self.advisory_rx.try_recv() { + match ev { + AdvisoryEvent::SpeechStarted { at } => { + self.muted = true; + self.barge_epoch = self.barge_epoch.wrapping_add(1); + self.inner.barge_in_flush(); + self.metrics.barge_in_count.fetch_add(1, Ordering::Relaxed); + tracing::info!(epoch = self.barge_epoch, ?at, "barge-in"); + } + AdvisoryEvent::SpeechStopped { at: _ } => { + self.metrics + .advisory_observed_speech_stopped + .fetch_add(1, Ordering::Relaxed); + // No state change — see slice-4 spec §3.2. + } + } + } + } +} + +impl AudioSource for Reflex

{ + fn next_pcm_frame(&mut self) -> Option { + 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 AudioSink for Reflex

{ + 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 AudioPipe for Reflex

{ + fn clear_playout_ring(&mut self) { + self.inner.clear_playout_ring() + } + fn barge_in_flush(&mut self) { + self.inner.barge_in_flush() + } +} + #[cfg(test)] mod tests { use super::*; + // Task 2: tokio's mpsc provides the advisory channel the production + // `Reflex` consumes from a tokio task (the TapEngine); the tests drive + // it from `#[tokio::test]`. The `pcm` items name the trait bounds on + // `Reflex` + the `PcmFrame` the source/sink methods pass. + use crate::pcm::{AudioPipe, AudioSink, AudioSource, PcmFrame}; + use tokio::sync::mpsc; #[test] fn reflex_metrics_snapshot_reads_zeroes_initially() { @@ -126,4 +246,176 @@ mod tests { let st = AdvisoryEvent::SpeechStopped { at: Instant::now() }; let _ = format!("{:?}", st); } + + /// 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, + 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 { + 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, + mpsc::Sender, + Arc, + ) { + let (tx, rx) = mpsc::channel::(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). + } }