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

//! # latency — post-hoc p50/p99 metric computer (spec §3.3)
//!
//! Consumes a vector of `Capture` events from a `SimAudioPipe` and
//! computes the two p50/p99 metrics the threshold gates assert against:
//!
//! - barge-in kill-time: caller-speech-onset (`CallerLoudOnset`) → first
//!   `BargeKillObserved` thereafter. Per-call measurement; slice-4's
//!   ≤60 ms kill budget is the load-bearing assertion.
//! - mouth-to-ear: caller-speech-onset (`CallerLoudOnset`) → next
//!   `CallerHeardReply` thereafter. Per-call measurement; slice-1's 200 ms
//!   notification + slice-3's ~300 ms mock brain round-trip is the budget.
//!
//! # Why post-hoc (not on-tick)
//!
//! The hot path (the `SimAudioPipe::next_pcm_frame`/`on_pcm_frame` calls)
//! captures `Instant::now()` timestamps but defers the metric math to
//! post-run. This keeps the tick free of allocations (a p99 computation
//! needs a sorted sample vector — sort + index isn't free) + lets the
//! assertions be made against the canonical timeline once, not on every
//! capture. The `Instant::now()` calls inside `SimAudioPipe` are the only
//! measurement-side cost on the hot path; the `LatencyProbe::kill_times()`
//! etc. scan + sort happen after the `SimCall::run` returns.
//!
//! # Pairing semantics + the `BargeKillObserved` noise problem
//!
//! The `SimAudioPipe` captures `BargeKillObserved` *unconditionally* on
//! every empty `reply_ring` (see `sim_audio_pipe` module docs). Most of
//! those captures are noise — there's no `CallerLoudOnset` to pair them
//! with. The `LatencyProbe` (this module) is the dedup gate: it pairs
//! each `CallerLoudOnset` with the next `BargeKillObserved` thereafter,
//! and silently drops `BargeKillObserved` captures without a prior
//! onset. The kill-time metric thus reflects only post-onset kills,
//! never bare noise.

use std::time::{Duration, Instant};

use crate::sim_audio_pipe::Capture;

/// The post-hoc metric computer. Construct from a `Vec<Capture>` drained
/// out of a `SimAudioPipe` via `take_captures()`.
///
/// # Example
///
/// ```no_run
/// use rutster_sim::{SimAudioPipe, LatencyProbe};
/// # fn wrapper(mut pipe: SimAudioPipe) {
/// // ... drive the pipe through a scenario ...
/// let captures = pipe.take_captures();
/// let probe = LatencyProbe::from_captures(captures);
/// let p99_kill = probe.p99_kill_ms();
/// let p99_m2e = probe.p99_mouth_to_ear_ms();
/// # }
/// ```
///
/// # Why this is a struct (not free fns on Vec<Capture>)
///
/// The struct holds the captures by value (one allocation per run, post
/// the hot path). Free fns would require either passing `&[Capture]`
/// everywhere (lifetime noise at every call site) OR cloning the vector
/// on every percentile computation (the p50/p99 helpers would each
/// clone + sort independently — wasteful). The struct pattern matches
/// the std-library convention for "data + its derived computations"
/// (cf. `std::process::Output`).
pub struct LatencyProbe {
    captures: Vec<Capture>,
}

impl LatencyProbe {
    /// Construct from a `Vec<Capture>`. Takes ownership — the probe is
    /// the sole consumer of this timeline.
    pub fn from_captures(captures: Vec<Capture>) -> Self {
        Self { captures }
    }

    /// Access the raw capture stream (read-only). Useful for
    /// debugging + for the `SweepReport`'s per-call logging.
    pub fn captures(&self) -> &[Capture] {
        &self.captures
    }

    /// Barge-in kill-times: pair each `CallerLoudOnset` with the *next*
    /// `BargeKillObserved` thereafter. Per-call measurement.
    ///
    /// # The pairing cursor
    ///
    /// A single linear scan walks the captures. `last_onset` holds the
    /// most recent unpaired `CallerLoudOnset`. On a `BargeKillObserved`:
    /// if `last_onset` is `Some(_)`, compute the duration + push + clear
    /// the cursor; if `None`, ignore (this is the noise case — a kill
    /// observed without a prior onset means an empty-ring tick before
    /// any caller speech — see module docs).
    ///
    /// # Why take() and not just read()
    ///
    /// `last_onset.take()` is `Option::take` — a Rust idiom for "replace
    /// with `None`, return the prior value." The cursor advances: a
    /// paired onset can't be re-paired with a later kill. This gives
    /// exactly one kill-time per onset; over-counting would corrupt the
    /// p99 sample.
    pub fn kill_times(&self) -> Vec<Duration> {
        let mut out = vec![];
        let mut last_onset: Option<Instant> = None;
        for c in &self.captures {
            match c {
                Capture::CallerLoudOnset { at } => last_onset = Some(*at),
                Capture::BargeKillObserved { at } => {
                    if let Some(on) = last_onset.take() {
                        // `saturating_duration_since` (not `duration_since`):
                        // a panic on out-of-order timestamps would be a
                        // sharp edge in the assertion path. The
                        // `captures_are_in_temporal_order` test in
                        // `sim_audio_pipe` guards against reordering at
                        // the capture site; here we defend in depth.
                        out.push(at.saturating_duration_since(on));
                    }
                    // (Else: kill without prior onset — noise; ignored.)
                }
                Capture::CallerHeardReply { .. } => {
                    // irrelevant to kill metric; mouth_to_ear_times handles
                }
            }
        }
        out
    }

    /// Mouth-to-ear: pair each `CallerLoudOnset` with the *next*
    /// `CallerHeardReply` thereafter. Per-call measurement.
    pub fn mouth_to_ear_times(&self) -> Vec<Duration> {
        let mut out = vec![];
        let mut last_onset: Option<Instant> = None;
        for c in &self.captures {
            match c {
                Capture::CallerLoudOnset { at } => last_onset = Some(*at),
                Capture::CallerHeardReply { at } => {
                    if let Some(on) = last_onset.take() {
                        out.push(at.saturating_duration_since(on));
                    }
                }
                Capture::BargeKillObserved { .. } => {
                    // irrelevant to mouth-to-ear metric
                }
            }
        }
        out
    }

    pub fn p50_kill_ms(&self) -> f64 {
        percentile_ms(&self.kill_times(), 50)
    }
    pub fn p99_kill_ms(&self) -> f64 {
        percentile_ms(&self.kill_times(), 99)
    }
    pub fn p50_mouth_to_ear_ms(&self) -> f64 {
        percentile_ms(&self.mouth_to_ear_times(), 50)
    }
    pub fn p99_mouth_to_ear_ms(&self) -> f64 {
        percentile_ms(&self.mouth_to_ear_times(), 99)
    }
}

/// Compute a percentile from a slice of durations, returning
/// milliseconds as `f64`.
///
/// # Returns
///
/// - `f64::NAN` for an empty slice (callers — the threshold assertion
///   tests — treat NaN as a deliberate fail-the-build signal: `assert!(
///   row.p99_kill_ms <= BARGE_IN_KILL_TIME_P99_MS)` panics on NaN).
/// - The percentile value in ms otherwise.
///
/// # Algorithm
///
/// 1. Convert each `Duration` to `u128` milliseconds
///    (`Duration::as_millis`).
/// 2. Sort the vector (unstable sort — captures don't carry additional
///    data so stability is irrelevant; unstable is faster).
/// 3. Index the sorted vector at `(len-1) * (pct/100)`, rounded.
///
/// # Why `(len-1) * (pct/100).round()` (not `len * pct / 100`)
///
/// Percentile conventions vary. The "rank" formula here uses the
/// nearest-rank method on a 0-based index: at pct=50 with len=5, the
/// index is `(5-1) * 0.5 = 2.0` → sorted[2] (the median). At pct=99 with
/// len=5, the index is `(5-1) * 0.99 = 3.96` → rounded to 4 → sorted[4]
/// (the max). This matches numpy's `np.percentile` "lower" interpolation;
/// it gives the worst-acceptable-case at p99 (the highest sample), which
/// is the load-bearing semantics for "the worst acceptable" assertion
/// (see spec §6.6 — p99, not p50, is the assertion gate).
///
/// `pub(crate)` so `ConcurrencyRunner` (S5) can compute p50/p99 over
/// the *merged sample across N probes* (each probe yields its own
/// `kill_times()` + `mouth_to_ear_times()`; merging samples + computing
/// the p99 in one pass avoids the "interleaved-captures across probes
/// corrupt the LatencyProbe pairing cursor" problem that would result
/// from combining `Capture` vectors naively).
pub(crate) fn percentile_ms(durations: &[Duration], pct: u8) -> f64 {
    if durations.is_empty() {
        return f64::NAN;
    }
    let mut sorted: Vec<u128> = durations.iter().map(|d| d.as_millis()).collect();
    sorted.sort_unstable();
    let idx = ((sorted.len() as f64 - 1.0) * (pct as f64 / 100.0)).round() as usize;
    // Clamp to len-1 to guard against rounding overflow at pct=100 (the
    // formula already stays in-bounds for pct<100, but pct=100 with
    // len=1 produces idx=0 which is fine; pct=100 with len>1 produces
    // idx=len-1 which is also fine). The .min() is belt-and-braces.
    let idx = idx.min(sorted.len() - 1);
    sorted[idx] as f64
}

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

    #[test]
    fn kill_times_empty_for_no_captures() {
        // The `NaN` return for empty inputs is the deliberate fail-the-
        // build signal: the threshold assertion `assert!(row.p99_kill_ms
        // <= THRESHOLD)` panics on NaN, surfacing "no captures → no
        // measurement → did the scenario run?" rather than silently
        // passing on `0.0`.
        let p = LatencyProbe::from_captures(vec![]);
        assert!(p.kill_times().is_empty());
        assert!(p.p99_kill_ms().is_nan());
    }

    #[test]
    fn kill_times_pairs_onset_with_next_barge_kill() {
        // The canonical pairing: one onset → one kill → one duration.
        // The 50 ms here corresponds to slice-4's kill budget (≤60 ms
        // budget + 20 ms observer slack = ≤80 ms CI assertion per
        // `BARGE_IN_KILL_TIME_P99_MS`).
        let t0 = Instant::now();
        let captures = vec![
            Capture::CallerLoudOnset { at: t0 },
            Capture::BargeKillObserved {
                at: t0 + Duration::from_millis(50),
            },
        ];
        let p = LatencyProbe::from_captures(captures);
        let kills = p.kill_times();
        assert_eq!(kills.len(), 1);
        assert_eq!(kills[0].as_millis(), 50);
    }

    #[test]
    fn mouth_to_ear_times_pairs_onset_with_next_reply() {
        // Same pairing shape, different capture variant. The 200 ms here
        // is slice-1's notification budget (the upperbound on the
        // "FOB saw the caller" → "FOB started pushing reply audio"
        // window before the brain round-trip lands).
        let t0 = Instant::now();
        let captures = vec![
            Capture::CallerLoudOnset { at: t0 },
            Capture::CallerHeardReply {
                at: t0 + Duration::from_millis(200),
            },
        ];
        let p = LatencyProbe::from_captures(captures);
        let m2e = p.mouth_to_ear_times();
        assert_eq!(m2e.len(), 1);
        assert_eq!(m2e[0].as_millis(), 200);
    }

    #[test]
    fn p99_returns_higher_than_p50_with_outlier() {
        // 5 onset→kill pairs: 50, 55, 60, 65, 200 ms. The outlier (200)
        // is the p99 case — the worst acceptable sample. p50 = median =
        // 60 ms. The (len-1)*0.99 = 3.96 → round = 4 → sorted[4] = 200;
        // (len-1)*0.5 = 2 → sorted[2] = 60.
        //
        // This test guards `percentile_ms` against the most common bug:
        // confusing p50 and p99 (returning the same value for both, or
        // returning min for p99 instead of max). A regression here would
        // make the threshold assertion falsely pass — the load-bearing
        // CI-regressed guarantee from ADR-0010 would silently degrade.
        let t0 = Instant::now();
        let mut captures = vec![];
        for ms in [50u64, 55, 60, 65, 200] {
            captures.push(Capture::CallerLoudOnset { at: t0 });
            captures.push(Capture::BargeKillObserved {
                at: t0 + Duration::from_millis(ms),
            });
        }
        let p = LatencyProbe::from_captures(captures);
        assert!(p.p99_kill_ms() > p.p50_kill_ms(), "p99 > p50 with outlier");
        assert!(p.p50_kill_ms() <= 65.0, "p50 = median");
    }

    #[test]
    fn barge_kill_without_prior_onset_is_ignored() {
        // The noise-suppression contract: a BargeKillObserved without a
        // prior CallerLoudOnset is dropped (the SimAudioPipe emits noise
        // captures on every empty ring; the LatencyProbe is the gate
        // that strips them). Without this filtering, the kill-times
        // vector would contain spurious sub-microsecond durations (the
        // gap between two consecutive noise captures), corrupting the
        // p99 sample.
        let captures = vec![
            Capture::BargeKillObserved { at: Instant::now() },
            Capture::BargeKillObserved { at: Instant::now() },
            Capture::BargeKillObserved { at: Instant::now() },
        ];
        let p = LatencyProbe::from_captures(captures);
        assert!(p.kill_times().is_empty(), "noise captures dropped");
    }
}