feat(sim): ConcurrencyRunner -- N concurrent SimCalls + SweepReport aggregation (slice-4½ S5)

ConcurrencyRunner::in_process(max_concurrency) filters SWEEP_CONCURRENCIES
to levels <= max_concurrency for test ergonomics (in_process(1) for fast
unit tests, in_process(50) for the full CI sweep). The runner sweeps each
level sequentially; within a level, spawns N concurrent SimCalls via
tokio::spawn + awaits all.

Aggregate samples across N probes by computing kill_times() + mouth_to_ear_times()
on each probe INDEPENDENTLY, then merging sample vectors + running percentile_ms once
on the merged set. This avoids the interleaved-captures-corrupt-LatencyProbe-pairing
problem that would result from concatenating Capture vectors naively when probes
interleave in the wall clock.

SweepReport / PerConcurrencyReport match spec section 3.4. Tick-lag fields
(max_tick_lag_micros / tick_overruns / total_ticks / tick_overrun_pct) are
zero-initialized -- S6 fills them in.

percentile_ms in latency.rs is pub(crate) so ConcurrencyRunner can compute p50/p99
on the merged sample (was private).

Signed-off-by: Aaron D. Lee <himself@adlee.work>

crates/rutster-sim/src/concurrency.rs

@@ -1,11 +1,235 @@
-//! # concurrency — `ConcurrencyRunner`: N concurrent `SimCall`s + sweep-report
-//! aggregation
+//! # concurrency — `ConcurrencyRunner`: N concurrent `SimCall`s + sweep
+//! report aggregation
 //!
-//! **Stub — lands in S5.**
-//!
-//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`
-//! §3.4 + §4.2 + §2.4 for the design + `docs/superpowers/plans/2026-07-05-slice-4-half-benchmark-sim.md`
-//! Task S5 for the implementation. Spawns N concurrent `SimCall`s
-//! (N ∈ `SWEEP_CONCURRENCIES = [1, 10, 50]`) against the same in-process
-//! `MockRealtimeBrain`, aggregates per-call latencies into the
+//! 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;
+
+/// 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 {
+        // Spawn N concurrent sim calls. Each task gets its own clone
+        // of the scenario (Scenario: Clone — cheap, just a name + vec).
+        let mut handles = Vec::with_capacity(n);
+        for _ in 0..n {
+            let scenario_clone = scenario.clone();
+            handles.push(tokio::spawn(async move {
+                SimCall::new(scenario_clone).run().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),
+            // S6 (TickLagGauge) fills these in during the sweep; S5
+            // leaves them zero-initialized so the SweepReport's
+            // structure is stable across task landings.
+            max_tick_lag_micros: 0,
+            tick_overruns: 0,
+            total_ticks: 0,
+            tick_overrun_pct: 0.0,
+        }
+    }
+}
+
+/// 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
+            );
+        }
+    }
+}

crates/rutster-sim/src/latency.rs

@@ -186,7 +186,14 @@ impl LatencyProbe {
 /// 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).
-fn percentile_ms(durations: &[Duration], pct: u8) -> f64 {
+///
+/// `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;
     }

crates/rutster-sim/src/lib.rs

@@ -58,6 +58,7 @@ pub mod sim_audio_pipe;
 pub mod thresholds;
 pub mod tick_lag;
 
+pub use concurrency::{ConcurrencyRunner, PerConcurrencyReport, SweepReport};
 pub use latency::LatencyProbe;
 pub use runner::{ScenarioRunner, SimCall};
 pub use scenario::{Scenario, ScenarioError, ScenarioStep};