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