crates/rutster-media/src/lib.rs

@@ -40,7 +40,10 @@ 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, Reflex, ReflexMetrics, ReflexMetricsSnapshot};
+pub use reflex::{
+    AdvisoryEvent, LocalVadReflex, Reflex, ReflexMetrics, ReflexMetricsSnapshot,
+    VAD_DEBOUNCE_FRAMES, VAD_RMS_THRESHOLD,
+};
 pub use rtc_session::{RtcSession, RtcSessionError};
 
 use thiserror::Error;

crates/rutster-media/src/reflex.rs

@@ -12,10 +12,11 @@
 //!
 //! 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:
+//! §6.4). `LocalVadReflex<P>` is the PRIMARY trigger this slice — the
+//! local RMS/energy VAD fires in the 20 ms tick with zero brain round-trip —
+//! 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;
@@ -193,6 +194,102 @@ impl<P: AudioPipe> AudioPipe for Reflex<P> {
     }
 }
 
+/// RMS energy threshold for caller-speech detection (slice-4 spec §3.4).
+/// The MVP ships with a single tuned-for-synthetic-loud-signal const;
+/// the tuning framework (per-environment calibration, adaptive noise
+/// floor) is deferred per slice-4 §1.2.
+pub const VAD_RMS_THRESHOLD: f64 = 500.0;
+
+/// Number of consecutive above-threshold frames required before the VAD
+/// trips (slice-4 spec §3.4). At 20 ms/frame, N=3 = 60 ms of above-
+/// threshold audio — well below the brain's ~300 ms ASR-VAD latency.
+pub const VAD_DEBOUNCE_FRAMES: u32 = 3;
+
+/// The PRIMARY barge-in trigger (slice-4 spec §3.4): a local in-core
+/// RMS/energy VAD running in `on_pcm_frame` on the dedicated thread, in
+/// the 20 ms loop, with ZERO brain round-trip. Proves wedge #1 ("VAD
+/// killing TTS the instant the caller speaks, without the brain" —
+/// README:98-100, ARCHITECTURE.md:79-81). Composes as
+/// `LocalVadReflex<Reflex<TapAudioPipe>>` — the outer wrapper does local
+/// VAD; the inner wrapper applies the mute state machine to the advisory
+/// stream (which has TWO sources: local VAD + brain advisory, both
+/// feeding the same mpsc).
+pub struct LocalVadReflex<P: AudioPipe> {
+    pub(crate) inner: P,
+    pub(crate) advisory_tx: mpsc::Sender<AdvisoryEvent>,
+    pub(crate) above_threshold_streak: u32,
+    pub(crate) vad_armed: bool,
+}
+
+impl<P: AudioPipe> LocalVadReflex<P> {
+    pub fn new(inner: P, advisory_tx: mpsc::Sender<AdvisoryEvent>) -> Self {
+        Self {
+            inner,
+            advisory_tx,
+            above_threshold_streak: 0,
+            vad_armed: true,
+        }
+    }
+
+    /// Compute RMS energy of a PCM frame. ~480 multiplications + one
+    /// sqrt — well under the 20 ms tick budget. Hot-path, no allocations.
+    fn rms(frame: &PcmFrame) -> f64 {
+        let sum_sq: u64 = frame
+            .samples
+            .iter()
+            .map(|&s| (s as i64 * s as i64) as u64)
+            .sum();
+        (sum_sq as f64 / frame.samples.len() as f64).sqrt()
+    }
+
+    /// Inspect a caller PCM frame + apply the debounce state machine.
+    /// Returns true if the VAD tripped THIS call (so on_pcm_frame can
+    /// push the advisory). Called from `on_pcm_frame` (the sink path).
+    fn observe(&mut self, frame: &PcmFrame) -> bool {
+        let energy = Self::rms(frame);
+        if energy >= VAD_RMS_THRESHOLD {
+            self.above_threshold_streak += 1;
+            if self.above_threshold_streak >= VAD_DEBOUNCE_FRAMES && self.vad_armed {
+                self.vad_armed = false;
+                return true;
+            }
+        } else {
+            self.above_threshold_streak = 0;
+            self.vad_armed = true;
+        }
+        false
+    }
+}
+
+impl<P: AudioPipe> AudioSource for LocalVadReflex<P> {
+    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
+        self.inner.next_pcm_frame()
+    }
+}
+
+impl<P: AudioPipe> AudioSink for LocalVadReflex<P> {
+    fn on_pcm_frame(&mut self, frame: PcmFrame) {
+        // THE PRIMARY TRIGGER: inspect BEFORE delegating.
+        if self.observe(&frame) {
+            let _ = self
+                .advisory_tx
+                .try_send(AdvisoryEvent::SpeechStarted { at: Instant::now() });
+            // try_send failure (channel full) → drop + observe (hot-path
+            // policy). The brain's advisory path is the backstop.
+        }
+        self.inner.on_pcm_frame(frame)
+    }
+}
+
+impl<P: AudioPipe> AudioPipe for LocalVadReflex<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::*;
@@ -421,4 +518,99 @@ mod tests {
             "inbound audio must reach inner even during barge"
         );
     }
+
+    /// RMS of a zeroed frame is 0.0 (perfect silence).
+    #[test]
+    fn rms_of_silence_is_zero() {
+        let frame = PcmFrame::zeroed();
+        assert_eq!(LocalVadReflex::<MockPipe>::rms(&frame), 0.0);
+    }
+
+    /// RMS of a loud frame is well above the threshold.
+    #[test]
+    fn rms_of_loud_frame_exceeds_threshold() {
+        let mut frame = PcmFrame::zeroed();
+        for s in frame.samples.iter_mut() {
+            *s = 1000; // well above VAD_RMS_THRESHOLD (500.0)
+        }
+        assert!(LocalVadReflex::<MockPipe>::rms(&frame) >= VAD_RMS_THRESHOLD);
+    }
+
+    /// Debounce: N-1 above-threshold frames do NOT trip; the Nth does.
+    #[tokio::test]
+    async fn debounce_requires_n_consecutive_above_threshold_frames() {
+        let (tx, mut rx) = mpsc::channel::<AdvisoryEvent>(16);
+        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
+        let mut loud = PcmFrame::zeroed();
+        for s in loud.samples.iter_mut() {
+            *s = 1000;
+        }
+
+        // VAD_DEBOUNCE_FRAMES - 1 frames: no trip.
+        for _ in 0..(VAD_DEBOUNCE_FRAMES - 1) {
+            vad.on_pcm_frame(loud.clone());
+            assert!(
+                rx.try_recv().is_err(),
+                "no advisory before debounce threshold"
+            );
+        }
+        // Nth frame: trip!
+        vad.on_pcm_frame(loud.clone());
+        let ev = rx.try_recv().expect("advisory after debounce threshold");
+        assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. }));
+    }
+
+    /// Re-arm: a below-threshold frame resets the streak + re-arms.
+    #[tokio::test]
+    async fn below_threshold_re_arms_vad() {
+        let (tx, mut rx) = mpsc::channel::<AdvisoryEvent>(16);
+        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
+        let mut loud = PcmFrame::zeroed();
+        for s in loud.samples.iter_mut() {
+            *s = 1000;
+        }
+        let quiet = PcmFrame::zeroed();
+
+        // Trip the VAD.
+        for _ in 0..VAD_DEBOUNCE_FRAMES {
+            vad.on_pcm_frame(loud.clone());
+        }
+        let _ = rx.try_recv().expect("first trip");
+
+        // Caller goes quiet — re-arm.
+        vad.on_pcm_frame(quiet);
+
+        // Next streak trips again.
+        for _ in 0..VAD_DEBOUNCE_FRAMES {
+            vad.on_pcm_frame(loud.clone());
+        }
+        let ev = rx.try_recv().expect("second trip after re-arm");
+        assert!(matches!(ev, AdvisoryEvent::SpeechStarted { .. }));
+    }
+
+    /// on_pcm_frame ALWAYS delegates to inner (caller audio reaches the brain
+    /// even during barge — the FOB only kills playout, not the caller's path).
+    #[tokio::test]
+    async fn on_pcm_frame_always_delegates_to_inner() {
+        let (tx, _rx) = mpsc::channel::<AdvisoryEvent>(16);
+        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
+        let frame = PcmFrame::zeroed();
+        vad.on_pcm_frame(frame.clone());
+        // The inner MockPipe captured it — verified by the lack of panic
+        // + the MockPipe's on_pcm_frame being called (push_back_bounded
+        // on the underlying queue, which we don't observe here directly;
+        // the absence of a drop is the assertion).
+    }
+
+    /// next_pcm_frame is pure delegation — the VAD only observes the SINK path.
+    #[tokio::test]
+    async fn next_pcm_frame_delegates_to_inner() {
+        let (tx, _rx) = mpsc::channel::<AdvisoryEvent>(16);
+        let mut vad = LocalVadReflex::new(MockPipe::new(), tx);
+        // Inner has no frames queued → None.
+        assert!(vad.next_pcm_frame().is_none());
+        // Queue a frame on the inner directly + verify it comes through.
+        vad.inner.push_frame(PcmFrame::zeroed());
+        assert!(vad.next_pcm_frame().is_some());
+    }
 }