feat(media): AdvisoryEvent + ReflexMetrics + barge_in_flush trait (slice-4 §3.1, §3.3)

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: opencode controller <controller@local>

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;

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 {}

crates/rutster-media/src/reflex.rs

@@ -0,0 +1,129 @@
+//! # Reflex — the FOB barge-in reflex (spec §3.1, §3.2; slice-4)
+//!
+//! `Reflex<P: AudioPipe>` 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<P>` composes outside the advisory
+//! `Reflex<P>`, the same way `Reflex<TapAudioPipe>` composes today (spec
+//! §6.4). `LocalVadReflex<P>` 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<P>` lands: `Reflex` holds the
+// `mpsc::Receiver<AdvisoryEvent>` 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<Self> {
+        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);
+    }
+}