rutster/crates/rutster-media/src/reflex.rs

//! # 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);
    }
}