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); + } +}