From 6da8e4095a18a3c3881115d2f3dda31a6589b465 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sun, 5 Jul 2026 03:20:37 -0400 Subject: [PATCH] =?UTF-8?q?feat(sim):=20ConcurrencyRunner=20--=20N=20concu?= =?UTF-8?q?rrent=20SimCalls=20+=20SweepReport=20aggregation=20(slice-4?= =?UTF-8?q?=C2=BD=20S5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/rutster-sim/src/concurrency.rs | 242 +++++++++++++++++++++++++- crates/rutster-sim/src/latency.rs | 9 +- crates/rutster-sim/src/lib.rs | 1 + 3 files changed, 242 insertions(+), 10 deletions(-) diff --git a/crates/rutster-sim/src/concurrency.rs b/crates/rutster-sim/src/concurrency.rs index 8b58df0..89b4966 100644 --- a/crates/rutster-sim/src/concurrency.rs +++ b/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`) 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, +} + +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 = 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 = Vec::new(); + let mut all_m2e: Vec = 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, +} + +/// 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 + ); + } + } +} diff --git a/crates/rutster-sim/src/latency.rs b/crates/rutster-sim/src/latency.rs index 7ded63c..ec40bf3 100644 --- a/crates/rutster-sim/src/latency.rs +++ b/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; } diff --git a/crates/rutster-sim/src/lib.rs b/crates/rutster-sim/src/lib.rs index 80b0a55..512ada4 100644 --- a/crates/rutster-sim/src/lib.rs +++ b/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};