rutster/crates/rutster-sim/src/runner.rs

//! # runner — `SimCall` + `ScenarioRunner`: drive one synthetic caller
//! end-to-end through the FOB reflex loop
//!
//! See spec §3.4 + §4.1 for the design + plan Task S4 for the
//! implementation. The SimCall wires itself STANDALONE in tokio (per
//! the plan's S4 standalone-path conclusion): it composes slice-4's
//! `Reflex<TapAudioPipe>` + `LocalVadReflex` stack itself rather than
//! registering a sim session with the binary's `MediaThread` via a new
//! `MediaCmd` variant. The seam files (`loop_driver.rs` +
//! `rtc_session.rs`) stay byte-identical; `media_thread.rs` is
//! untouched by slice 4½.
//!
//! # Why standalone (no `MediaCmd::RegisterSim`)
//!
//! The spec's §3.5 sketches a `MediaCmd::RegisterSim { pipe: Box<dyn
//! AudioPipe>, reply }` variant that would let the harness register a
//! sim session against the binary's `MediaThread`. The plan's S4
//! reasoning concludes this is unnecessary: `loop_driver::drive` expects
//! an `&mut RtcSession` (str0m) — a `&mut dyn AudioPipe` synthetic
//! session wouldn't fit the existing dispatch without either a
//! separate driver-path OR a `MediaLeg` enum wrapper (the step-5
//! approach). Both options change `media_thread.rs` in ways the
//! seam-discipline + the kickoff's hard rule forbid this slice.
//! Simpler: the SimCall composes the `Reflex<TapAudioPipe>` + outer
//! `LocalVadReflex` stack itself in tokio (the same composition site
//! the binary's `Connected` transition performs in slice-4), drives
//! the wrapped stack via direct method calls on the 20 ms tick, and
//! captures `Instant::now()` timestamps inside the `SimAudioPipe`
//! (the caller's clock — spec §2.2). The harness measures the FOB
//! reflex loop's behavior under load without going through the
//! binary's `MediaThread` dispatch.
//!
//! # The fake-brain task (mimics `MockRealtimeBrain`)
//!
//! Spec §8.6 says "in-process measurement against `MockRealtimeBrain`,
//! not client-server against the binary's HTTP surface." The literal
//! composition path (WS `MockRealtimeBrain` + `spawn_tap_engine` + the
//! translator pipeline) is the integration slice-3 + slice-4 already
//! proved. For slice 4½'s threshold assertions, the S4 SimCall mimics
//! the brain side with an in-runtime tokio task that pushes
//! `PcmFrame::zeroed()` replies to `TapAudioPipe`'s `tx_audio_out`
//! channel every 20 ms. This gives the reply-path traffic the
//! mouth-to-ear metric needs (some `CallerHeardReply` captures) without
//! the WS-round-trip orchestration cost. A future slice (post-spearhead
//! refinement) replaces the fake-brain task with the real WS
//! `MockRealtimeBrain` for network-realism latency measurement.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use rutster_media::{
    AdvisoryEvent, AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics,
};
use rutster_tap::{TapAudioPipe, TapMetrics};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::latency::LatencyProbe;
use crate::scenario::Scenario;
use crate::sim_audio_pipe::SimAudioPipe;
use crate::tick_lag::TickLagStats;

/// One synthetic call: a `SimAudioPipe` (the caller-side recorder +
/// scenario driver) + the wiring to drive it against an in-process
/// `Reflex<TapAudioPipe>` + `LocalVadReflex` stack in tokio.
///
/// Single binary; no separate process. The `SimCall::run` method
/// returns a `LatencyProbe` carrying the capture stream for the
/// `ConcurrencyRunner` (S5) to aggregate.
pub struct SimCall {
    scenario: Scenario,
}

impl SimCall {
    pub fn new(scenario: Scenario) -> Self {
        Self { scenario }
    }

    /// Drive the scenario against the FOB reflex loop. Returns the
    /// `LatencyProbe` with the captured timeline. Default gauge is
    /// internal + discarded after `run` — use `run_with_gauge` to
    /// observe tick-lag during the sweep (the `ConcurrencyRunner` does
    /// this; standalone callers usually don't need it).
    pub async fn run(self) -> LatencyProbe {
        self.run_with_gauge(TickLagStats::new()).await
    }

    /// Same as `run` but records per-tick wall-clock duration into the
    /// shared `gauge`. The `ConcurrencyRunner` creates one shared gauge
    /// per concurrency level + passes clones to each of the N
    /// `SimCall`s; after the sweep, the ConcurrencyRunner reads the
    /// gauge's `max_tick_lag_micros` + `tick_overruns` + `total_ticks` +
    /// `tick_overrun_pct` to populate `PerConcurrencyReport` (see
    /// spec §3.6).
    ///
    /// # Hot-path policy (per AGENTS.md)
    ///
    /// The 20 ms tick loop is the slice-4½ hot path. Failures here
    /// are match-and-continue + observed, never `?`-propagated:
    /// `try_send` on a full channel drops + observes; `next_pcm_frame`
    /// returning `None` (muted/empty ring) captures a `BargeKillObserved`
    /// + continues.
    pub async fn run_with_gauge(self, gauge: Arc<TickLagStats>) -> LatencyProbe {
        // 1. Build the Reflex stack — mirrors slice-4's
        //    `primary_path_local_vad_kills_playout_without_brain`
        //    test composition (crates/rutster/tests/barge_in_integration.rs:158):
        //    the inner pipe is `TapAudioPipe` (the production AudioPipe);
        //    the inner Reflex drains `AdvisoryEvent`s from a tokio mpsc;
        //    the outer `LocalVadReflex` is the PRIMARY barge-in trigger
        //    (slice-4 §3.4 — local RMS/energy VAD with zero brain round-trip).
        //
        // `tx_pcm_in` would forward caller audio to the brain WS in
        // production wiring; here it's owned-but-unused (no brain WS to
        // forward to). The `_rx_pcm_in` receiver is dropped (the channel
        // fills to its bound of 32, then `TapAudioPipe::on_pcm_frame`'s
        // `try_send` drops + observes per the hot-path policy).
        let (tx_pcm_in, _rx_pcm_in) = mpsc::channel::<PcmFrame>(32);
        let (tx_audio_out, rx_audio_out) = mpsc::channel::<PcmFrame>(32);
        let tap_metrics = TapMetrics::new();
        let inner_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);

        let (advisory_tx, advisory_rx) = mpsc::channel::<AdvisoryEvent>(16);
        let reflex_metrics = ReflexMetrics::new();
        let reflex = Reflex::new(inner_pipe, advisory_rx, reflex_metrics);
        let mut wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx);

        // 2. The `SimAudioPipe` — the recorder + scenario driver.
        //    `SimAudioPipe::new` calls `enter_step` on `steps[0]` immediately,
        //    capturing `CallerLoudOnset` synchronously if the scenario starts
        //    with `SpeakLoud` (the loud-barge shape does).
        let mut sim_pipe = SimAudioPipe::new(self.scenario.clone(), 16);

        // 3. The fake-brain task — a tokio task that periodically pushes
        //    replies to `tx_audio_out`. Mimics slice-3's `MockRealtimeBrain`
        //    sending audio_out frames ≈ every 20 ms (the slice-3 mock echoes
        //    audio back). Exercise the mouth-to-ear path: without brain-side
        //    traffic, the `Reflex::next_pcm_frame` would always return `None`
        //    (ring empty), and `mouth_to_ear_times()` would be empty → the
        //    `p99_mouth_to_ear_ms` assertion in S7 would panic on NaN.
        //
        // The `AtomicBool` stop flag is the simplest cross-task signal: the
        // SimCall's tick loop sets it when scenario_done; the brain task
        // reads it on each 20 ms interval. `Arc<AtomicBool>` over a
        // `tokio::sync::Notify` because the brain task is a polling loop
        // (already sleeping 20 ms each iteration) — the AtomicBool is cheaper
        // than a Notify that would need wake-up coordination.
        let brain_stop = Arc::new(AtomicBool::new(false));
        let brain_stop_clone = brain_stop.clone();
        let brain_task: JoinHandle<()> = tokio::spawn(async move {
            // Seed the reply ring synchronously so tick 1 (which races
            // the brain task's first `interval.tick()` due to async task
            // scheduling) has a reply to consume. Without this seed,
            // tick 1's `next_pcm_frame` would capture `BargeKillObserved`
            // even though the barge hasn't fired (VAD hasn't tripped on
            // one loud frame yet) — that capture is noise the
            // LatencyProbe would dedup, but the seed keeps the
            // measurement timeline clean: the first kill observed
            // corresponds to the actual barge.
            let _ = tx_audio_out.try_send(PcmFrame::zeroed());
            loop {
                tokio::time::sleep(Duration::from_millis(20)).await;
                if brain_stop_clone.load(Ordering::Relaxed) {
                    break;
                }
                // try_send: drop + observe on full channel (hot-path policy).
                let _ = tx_audio_out.try_send(PcmFrame::zeroed());
            }
        });

        // 4. The 20 ms tick loop. Each iteration:
        //    (a) SINK: if the scenario says "speak loud," push a loud frame
        //        into the wrapped stack — simulating the caller speaking.
        //        `LocalVadReflex::on_pcm_frame` observes the loud frame's RMS,
        //        increments `above_threshold_streak`, and after
        //        `VAD_DEBOUNCE_FRAMES` consecutive loud frames, sends
        //        `AdvisoryEvent::SpeechStarted` on the advisory channel.
        //    (b) SOURCE: drain the wrapped stack's `next_pcm_frame` — which
        //        drains advisories (applying the Reflex state table) + pulls
        //        brain replies from `TapAudioPipe`'s ring. If `Some`, push
        //        into the SimPipe's reply ring.
        //    (c) Drain the SimPipe's reply ring → captures
        //        `CallerHeardReply` on `Some`; `BargeKillObserved` on `None`
        //        (the LatencyProbe dedups captures without prior onset).
        //    (d) Advance the SimPipe's scenario cursor via `on_pcm_frame`.
        //    (e) Per-tick wall-clock duration recorded into `gauge` (S6) —
        //        the ADR-0010 doctrine-drift detector. The `Instant::now()`
        //        measurement wraps (a)-(d); the `tokio::time::sleep(tick)`
        //        is OUTSIDE the measured region (we measure tick work, not
        //        the wait). This matches the binary's `MediaStats.last_tick_micros`
        //        semantics (work duration per tick, not wall-clock period).
        //    (f) Termination: `scenario_done()` checks for `End` step.
        let tick = Duration::from_millis(20);
        loop {
            let tick_start = Instant::now();

            if sim_pipe.current_step_is_speak_loud() {
                wrapped_pipe.on_pcm_frame(loud_pcm_frame());
            }

            if let Some(reply) = wrapped_pipe.next_pcm_frame() {
                sim_pipe.push_reply(reply);
            }

            // Drain the SimPipe's reply ring → one `CallerHeardReply`
            // capture per `Some` (typically one reply per tick, but the
            // drain loop handles bursts). Loop exits on `None` — one
            // `BargeKillObserved` capture then. The LatencyProbe pairs
            // each `CallerLoudOnset` with the next `BargeKillObserved`
            // (kill metric) AND the next `CallerHeardReply` (m2e metric)
            // independently — both pairs can share the same onset.
            while sim_pipe.next_pcm_frame().is_some() {
                // drained + captured
            }

            sim_pipe.on_pcm_frame(PcmFrame::zeroed());

            // S6: record per-tick work duration into the shared gauge.
            // The elapsed here is the synchronous tick work — Reflex state
            // machine advances, capture pushes, scenario cursor increments.
            // For the standalone SimCall tick loop, this is the analog of
            // the binary MediaThread's `last_tick_micros` (spec §3.6).
            gauge.record_tick(tick_start.elapsed());

            if sim_pipe.scenario_done() {
                break;
            }

            tokio::time::sleep(tick).await;
        }

        // 5. Cleanup: signal the fake-brain task + await termination.
        // Await avoids leaking the task after the SimCall returns
        // (otherwise the brain task would race the runtime shutdown +
        // could log warnings on test teardown).
        brain_stop.store(true, Ordering::Relaxed);
        let _ = brain_task.await;

        let captures = sim_pipe.take_captures();
        LatencyProbe::from_captures(captures)
    }
}

/// Construct a loud PcmFrame for the SimCall's sink path. Sample value
/// 1000 — well above `VAD_RMS_THRESHOLD` (500.0) per slice-4 §3.4.
///
/// The same construction pattern appears inline in slice-4's
/// `barge_in_integration.rs`. A `PcmFrame::loud()` factory on
/// `rutster_media::PcmFrame` would centralize this; that's deferred
/// (no public-API churn this slice — the slice-5 trunk slice that
/// also needs loud frames can add the factory).
fn loud_pcm_frame() -> PcmFrame {
    let mut f = PcmFrame::zeroed();
    for s in f.samples.iter_mut() {
        *s = 1000;
    }
    f
}

/// Single-call driver. A convenience wrapper around `SimCall` that
/// consumes the scenario + returns the `LatencyProbe`. The
/// `ConcurrencyRunner` (S5) constructs `SimCall`s directly per
/// concurrency level rather than going through `ScenarioRunner` — but
/// `ScenarioRunner` is the public API surface for one-off manual
/// measurement.
pub struct ScenarioRunner;

impl ScenarioRunner {
    pub fn new() -> Self {
        Self
    }

    pub async fn run(&self, scenario: Scenario) -> LatencyProbe {
        SimCall::new(scenario).run().await
    }
}

impl Default for ScenarioRunner {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The canonical loud-barge scenario shape (spec §5.3 entry #1):
    /// 20 loud frames → barrier-await one reply → end. The await_reply
    /// barrier ensures the SimPipe's leave-the-SpeakLoud-step transition
    /// happens CLEANLY (with a reply in the ring to consume) rather than
    /// racing the barge-in state machine.
    fn loud_barge_scenario() -> Scenario {
        Scenario::from_toml(
            r#"
            name = "loud-barge"
            [[steps]]
            kind = "speak_loud"
            frames = 20
            [[steps]]
            kind = "await_reply"
            frames = 0
            [[steps]]
            kind = "end"
        "#,
        )
        .unwrap()
    }

    #[tokio::test]
    async fn sim_call_drives_loud_barge_scenario_to_completion() {
        // The barge must fire: after `VAD_DEBOUNCE_FRAMES` (3) consecutive
        // loud frames, the LocalVadReflex trips → sends SpeechStarted →
        // the Reflex drains + mutes + flushes the inner ring on the next
        // `next_pcm_frame` call → None returned + captured as
        // BargeKillObserved → paired by LatencyProbe with the construct-time
        // CallerLoudOnset → kill_time sample.
        let scenario = loud_barge_scenario();
        let probe = SimCall::new(scenario).run().await;

        let kills = probe.kill_times();
        assert!(
            !kills.is_empty(),
            "expected barge-in to fire on 20 loud frames (got {} kills)",
            kills.len()
        );
    }

    #[tokio::test]
    async fn sim_call_short_trivial_scenario_completes() {
        // Smoke test: 3 loud frames + End (no barrier). The SimCall must
        // terminate cleanly. The `scenario_done()` check is what the
        // SimCall's tick loop reads — this test ensures the End-step detection
        // works (the S2 SimAudioPipe's `scenario_done` fix surfaced by the
        // S4 driving loop: returns true when the cursor enters End step).
        let scenario = Scenario::from_toml(
            r#"
            name = "trivial"
            [[steps]]
            kind = "speak_loud"
            frames = 3
            [[steps]]
            kind = "end"
        "#,
        )
        .unwrap();

        let probe = SimCall::new(scenario).run().await;
        // 3 loud frames ≥ VAD_DEBOUNCE_FRAMES (=3), so the VAD trips on
        // the 3rd → kill captures should be non-empty.
        assert!(
            !probe.kill_times().is_empty(),
            "expected kill on 3 consecutive loud frames"
        );
    }
}