From 0b054e76f49c5e4a039b4256686b90ac626d83e6 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Thu, 2 Jul 2026 17:47:43 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(media):=20AdvisoryEvent=20+=20ReflexMe?= =?UTF-8?q?trics=20+=20barge=5Fin=5Fflush=20trait=20(slice-4=20=C2=A73.1,?= =?UTF-8?q?=20=C2=A73.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. Signed-off-by: Aaron D. Lee --- crates/rutster-media/src/lib.rs | 2 + crates/rutster-media/src/pcm.rs | 13 +++ crates/rutster-media/src/reflex.rs | 129 +++++++++++++++++++++++++++++ 3 files changed, 144 insertions(+) create mode 100644 crates/rutster-media/src/reflex.rs diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index 00bfef8..d9e23e0 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -33,10 +33,12 @@ 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, ReflexMetrics, ReflexMetricsSnapshot}; // Reflex re-export re-enabled in Task 2 pub use rtc_session::{RtcSession, RtcSessionError}; use thiserror::Error; diff --git a/crates/rutster-media/src/pcm.rs b/crates/rutster-media/src/pcm.rs index 6ef0190..51ac5f0 100644 --- a/crates/rutster-media/src/pcm.rs +++ b/crates/rutster-media/src/pcm.rs @@ -112,6 +112,19 @@ pub trait AudioPipe: AudioSource + AudioSink { /// have nothing to flush and inherit this default. Concrete pipes with /// a playout ring (slice-2's `TapAudioPipe`) override. 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(); + } } impl AudioPipe for EchoAudioPipe {} diff --git a/crates/rutster-media/src/reflex.rs b/crates/rutster-media/src/reflex.rs new file mode 100644 index 0000000..3b79f49 --- /dev/null +++ b/crates/rutster-media/src/reflex.rs @@ -0,0 +1,129 @@ +//! # Reflex — the FOB barge-in reflex (spec §3.1, §3.2; slice-4) +//! +//! `Reflex` 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: `LocalVadReflex

` composes outside the advisory +//! `Reflex

`, the same way `Reflex` composes today (spec +//! §6.4). `LocalVadReflex

` is the PRIMARY trigger this slice — it lands +//! via Task 2b — so the pattern is exercised, not speculative. Keeping the +//! advisory `Reflex` and the local-VAD `Reflex` as separate decorators (rather +//! than fusing them into one type) preserves an independent override seam: +//! each layer can be swapped or tested in isolation. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +// 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` +// (plus `PcmFrame` for the source/sink methods it delegates to). They are +// intentionally absent here to keep Task 1 self-contained (no `tokio` dep +// added yet — Task 2 adds it together with the type that consumes it). + +/// 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 { + 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); + } +} From d06e79cc509e7cd8ced4967e17bdaa7d3ace3663 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Thu, 2 Jul 2026 17:50:54 -0400 Subject: [PATCH 2/2] =?UTF-8?q?docs(media):=20add=20reflex=20module-map=20?= =?UTF-8?q?entry=20to=20lib.rs=20catalogue=20(slice-4=20=C2=A73.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 1 review flagged the new reflex module was missing from the module-map //! catalogue. AGENTS.md learner-facing convention requires the catalogue entry describing what the module does + why it exists. Addresses the Minor finding from the Task 1 task review. Signed-off-by: Aaron D. Lee --- crates/rutster-media/src/lib.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index d9e23e0..de717dc 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -27,6 +27,8 @@ //! - [`pcm`] — `PcmFrame` + `AudioSource`/`AudioSink` traits (the tap //! seam) + `EchoAudioPipe` (slice-1 wiring). //! - [`opus_codec`] — `OpusDecoder`/`OpusEncoder` wrappers. +//! - [`reflex`] (slice-4) — `AdvisoryEvent` + `ReflexMetrics`; the FOB +//! barge-in reflex state machine will decorate `AudioPipe` here (Task 2). //! - [`loop_driver`] (Task 4) — the str0m poll loop on tokio. //! - [`rtc_session`] (Task 4) — `RtcSession`, the per-peer owner.