feat(media): Reflex<P> barge-in state machine (slice-4 §3.2, §3.4)

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 <himself@adlee.work>

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;

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<P>` 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<P>` lands: `Reflex` holds the
 // `mpsc::Receiver<AdvisoryEvent>` 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<dyn AudioPipe>`)
+///
+/// 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<dyn AudioPipe>` 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<P: AudioPipe> {
+    pub(crate) inner: P,
+    pub(crate) advisory_rx: mpsc::Receiver<AdvisoryEvent>,
+    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<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 slice-4 spec §3.2.
+                }
+            }
+        }
+    }
+}
+
+impl<P: AudioPipe> AudioSource for Reflex<P> {
+    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
+        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<P: AudioPipe> AudioSink for Reflex<P> {
+    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<P: AudioPipe> AudioPipe for Reflex<P> {
+    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<P: AudioPipe>` + 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<PcmFrame>,
+        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<PcmFrame> {
+            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<MockPipe>,
+        mpsc::Sender<AdvisoryEvent>,
+        Arc<ReflexMetrics>,
+    ) {
+        let (tx, rx) = mpsc::channel::<AdvisoryEvent>(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).
+    }
 }