crates/rutster-media/src/lib.rs

@@ -27,16 +27,20 @@
 //! - [`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.
 
 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);
+    }
+}

docs/superpowers/specs/2026-07-01-slice-4-barge-in-design.md

@@ -131,12 +131,15 @@ applies the state machine and delegates to `inner.next_pcm_frame()` per the tabl
 
 The reflex owns three pieces of state:
 - `advisory_rx: mpsc::Receiver<AdvisoryEvent>` — drained sync-non-blocking via `try_recv`
-  on the 20 ms tick before delegating to `inner`. Fed by the TapEngine task over tokio mpsc.
+  on the 20 ms tick before delegating to `inner`. Fed by TWO senders on the same channel:
+  the outer `LocalVadReflex` (local VAD, the primary trigger) and the TapEngine task (the
+  brain's advisory, secondary/confirmation) — both over tokio mpsc.
 - `muted: bool` — the kill state. `next_pcm_frame` returns `None` while muted, *unless* the
   inner returns `Some` (the resume condition — the first fresh `audio_out` clears mute).
-- `barge_epoch: u64` — incremented on every `SpeechStarted`. Not strictly required for the
-  advisory-only MVP (the flush + drain makes the resume race-free), but it's the seam for a
-  future local-VAD wrapper that could race the advisory. Documented as forward-compatible.
+- `barge_epoch: u64` — incremented on every `SpeechStarted`. Load-bearing this slice: the
+  local VAD (primary) and the brain's advisory (secondary) can both fire on the same barge,
+  and the epoch distinguishes a genuine re-barge from the slower confirmation landing on the
+  same event (§3.2). The flush + drain keeps resume race-free regardless.
 
 ### 2.2 The dedicated media thread
 
@@ -241,10 +244,12 @@ un-mutes too early, brain's audio overlaps caller's next word). The resume condi
 "the brain has yielded and started a new response," which is provably signaled by the first
 `audio_out` frame after the barge — not by the caller's silence.
 
-**Why `epoch`:** not strictly needed for advisory-only (MVP), but it's the seam for the
-local-VAD backstop (deferred per §1.2). A future `LocalVadReflex` wrapper racing the
-advisory would need to disambiguate "is this barge a re-barge of the same event or a new
-one" — the epoch is the disambiguator. Forward-compatible.
+**Why `epoch`:** with two concurrent trigger sources this slice — the local VAD (primary)
+and the brain's advisory (secondary/confirmation, §6.1) — the epoch disambiguates "is this
+barge a re-barge of the same event, or a new one" when both race into the same advisory
+channel. The local VAD trips first; the brain's slower ASR-grade advisory lands ~300 ms
+later on the *same* event and must not be counted as a fresh barge. The epoch is that
+disambiguator — load-bearing now, not a forward-compat seam.
 
 ### 3.3 `AudioPipe` trait extension
 
@@ -487,7 +492,7 @@ impl<P: AudioPipe> AudioPipe for Reflex<P> {
     }
 
     fn barge_in_flush(&mut self) {
-        // Allow outer wrappers (future local-VadReflex) to barge the inner.
+        // Allow the outer `LocalVadReflex` (primary trigger) to barge the inner.
         self.inner.barge_in_flush()
     }
 }
@@ -710,15 +715,15 @@ from §6.4, which was originally specced as deferred to "when local VAD arrives.
 
 ### 6.4 Why `Reflex<P>` as a wrapper (not inline in `TapAudioPipe`)
 
-- Composition: a future `LocalVadReflex<P>` composes outside the advisory `Reflex<P>`, the
-  same way `Reflex<TapAudioPipe>` composes today. The pattern (decorator over `AudioPipe`) is
-  forward-compatible without restructuring.
+- Composition: `LocalVadReflex<P>` composes outside the advisory `Reflex<P>` (§6.1), the
+  same way `Reflex<TapAudioPipe>` composes. The pattern (decorator over `AudioPipe`) stacks
+  the two trigger sources without restructuring either.
 - The seam: `loop_driver.rs` byte-identical (still calls `pipe.next_pcm_frame()`). If the
   reflex lived inline in `TapAudioPipe`, the binary-side wiring would still change but the
   `TapAudioPipe` module itself would grow the reflex state — less isolated.
-- YAGNI caveated: the wrapper is the right abstraction for advisory-only because there's
-  exactly one reflex. When local VAD arrives, the wrapper pattern pays off; the spec does
-  not pre-empt that by collapsing the wrapper now.
+- The payoff is realized this slice, not deferred: two stacked reflexes (local-VAD primary +
+  advisory secondary) live as independent, separately-testable decorators rather than one
+  module's commingled state. Keeping them as wrappers is what makes that separation free.
 
 ### 6.5 Why `barge_in_flush` on `AudioPipe` (not just `clear_playout_ring`)