//! # concurrency — `ConcurrencyRunner`: N concurrent `SimCall`s + sweep
//! report aggregation
//!
//! See spec §3.4 + §4.2 + §2.4 for the design + plan Task S5 for the
//! implementation. Spawns N concurrent `SimCall`s
//! (N ∈ `SWEEP_CONCURRENCIES = [1, 10, 50]`) against the same scenario,
//! awaits all, aggregates per-call latencies into the
//! `PerConcurrencyReport` rows of the `SweepReport`.
//!
//! # Why the merge happens at the sample level (not the captures level)
//!
//! Each `SimCall` produces a `LatencyProbe` with its own `Capture`
//! timeline. The naïve aggregation would be: concatenate all capture
//! vectors + run `LatencyProbe::kill_times()` on the merged timeline.
//! That fails when probes interleave: if probe A's `CallerLoudOnset`
//! is followed by probe B's `CallerLoudOnset` before A's
//! `BargeKillObserved`, the `LatencyProbe`'s pairing state
//! (`last_onset: Option<Instant>`) gets overwritten by B's onset —
//! A's kill would pair with B's onset, corrupting both metrics.
//!
//! The correct aggregation: compute each probe's `kill_times()` and
//! `mouth_to_ear_times()` INDEPENDENTLY (because each probe's captures
//! form a self-consistent timeline), then merge the *sample vectors*
//! and compute p50/p99 over the merged sample. This is what the
//! `ConcurrencyRunner` does. The `percentile_ms` helper in
//! `latency.rs` is `pub(crate)` for this purpose.
//!
//! # Tick-lag gauge (S6 fills this in)
//!
//! The `PerConcurrencyReport` schema includes `max_tick_lag_micros` +
//! `tick_overruns` + `total_ticks` + `tick_overrun_pct` per
//! spec §3.6. S5 leaves them zero-initialized; S6 (TickLagGauge)
//! fills them in by polling `MediaCmd::Stats` (or the in-standalone-
//! wiring equivalent) per second during the sweep.

use std::time::Duration;

use crate::latency::percentile_ms;
use crate::runner::SimCall;
use crate::scenario::Scenario;
use crate::thresholds::SWEEP_CONCURRENCIES;
use crate::tick_lag::TickLagStats;

/// The concurrency sweep runner. Spawns N `SimCall`s in parallel
/// (tokio), awaits all, aggregates per-call latencies into the sweep
/// report.
pub struct ConcurrencyRunner {
    /// Concurrency levels to sweep (per spec §2.4: 1/10/50).
    /// Filtered by `max_concurrency` at construction for test ergonomics
    /// (`in_process(1)` for fast unit tests; `in_process(50)` for the
    /// full CI sweep).
    concurrencies: Vec<usize>,
}

impl ConcurrencyRunner {
    /// Construct a runner that sweeps the canonical concurrency levels
    /// (`SWEEP_CONCURRENCIES = [1, 10, 50]`) capped at `max_concurrency`.
    /// The CI threshold sweep uses `in_process(50)`; unit tests use
    /// `in_process(1)` for speed.
    pub fn in_process(max_concurrency: usize) -> Self {
        let concurrencies: Vec<usize> = SWEEP_CONCURRENCIES
            .iter()
            .filter(|&&n| n <= max_concurrency)
            .copied()
            .collect();
        Self { concurrencies }
    }

    /// Run the full sweep; return the per-concurrency-level report.
    ///
    /// Each level runs sequentially (N=1 first; then N=10; then N=50).
    /// Within a level, the N `SimCall`s run concurrently via
    /// `tokio::spawn` + `tokio::join`. This phase structure matches
    /// spec §4.2: a clean before-and-after read of the tick-lag gauge
    /// per level (S6 polls the gauge during the sweep).
    pub async fn run(&self, scenario: Scenario) -> SweepReport {
        let mut per_concurrency = Vec::with_capacity(self.concurrencies.len());
        for &n in &self.concurrencies {
            let row = self.run_one_concurrency(n, scenario.clone()).await;
            per_concurrency.push(row);
        }
        SweepReport { per_concurrency }
    }

    /// Drive one concurrency level: spawn N `SimCall`s concurrently
    /// and aggregate their per-call `LatencyProbe` samples into
    /// p50/p99 + carry the empty tick-lag fields for S6 to fill.
    async fn run_one_concurrency(&self, n: usize, scenario: Scenario) -> PerConcurrencyReport {
        // S6: per-level shared gauge. Each SimCall's `run_with_gauge`
        // records per-tick wall-clock duration into this shared
        // accumulator. After the sweep, read max_tick_lag_micros +
        // tick_overruns + total_ticks + tick_overrun_pct to populate
        // PerConcurrencyReport's tick-lag fields (the ADR-0010
        // doctrine-drift detector).
        let gauge = TickLagStats::new();

        // Spawn N concurrent sim calls. Each task gets its own clone
        // of the scenario (Scenario: Clone — cheap, just a name + vec)
        // + a clone of the gauge handle (Arc — cheap refcount bump).
        let mut handles = Vec::with_capacity(n);
        for _ in 0..n {
            let scenario_clone = scenario.clone();
            let gauge_clone = gauge.clone();
            handles.push(tokio::spawn(async move {
                SimCall::new(scenario_clone)
                    .run_with_gauge(gauge_clone)
                    .await
            }));
        }

        // Await all + collect probes. `expect` here is OK (not the hot
        // path): a JoinError means a sim task panicked — surfaced as a
        // test failure, not silently swallowed per the "no fudged
        // assertions" rule from AGENTS.md.
        let mut probes = Vec::with_capacity(n);
        for h in handles {
            probes.push(h.await.expect("sim task panicked"));
        }

        // Aggregate samples across all N probes — see module docs for why
        // this happens at the sample-vector level (independent per-probe
        // pairing) rather than at the captures-vector level.
        let mut all_kills: Vec<Duration> = Vec::new();
        let mut all_m2e: Vec<Duration> = Vec::new();
        for p in &probes {
            all_kills.extend(p.kill_times());
            all_m2e.extend(p.mouth_to_ear_times());
        }

        PerConcurrencyReport {
            concurrency: n,
            p50_kill_ms: percentile_ms(&all_kills, 50),
            p99_kill_ms: percentile_ms(&all_kills, 99),
            p50_mouth_to_ear_ms: percentile_ms(&all_m2e, 50),
            p99_mouth_to_ear_ms: percentile_ms(&all_m2e, 99),
            max_tick_lag_micros: gauge.max_tick_lag_micros(),
            tick_overruns: gauge.tick_overruns(),
            total_ticks: gauge.total_ticks(),
            tick_overrun_pct: gauge.tick_overrun_pct(),
        }
    }
}

/// The artifact feeding the CI assertions (spec §3.4). The thresholds
/// in S7 assert `report.per_concurrency[i].p99_kill_ms <=
/// BARGE_IN_KILL_TIME_P99_MS` etc.
#[derive(Debug)]
pub struct SweepReport {
    pub per_concurrency: Vec<PerConcurrencyReport>,
}

/// One row of the sweep (one concurrency level's measurements). The
/// tick-lag fields (`max_tick_lag_micros`, `tick_overruns`,
/// `total_ticks`, `tick_overrun_pct`) are zero-initialized by S5 +
/// filled by S6.
#[derive(Debug)]
pub struct PerConcurrencyReport {
    pub concurrency: usize,
    pub p50_kill_ms: f64,
    pub p99_kill_ms: f64,
    pub p50_mouth_to_ear_ms: f64,
    pub p99_mouth_to_ear_ms: f64,
    /// From slice-5/seams `MediaCmd::Stats` (when wired through
    /// MediaThread) — OR from the S6 in-standalone-wiring equivalent
    /// (the SimCall's own tick-loop duration samples, since S4's
    /// standalone path doesn't go through MediaThread). The
    /// "doctrine-drift detector" for the timing-thread debt — ADR-0010's
    /// debt-pairing readout.
    pub max_tick_lag_micros: u64,
    pub tick_overruns: u64,
    pub total_ticks: u64,
    pub tick_overrun_pct: f64,
}

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

    /// 1-concurrency sweep produces a single-row report. The trivial
    /// scenario (3 loud frames + End) terminates fast (sub-second) —
    /// keeps test time low. The threshold assertions in S7 use scenarios
    /// with 20 loud frames (real `loud-barge.toml` shape).
    #[tokio::test]
    async fn concurrency_run_at_1_produces_report() {
        let runner = ConcurrencyRunner::in_process(1);
        let scenario = Scenario::from_toml(
            r#"
            name = "trivial"
            [[steps]]
            kind = "speak_loud"
            frames = 3
            [[steps]]
            kind = "end"
        "#,
        )
        .unwrap();
        let report = runner.run(scenario).await;
        assert_eq!(report.per_concurrency.len(), 1);
        let row = &report.per_concurrency[0];
        assert_eq!(row.concurrency, 1);
        // At 1 concurrency with 3 loud frames, the VAD trips on the 3rd
        // → at least one kill_time sample → p99_kill_ms non-NaN + ≤
        // BARGE_IN_KILL_TIME_P99_MS (80ms).
        assert!(
            !row.p99_kill_ms.is_nan(),
            "expected non-NaN p99_kill_ms at N=1"
        );
    }

    /// 10-concurrency sweep produces a single-row report at N=10 (since
    /// `in_process(10)` filters SWEEP_CONCURRENCIES to [1, 10]). Each
    /// row's report is checked for structure (per_concurrency[0] is
    /// N=1, [1] is N=10 if S5 ran both levels — but the test below
    /// scopes to in_process(10) to trim test duration).
    #[tokio::test]
    async fn concurrency_run_at_10_reports_at_least_one_kill() {
        let runner = ConcurrencyRunner::in_process(10);
        let scenario = Scenario::from_toml(
            r#"
            name = "trivial"
            [[steps]]
            kind = "speak_loud"
            frames = 3
            [[steps]]
            kind = "end"
        "#,
        )
        .unwrap();
        let report = runner.run(scenario).await;
        // in_process(10) returns concurrency levels [1, 10].
        assert_eq!(report.per_concurrency.len(), 2);
        assert_eq!(report.per_concurrency[0].concurrency, 1);
        assert_eq!(report.per_concurrency[1].concurrency, 10);

        // Each row should have non-NaN p99_kill_ms (each SimCall
        // triggers at least one VAD bar).
        for row in &report.per_concurrency {
            assert!(
                !row.p99_kill_ms.is_nan(),
                "expected non-NaN p99_kill_ms at N={}",
                row.concurrency
            );
        }
    }
}
