# Slice 4½ — Benchmark + simulation harness — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Stand up spearhead step 4½ — a self-hostable benchmark + simulation harness in a new `crates/rutster-sim/` crate. Drives synthetic callers through the SAME media-leg path real callers use; measures p50/p99 mouth-to-ear latency + barge-in kill-time at 1/10/50 concurrent calls; a separate CI job (`cargo test --all --features=sim-bench`) asserts thresholds per commit. A latency regression fails the build (ADR-0010's central demand). **Architecture:** A new `crates/rutster-sim/` crate (currently non-existent) houses the harness: `Scenario` + `ScenarioStep` types deserialize from TOML; `SimAudioPipe: AudioPipe` drives a scenario on `on_pcm_frame` (caller speaks) and captures brain replies on `next_pcm_frame` (caller hears) — both timestamps anchored to `Instant::now()` (monotonic, identical clock). `LatencyProbe` post-hoc computes p50/p99 kill + mouth-to-ear durations from the `SimAudioPipe`'s captures. `ConcurrencyRunner` spawns N concurrent `SimCall`s against an in-process `MediaThread` + aggregates per-call latencies + the slice-5/seams `MediaStats.{tick_overruns, last_tick_micros}` gauge. CI asserts thresholds at each of [1, 10, 50] concurrency. **Tech Stack:** Rust stable + 1.85 (CI matrix — for the routine gate; `sim-bench` job is stable-only), the existing workspace deps (`tokio`, `serde`, `toml`, `axum`), no new deps. ## Global Constraints - **License:** GPL-3.0-or-later on every crate manifest (ADR-0004). - **DCO:** every commit signed off — `git commit -s` (AGENTS.md Git workflow). Signoff identity = the human maintainer's git config, not the agent. - **Seam gate (UNCHANGED from slice-4 Task 10):** `crates/rutster-media/src/loop_driver.rs` + `crates/rutster-media/src/rtc_session.rs` stay byte-identical (CI pinned-blob gate). The new `MediaCmd::RegisterSim` variant lives in `crates/rutster/src/media_thread.rs`, NOT in the seam files. - **Hot-path policy (when the harness drives the 20 ms tick):** never `?`-propagate; match-and- continue; "drop + observe (log + counter), don't crash." No `unwrap()`/`expect()` outside tests/const-init. - **Code style:** `cargo fmt` is the single whitespace source of truth. `clippy -D warnings` is the lint bar. Newtype wrappers over primitives — not needed this slice (no new primitive types surface in the public API). - **Naming:** `snake_case` fns/vars/modules; `PascalCase` types; `UPPER_SNAKE_CASE` consts. - **Learner-facing comments:** this project OVERRIDES the no-comments default. Every new public item gets `///` docs; every new module gets `//!` docs citing the design intent. Snippets below show the load-bearing comments; implementers keep that density. - **Measurement discipline:** the harness captures timestamps ONLY inside `SimAudioPipe` (the caller's clock — see spec §2.2). Do NOT add `Instant::now()` calls in `loop_driver.rs`, `rtc_session.rs`, or `media_thread.rs` for measurement — that would re-introduce instrumentation in the seam or the binary's hot path, defeating §2.3's design choice. - **CI gates:** `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all` (stable + 1.85 — routine gate, UNCHANGED, no sim-bench feature on by default), `cargo deny check`, and the NEW `cargo test --all --features=sim-bench` job on stable. - **Branch/PR:** branch `slice-4-half/sim-harness-dev-a` off `main`; PR via `tea` (not `gh`). ## File Structure ### New files | Path | Responsibility | |---|---| | `crates/rutster-sim/Cargo.toml` | Workspace member manifest. Edition 2024. Deps: `rutster-media`, `rutster`, `tokio`, `serde`, `toml`, `tracing`. The `sim-bench` feature is defined here (default off). | | `crates/rutster-sim/src/lib.rs` | `//! module docs; pub mod scenario; pub mod sim_audio_pipe; pub mod latency; pub mod runner; pub mod concurrency; pub mod tick_lag; pub mod thresholds;` + the `pub use` re-exports. | | `crates/rutster-sim/src/scenario.rs` | `Scenario`, `ScenarioStep` enums + `Scenario::load(path)` TOML deserializer. | | `crates/rutster-sim/src/sim_audio_pipe.rs` | `SimAudioPipe: AudioPipe` implementation + `Capture` enum. | | `crates/rutster-sim/src/latency.rs` | `LatencyProbe` — post-hoc p50/p99 computation from `Capture` event stream. | | `crates/rutster-sim/src/runner.rs` | `SimCall` (one synthetic caller) + `ScenarioRunner` (single-call driver). | | `crates/rutster-sim/src/concurrency.rs` | `ConcurrencyRunner` — N concurrent `SimCall`s + `SweepReport` / `PerConcurrencyReport` aggregations. | | `crates/rutster-sim/src/tick_lag.rs` | `TickLagGauge` — polls `MediaCmd::Stats` during the sweep + surfaces `tick_overruns` / `last_tick_micros` per spec §3.6. | | `crates/rutster-sim/src/thresholds.rs` | Threshold constants (`BARGE_IN_KILL_TIME_P99_MS = 80.0`, etc.) + the `#[cfg(feature = "sim-bench")] #[tokio::test]` threshold-assertion tests. | | `crates/rutster-sim/scenarios/loud-barge.toml` | Scenario pack: 20 loud frames → await reply → end. Drives the PRIMARY barge-in path. | | `crates/rutster-sim/scenarios/quiet-advisory.toml` | Scenario pack: quiet frames, exercises slice-3 advisory plumbing. | | `crates/rutster-sim/scenarios/sustained-call.toml` | Scenario pack: 5 minutes of talk with 3 barge cycles. | | `crates/rutster-sim/src/mulaw_table.rs` (data file) | Not used in 4½ — included in case it's needed for a future harness extension. (Skip this file; listed only for symmetry.) | ### Modified files | Path | What changes | |---|---| | `Cargo.toml` (workspace root) | Add `"crates/rutster-sim"` to `[workspace] members`. | | `crates/rutster/src/media_thread.rs` | Add `MediaCmd::RegisterSim { pipe: Box, reply: oneshot::Sender }` variant; the std thread's `cmd_rx.try_recv()` loop handles it by allocating a `ChannelId`, constructing a synthetic "session" entry that drives the harness's `SimAudioPipe` through the same `loop_driver::drive` calls as real WebRTC sessions (the seam holds). | | `crates/rutster/src/lib.rs` | `pub mod media_thread;` already exists (slice-4); no change this slice. | | `.github/workflows/ci.yml` | Add `sim-bench` job: stable-only, runs `cargo test --all --features=sim-bench -- --test-threads=1`. | | `LEARNING.md` | Pointer to `crates/rutster-sim/` after Task S8 lands. | ### SEAM-INVARIANT files (DO NOT TOUCH) - `crates/rutster-media/src/loop_driver.rs` — **byte-identical** to slice-3 (and slice-4). - `crates/rutster-media/src/rtc_session.rs` — **byte-identical** to slice-3 (and slice-4). Every dispatched dev MUST respect this. The new `MediaCmd::RegisterSim` variant lands in `media_thread.rs` (the binary-side bridge), NOT in the seam files. ## Task ordering (for Kimi-worker subagent dispatch) The chain is **strictly linear** (dev-a solo in the strategic plan §4.1): each task consumes the prior task's types. Fanned-out across this 8-task chain, parallelism stalls. Instead, this slice is best executed by one Kimi dev driving the chain serially. - **S1** — CRITICAL-PATH FOUNDATION. `crates/rutster-sim/` skeleton + `Scenario`/`ScenarioStep` types. Lands first; every later task imports these. - **S2** — `SimAudioPipe: AudioPipe`. Depends on S1 (uses `Scenario` + `ScenarioStep`). - **S3** — `LatencyProbe`. Depends on S2 (consumes `Capture` events from the SimAudioPipe). - **S4** — `SimCall` + `ScenarioRunner`. Depends on S2 + the merged `MediaThread` (slice-4 + slice-5/seams). Drives a single end-to-end sim call against an in-process MediaThread. - **S5** — `ConcurrencyRunner`. Depends on S4 (spawns N SimCalls + aggregates). - **S6** — `TickLagGauge`. Depends on S5 (the sweep needs to poll tick-lag stats per second). - **S7** — `cargo test --features=sim-bench` CI job + threshold consts + assertion tests. Depends on S5 + S6 (the assertions are end-to-end). - **S8** — Scenario pack + LEARNING.md pointer. Filler; any time after S4. Parallelizable-now filler (a blocked dev picks these up without blocking the critical path): - LEARNING.md pointer to the new `scenario.rs` (after S1 lands) + `sim_audio_pipe.rs` (after S2). - `cargo doc` rendering checks (after S2 + the crate skeleton stabilize). --- ### Task S1: `crates/rutster-sim/` skeleton + `Scenario`/`ScenarioStep` types — the critical-path foundation **Files:** - Create: `crates/rutster-sim/Cargo.toml` - Create: `crates/rutster-sim/src/lib.rs` - Create: `crates/rutster-sim/src/scenario.rs` - Modify: `Cargo.toml` (workspace root — add the new member) - Test: `crates/rutster-sim/src/scenario.rs` (inline `#[cfg(test)] mod tests`) **Interfaces:** - Consumes: nothing (pure-data types + TOML deserialization). - Produces: - `pub struct Scenario { pub name: String, pub steps: Vec }` - `pub enum ScenarioStep { SpeakLoud { frames: u32 }, SpeakQuiet { frames: u32 }, Pause { frames: u32 }, AwaitReply { frames: u32 }, End }` - `impl Scenario { pub fn load(path: impl AsRef) -> Result }` - `pub struct ScenarioError(pub String);` - [ ] **Step 1: Write the workspace member manifest** Create `crates/rutster-sim/Cargo.toml`: ```toml # crates/rutster-sim/Cargo.toml [package] name = "rutster-sim" version = "0.0.0" license.workspace = true edition.workspace = true repository.workspace = true description = "Self-hostable benchmark + simulation harness (ADR-0010 spearhead step 4½)." [dependencies] rutster-media = { path = "../rutster-media" } rutster = { path = "../rutster" } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] } serde = { workspace = true, features = ["derive"] } toml = { workspace = true } tracing = { workspace = true } [features] default = [] # The CI-regressed threshold sweep. Default OFF so `cargo test --all` (the # routine gate) stays fast. A separate CI job runs # `cargo test --all --features=sim-bench`. See spec §5.4 + §6.5. sim-bench = [] ``` Add the new member to the workspace root `Cargo.toml`: ```toml # Cargo.toml (root) [workspace] members = [ "crates/rutster", "crates/rutster-brain-realtime", "crates/rutster-call-model", "crates/rutster-media", "crates/rutster-sim", # <- NEW "crates/rutster-spend", "crates/rutster-tap", "crates/rutster-tap-echo", "crates/rutster-trunk", ] ``` - [ ] **Step 2: Write the crate's `lib.rs` module-doc header** Create `crates/rutster-sim/src/lib.rs`: ```rust //! # rutster-sim — the self-hostable benchmark + simulation harness //! //! **Status:** spearhead step 4½ (ADR-0010). The wedge's measurement surface. //! //! This crate drives synthetic callers through the SAME media-leg path real //! callers use, measures p50/p99 mouth-to-ear latency + barge-in kill-time //! against slice-4's ≤60 ms kill budget, and runs the same measurements at //! 1 / 10 / 50 concurrent calls. A separate CI job //! (`cargo test --all --features=sim-bench`) asserts thresholds per commit; //! a latency regression fails the build (ADR-0010). //! //! # Why this crate exists (the FOB differentiator) //! //! Slice-4 ships a reflex loop + a synthetic e2e test. SIM-BENCH is the //! artifact that turns arithmetic latency claims into CI-regressed //! measurement. See //! [`docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`] //! for the design. //! //! # Why a separate crate (not in-tree tests) //! //! The harness is hot-path-adjacent + differentiating (ADR-0008 FOB) — it //! earns cratehood the same way `rutster-tap` did. The dep direction is //! clean: `rutster-sim` → `rutster-media` + `rutster`. The harness //! consumes types; it doesn't ride on the binary's internal plumbing. pub mod concurrency; pub mod latency; pub mod runner; pub mod scenario; pub mod sim_audio_pipe; pub mod thresholds; pub mod tick_lag; // Re-exports for the public API surface. pub use concurrency::{ConcurrencyRunner, PerConcurrencyReport, SweepReport}; pub use latency::LatencyProbe; pub use runner::{ScenarioRunner, SimCall}; pub use scenario::{Scenario, ScenarioError, ScenarioStep}; pub use sim_audio_pipe::{Capture, SimAudioPipe}; pub use thresholds::{ BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS, TICK_OVERRUN_PCT_MAX, }; pub use tick_lag::TickLagGauge; ``` (Other modules are added in later tasks; their `pub mod` lines won't resolve until then — create empty stub files OR comment out the `pub mod` lines until the corresponding task lands. Recommendation: create empty files with just a `//! module docs` placeholder header each, so the crate compiles at every commit boundary.) For S1's commit alone, create these stub files alongside `scenario.rs`: - `src/sim_audio_pipe.rs` — `//! stub; lands in S2` - `src/latency.rs` — `//! stub; lands in S3` - `src/runner.rs` — `//! stub; lands in S4` - `src/concurrency.rs` — `//! stub; lands in S5` - `src/tick_lag.rs` — `//! stub; lands in S6` - `src/thresholds.rs` — `//! stub; lands in S7` (with the consts as immediate module-level `pub const` items per spec §5.1 — they're used by S5/S6/S7 wiring) - [ ] **Step 3: Write the failing test for `Scenario` deserialization** Create `crates/rutster-sim/src/scenario.rs` with the test module only: ```rust //! # Scenario — the scripted-caller data type //! //! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md` §3.1. //! //! A `Scenario` is a sequence of `ScenarioStep`s read from a TOML file under //! `crates/rutster-sim/scenarios/*.toml`. Deterministic by construction — //! the entire point is reproducible thresholds in CI (LLM-driven callers //! land in a post-spearhead refinement tier; see §1.2). //! //! # Why TOML (not YAML, not RON) //! //! `serde` + `toml` is already a workspace member. TOML keeps the scenario //! file readable as a one-shot script (a sequence of named steps + numbers); //! YAML would invite flow-mapping complexity this format doesn't need. use std::path::Path; /// The scripted-caller scenario. Read from a TOML file. Deterministic. #[derive(Debug, Clone, serde::Deserialize, PartialEq, Eq)] pub struct Scenario { /// Human-readable identifier; surfaces in CI failure messages. pub name: String, /// Time-ordered sequence of caller actions. pub steps: Vec, } /// One axis of caller behavior. A scenario is a time-ordered sequence. /// `SimAudioPipe` consumes them in order during `on_pcm_frame`. #[derive(Debug, Clone, serde::Deserialize, PartialEq, Eq)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum ScenarioStep { /// Send N loud PCM frames (sample value 1000, well above VAD_RMS_THRESHOLD=500.0). /// Triggers the local VAD via slice-4's `LocalVadReflex::on_pcm_frame`. SpeakLoud { frames: u32 }, /// Send N zero frames (sample value 0, well below VAD_RMS_THRESHOLD). /// Drives mock-brain advisory path (slice-4 §5.2 secondary path). SpeakQuiet { frames: u32 }, /// Insert N zero frames before the next step (silence pacing). Pause { frames: u32 }, /// Wait until the harness receives M "ear" frames before advancing. /// Barrier: brain's reply must arrive before the next caller action. AwaitReply { frames: u32 }, /// End the scenario. `SimAudioPipe` returns None from next_pcm_frame thereafter. End, } /// Errors surfaced during scenario loading. Cold-path; OK to be String-y. #[derive(Debug, thiserror::Error)] pub enum ScenarioError { #[error("scenario file read failed: {0}")] Io(#[from] std::io::Error), #[error("scenario TOML parse failed: {0}")] Parse(#[from] toml::de::Error), } impl Scenario { /// Load a scenario from a TOML file. Cold-path. pub fn load(path: impl AsRef) -> Result { let raw = std::fs::read_to_string(path)?; Self::from_toml(&raw) } /// Parse a scenario from an in-memory TOML string. Useful for tests. pub fn from_toml(s: &str) -> Result { Ok(toml::from_str(s)?) } } #[cfg(test)] mod tests { use super::*; #[test] fn scenario_parses_minimal_end_only() { let toml = r#" name = "trivial" [[steps]] kind = "end" "#; let s = Scenario::from_toml(toml).expect("parse"); assert_eq!(s.name, "trivial"); assert_eq!(s.steps, vec![ScenarioStep::End]); } #[test] fn scenario_parses_loud_barge_shape() { let toml = r#" name = "loud-barge" [[steps]] kind = "speak_loud" frames = 20 [[steps]] kind = "await_reply" frames = 0 [[steps]] kind = "end" "#; let s = Scenario::from_toml(toml).expect("parse"); assert_eq!(s.name, "loud-barge"); assert_eq!(s.steps, vec![ ScenarioStep::SpeakLoud { frames: 20 }, ScenarioStep::AwaitReply { frames: 0 }, ScenarioStep::End, ]); } #[test] fn scenario_parses_sustained_call_shape() { let toml = r#" name = "sustained" [[steps]] kind = "speak_loud" frames = 10 [[steps]] kind = "speak_quiet" frames = 10 [[steps]] kind = "speak_loud" frames = 10 [[steps]] kind = "end" "#; let s = Scenario::from_toml(toml).expect("parse"); assert_eq!(s.steps.len(), 4); assert!(matches!(s.steps[0], ScenarioStep::SpeakLoud { frames: 10 })); assert!(matches!(s.steps[1], ScenarioStep::SpeakQuiet { frames: 10 })); assert!(matches!(s.steps[2], ScenarioStep::SpeakLoud { frames: 10 })); assert!(matches!(s.steps[3], ScenarioStep::End)); } #[test] fn scenario_unknown_kind_errors() { let toml = r#" name = "bad" [[steps]] kind = "ship_a_real_caller" "#; assert!(Scenario::from_toml(toml).is_err()); } } ``` Add `thiserror` to the crate's deps (workspace member already): ```toml # crates/rutster-sim/Cargo.toml — append under [dependencies] thiserror = { workspace = true } ``` - [ ] **Step 4: Run the test to verify it passes** ```bash cargo test -p rutster-sim --lib scenario::tests ``` Expected: PASS (4 tests). If `thiserror` isn't in the workspace, switch to a plain `pub struct ScenarioError(pub String)` + manual `From` impls (avoid pulling a new dep this late; the workspace pattern is to consolidate). If `thiserror` IS already a workspace member (rustster-tap uses it), prefer the `#[derive(thiserror::Error)]` form above. - [ ] **Step 5: fmt + clippy + full test + commit** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo test --all # routine gate; sim-bench feature is OFF by default — must still pass git add Cargo.toml crates/rutster-sim/Cargo.toml crates/rutster-sim/src/ git commit -s -m "feat(sim): rutster-sim crate skeleton + Scenario/ScenarioStep types (slice-4½ S1) The critical-path foundation for the benchmark + simulation harness. Scenario is a TOML-deserializable scripted-caller data type; ScenarioStep covers speak_loud / speak_quiet / pause / await_reply / end. Determinism is the point — reproducible thresholds in CI (ADR-0010). All other sim modules land as stubs here + fill in across S2-S7. Task S1 of slice-4½ - everything else depends on this landing." ``` --- ### Task S2: `SimAudioPipe: AudioPipe` + `Capture` enum **Files:** - Modify: `crates/rutster-sim/src/sim_audio_pipe.rs` (replace stub) - Modify: `crates/rutster-sim/src/lib.rs` (already re-exports `SimAudioPipe` + `Capture`) - Test: `crates/rutster-sim/src/sim_audio_pipe.rs` (inline tests) **Interfaces:** - Consumes: `Scenario` + `ScenarioStep` (from S1), `PcmFrame` + `AudioPipe` + `AudioSource` + `AudioSink` from `rutster-media`. - Produces: - `pub enum Capture { CallerLoudOnset { at: Instant }, BargeKillObserved { at: Instant }, CallerHeardReply { at: Instant } }` - `pub struct SimAudioPipe { scenario, step_idx, step_frames_remaining, reply_frames_received, captures, reply_ring }` - `impl AudioSource for SimAudioPipe` + `impl AudioSink for SimAudioPipe` + `impl AudioPipe for SimAudioPipe` - `impl SimAudioPipe { pub fn new(scenario: Scenario, reply_ring_cap: usize) -> Self; pub fn take_captures(&mut self) -> Vec }` - [ ] **Step 1: Write the failing tests for the sim pipe's state machine** Replace `crates/rutster-sim/src/sim_audio_pipe.rs`'s stub with the test module + the impl. The state machine mirrors the spec §3.2.1 + §3.2.2: - `on_pcm_frame` advances the scenario cursor + emits `CallerLoudOnset` capture on a `SpeakLoud` step boundary. - `next_pcm_frame` returns frames from the `reply_ring` (filled externally — by the SimCall wiring in S4); emits `CallerHeardReply` on `Some`, `BargeKillObserved` on `None` (depending on scenario-phase). Tests to write (each, ≤30 LOC): ```rust #[cfg(test)] mod tests { use super::*; use crate::scenario::{Scenario, ScenarioStep}; use rutster_media::PcmFrame; use tokio::sync::mpsc; use std::time::Instant; fn trivial_scenario() -> Scenario { Scenario::from_toml(r#" name = "trivial" [[steps]] kind = "speak_loud" frames = 3 [[steps]] kind = "end" "#).unwrap() } #[test] fn speak_loud推进_step_cursor_on_each_on_pcm_frame() { let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); // First 3 on_pcm_frame calls consume the SpeakLoud step. for _ in 0..3 { pipe.on_pcm_frame(PcmFrame::zeroed()); } // After 3 frames, the 4th call advances to End (a no-op step) + captures CallerLoudOnset on the first speak. // Inspect captures: CallerLoudOnset captured once on the boundary of SpeakLoud. let caps = pipe.take_captures(); assert!(caps.iter().any(|c| matches!(c, Capture::CallerLoudOnset { .. })), "expected CallerLoudOnset captured when SpeakLoud step began"); } #[test] fn next_pcm_frame_returns_none_when_reply_ring_empty_and_emits_barge_kill_capture() { let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); // Reply ring is empty; next_pcm_frame returns None + captures BargeKillObserved. let r = pipe.next_pcm_frame(); assert!(r.is_none(), "empty reply_ring returns None"); let caps = pipe.take_captures(); assert!(caps.iter().any(|c| matches!(c, Capture::BargeKillObserved { .. })), "expected BargeKillObserved captured when reply_ring was empty"); } #[test] fn next_pcm_frame_returns_frame_and_emits_caller_heard_reply() { let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); // Push a synthetic reply frame into the ring. pipe.push_reply(PcmFrame::zeroed()); let r = pipe.next_pcm_frame().expect("reply"); let caps = pipe.take_captures(); assert!(caps.iter().any(|c| matches!(c, Capture::CallerHeardReply { .. })), "expected CallerHeardReply captured"); } #[test] fn captures_are_in_temporal_order() { let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); pipe.push_reply(PcmFrame::zeroed()); let _ = pipe.next_pcm_frame(); // CallerHeardReply pipe.on_pcm_frame(PcmFrame::zeroed()); // CallerLoudOnset (first SpeakLoud frame) let caps = pipe.take_captures(); assert!(caps.len() >= 2, "captured at least 2 events"); // Verify the events were captured in temporal order. for w in caps.windows(2) { // Each Capture::* variant's `at: Instant` — non-decreasing across the vector. let t1 = match &w[0] { Capture::CallerLoudOnset { at } | Capture::BargeKillObserved { at } | Capture::CallerHeardReply { at } => *at }; let t2 = match &w[1] { Capture::CallerLoudOnset { at } | Capture::BargeKillObserved { at } | Capture::CallerHeardReply { at } => *at }; assert!(t2 >= t1, "captures must be in non-decreasing Instant order"); } } #[test] fn take_captures_drains_and_subsequent_call_returns_empty() { let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); pipe.push_reply(PcmFrame::zeroed()); let _ = pipe.next_pcm_frame(); assert!(!pipe.take_captures().is_empty()); assert!(pipe.take_captures().is_empty(), "drained on first take_captures"); } } ``` - [ ] **Step 2: Run the test to verify it fails** ```bash cargo test -p rutster-sim --lib sim_audio_pipe::tests ``` Expected: compile errors — `SimAudioPipe`, `Capture` don't exist yet. - [ ] **Step 3: Implement `SimAudioPipe` + `Capture`** Add the struct + impls above the `#[cfg(test)] mod tests` in `sim_audio_pipe.rs`: ```rust //! # SimAudioPipe — the test-double AudioPipe that simulates a caller //! //! See spec §3.2. Drives a `Scenario` on `on_pcm_frame` (the sink path: //! caller speaks); receives brain response frames on `next_pcm_frame` //! (the source path: caller hears). Captures `Instant::now()` at every //! meaningful event for the `LatencyProbe` to consume. //! //! # Why this is THE measurement boundary //! //! Both clocks live INSIDE this pipe. The wall-clock the *caller* started //! speaking is captured here (we decided when to "speak"); the wall-clock //! the *caller* heard the reply is captured here (we observed the system's //! reply). See spec §2.2 — the harness can't lie about latency because the //! only clock it uses is the caller's. use std::collections::VecDeque; use std::time::Instant; use rutster_media::{AudioPipe, AudioSource, AudioSink, PcmFrame}; use crate::scenario::{Scenario, ScenarioStep}; /// A timestamped event captured by `SimAudioPipe`. Read by `LatencyProbe` /// post-run to compute p50/p99 latencies. #[derive(Debug, Clone, Copy)] pub enum Capture { /// The caller started speaking loudly (a `SpeakLoud` step began). CallerLoudOnset { at: Instant }, /// The FOB killed playout (a `next_pcm_frame` returned None immediately /// after a barge event). See spec §3.2.1. BargeKillObserved { at: Instant }, /// The caller heard a brain reply (a `next_pcm_frame` returned Some /// after the barge cleared). See spec §3.2.1. CallerHeardReply { at: Instant }, } /// The test-double AudioPipe. See module docs. pub struct SimAudioPipe { scenario: Scenario, step_idx: usize, /// Frames remaining in the current step (for SpeakLoud/SpeakQuiet/Pause). step_frames_remaining: u32, /// Frames received from `next_pcm_frame` while in `AwaitReply`. /// When this reaches the step's target, advance. await_reply_target: u32, /// Captures buffered for the LatencyProbe. Bounded; on overflow the /// oldest is dropped (hot-path policy — measurement shouldn't crash). captures: VecDeque, /// Pre-allocated reply frames pushed externally by the SimCall wiring /// (S4). The next_pcm_frame call pops from here. reply_ring: VecDeque, } const CAPTURE_RING_CAP: usize = 1024; impl SimAudioPipe { pub fn new(scenario: Scenario, reply_ring_cap: usize) -> Self { let mut pipe = Self { scenario, step_idx: 0, step_frames_remaining: 0, await_reply_target: 0, captures: VecDeque::with_capacity(CAPTURE_RING_CAP), reply_ring: VecDeque::with_capacity(reply_ring_cap), }; pipe.enter_step(); pipe } /// Push a synthetic brain-reply PCM frame into the pipe's ring. /// Called by the SimCall's WS-pump-equivalent wiring (S4 will connect this /// to the TapEngine's tx_audio_out mpsc). pub fn push_reply(&mut self, frame: PcmFrame) { self.reply_ring.push_back(frame); } /// Drain captures for the LatencyProbe. Consumes the buffer. pub fn take_captures(&mut self) -> Vec { self.captures.drain(..).collect() } /// Advance the step cursor; initialize per-step counters. fn enter_step(&mut self) { if self.step_idx >= self.scenario.steps.len() { // End-of-scenario: nothing to do; next_pcm_frame returns None, // on_pcm_frame is a no-op. The SimCall (S4) detects end + stops. return; } match &self.scenario.steps[self.step_idx] { ScenarioStep::SpeakLoud { frames } => { self.step_frames_remaining = *frames; // Capture onset at step entry. self.push_capture(Capture::CallerLoudOnset { at: Instant::now() }); } ScenarioStep::SpeakQuiet { frames } => { self.step_frames_remaining = *frames; // No capture for quiet onsets — the wedge cares about LOUD barge. } ScenarioStep::Pause { frames } => { self.step_frames_remaining = *frames; } ScenarioStep::AwaitReply { frames } => { self.await_reply_target = *frames; } ScenarioStep::End => { /* no-op */ } } } /// Move to the next step. Called when step_frames_remaining reaches 0 /// OR when await_reply_target is met. fn advance_step(&mut self) { self.step_idx += 1; self.enter_step(); } fn push_capture(&mut self, c: Capture) { if self.captures.len() >= CAPTURE_RING_CAP { // Drop oldest; bounded ring. self.captures.pop_front(); } self.captures.push_back(c); } /// What the current step's target is (for AwaitReply). fn current_step_target(&self) -> u32 { self.await_reply_target } fn is_in_await_reply_step(&self) -> bool { matches!(self.scenario.steps.get(self.step_idx), Some(ScenarioStep::AwaitReply { .. })) } } impl AudioSource for SimAudioPipe { fn next_pcm_frame(&mut self) -> Option { match self.reply_ring.pop_front() { Some(frame) => { if self.is_in_await_reply_step() { // Count this reply toward await_reply_target; advance if met. // (await_reply_target counts "how many reply frames to receive // before advancing" — used by the barrier semantics in spec §3.1.) self.await_reply_target = self.await_reply_target.saturating_sub(1); if self.await_reply_target == 0 { self.advance_step(); } } self.push_capture(Capture::CallerHeardReply { at: Instant::now() }); Some(frame) } None => { // Empty reply_ring: the reflex muted us (slice-4 §3.2 state machine). // Capture BargeKillObserved — the LatencyProbe pairs this with the // most recent CallerLoudOnset for the kill-time metric. self.push_capture(Capture::BargeKillObserved { at: Instant::now() }); None } } } } impl AudioSink for SimAudioPipe { fn on_pcm_frame(&mut self, _frame: PcmFrame) { // The caller "speaks." The scenario drives here: each on_pcm_frame // call advances the current step's counter. if self.step_idx >= self.scenario.steps.len() { return; // post-End; no-op. } let advance = match &self.scenario.steps[self.step_idx] { ScenarioStep::SpeakLoud { .. } | ScenarioStep::SpeakQuiet { .. } | ScenarioStep::Pause { .. } => { self.step_frames_remaining = self.step_frames_remaining.saturating_sub(1); self.step_frames_remaining == 0 } ScenarioStep::AwaitReply { .. } => false, // await_reply advances via next_pcm_frame ScenarioStep::End => false, }; if advance { self.advance_step(); } } } impl AudioPipe for SimAudioPipe { fn clear_playout_ring(&mut self) { self.reply_ring.clear(); } fn barge_in_flush(&mut self) { self.reply_ring.clear(); } } ``` - [ ] **Step 4: Run the test to verify it passes** ```bash cargo test -p rutster-sim --lib sim_audio_pipe::tests cargo test -p rutster-sim --lib ``` Expected: PASS (S1's 4 tests + S2's 5 tests = 9 tests). - [ ] **Step 5: fmt + clippy + full test + commit** ```bash cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings && cargo test --all git add crates/rutster-sim/src/sim_audio_pipe.rs crates/rutster-sim/src/lib.rs git commit -s -m "feat(sim): SimAudioPipe + Capture enum (slice-4½ S2) The test-double AudioPipe that simulates a caller. Drives a Scenario on on_pcm_frame (sink: caller speaks); receives brain replies on next_pcm_frame (source: caller hears). Both timestamps anchored to Instant::now() inside this pipe — the harness can't lie about latency because the only clock it uses is the caller's (spec §2.2). LatencyProbe (S3) consumes the Capture stream post-run." ``` --- ### Task S3: `LatencyProbe` **Files:** - Modify: `crates/rutster-sim/src/latency.rs` (replace stub) - Test: inline **Interfaces:** - Consumes: `Capture` enum (from S2). - Produces: `pub struct LatencyProbe { captures: Vec }`, with `pub fn from_captures`, `pub fn kill_times`, `pub fn mouth_to_ear_times`, `pub fn p50_kill_ms`, `pub fn p99_kill_ms`, `pub fn p50_mouth_to_ear_ms`, `pub fn p99_mouth_to_ear_ms`. - [ ] **Step 1: Write the failing tests** ```rust #[cfg(test)] mod tests { use super::*; use crate::sim_audio_pipe::Capture; use std::time::{Duration, Instant}; #[test] fn kill_times_empty_for_no_captures() { let p = LatencyProbe::from_captures(vec![]); assert!(p.kill_times().is_empty()); assert!(p.p99_kill_ms().is_nan()); // empty => NaN } #[test] fn kill_times_pairs_onset_with_next_barge_kill() { 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() { 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() { let t0 = Instant::now(); let mut captures = vec![]; for ms in [50u64, 55, 60, 65, 200] { // 4 normal + 1 outlier 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"); } } ``` - [ ] **Step 2: Run to verify fail + Step 3: Implement + Step 4: Verify pass** The implementation mirrors spec §3.3 verbatim. Use `Vec`, pairings done via linear scan with cursor-on-onset. Output: `Vec`. Function skeleton: ```rust //! # LatencyProbe — the post-hoc 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 → first `BargeKillObserved` //! - mouth-to-ear: caller-speech-onset → next `CallerHeardReply` use std::time::{Duration, Instant}; use crate::sim_audio_pipe::Capture; pub struct LatencyProbe { captures: Vec, } impl LatencyProbe { pub fn from_captures(captures: Vec) -> Self { Self { captures } } /// Barge-in kill-times: pair each `CallerLoudOnset` with the next /// `BargeKillObserved` thereafter. Per-call measurement. pub fn kill_times(&self) -> Vec { let mut out = vec![]; let mut last_onset: Option = 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() { out.push(at.saturating_duration_since(on)); } // (Else: kill without prior onset — ignore; spray noise.) } Capture::CallerHeardReply { .. } => { /* irrelevant to kill metric */ } } } out } pub fn mouth_to_ear_times(&self) -> Vec { let mut out = vec![]; let mut last_onset: Option = 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 m2e */ } } } 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) } } fn percentile_ms(durations: &[Duration], pct: u8) -> f64 { if durations.is_empty() { return f64::NAN; } let mut sorted: Vec = 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; sorted[idx] as f64 } ``` - [ ] **Step 5: fmt + clippy + test + commit** ```bash cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings && cargo test --all git add crates/rutster-sim/src/latency.rs git commit -s -m "feat(sim): LatencyProbe — p50/p99 kill + mouth-to-ear (slice-4½ S3) Pairs Capture::CallerLoudOnset with the next BargeKillObserved (kill-time) and the next CallerHeardReply (mouth-to-ear). Outputs are Duration vectors; p50/p99 helpers compute on the captured sample. The threshold assertions in S7 read p99_kill_ms + p99_mouth_to_ear_ms." ``` --- ### Task S4: `SimCall` + `ScenarioRunner` **Files:** - Modify: `crates/rutster-sim/src/runner.rs` (replace stub) - Test: inline **Interfaces:** - Consumes: `SimAudioPipe` (S2) + `MediaThread` / `MediaCmd` (merged slice-4 + slice-5/seams). - Produces: `pub struct SimCall { pipe, media_cmd_tx, ... }` + `impl SimCall { pub async fn run(self) -> LatencyProbe }` + `pub struct ScenarioRunner { ... }`. **Approach:** A `SimCall` registers a session via `MediaCmd::RegisterSim { pipe, reply }` against the existing MediaThread (the harness DOES NOT stand up its own — it consumes the binary's MediaThread). Then it drives the scenario + captures reply frames. The exact driving mechanism: a tokio task that, per the SimAudioPipe's internal cursor, mpsc-pumps synthetic caller PCM into the session's `on_pcm_frame` AND drains the session's `next_pcm_frame` output back into the SimAudioPipe's `push_reply` channel. Wait — this is a wiring question that touches the binary-side wiring of the SimAudioPipe into MediaThread::RegisterSim. The spec §3.5 says the harness sends `Box` via the MediaCmd; the MediaThread inserts it as a synthetic session. The session's tick-driving happens by the same `loop_driver::drive` machinery — but loop_driver::drive expects an RtcSession (str0m). So the harness has to provide its OWN tick-driving path for the SimPipe. Hmm, this is the same fork the step-5 spec surfaced (§4.1 of step-5 spec) but for slice 4½. Let me revisit: spec §3.5 says "The std thread's `run_media_thread` handles `RegisterSim` by inserting a synthetic 'session' entry that drives the harness's `SimAudioPipe` through the same `loop_driver::drive` calls as real WebRTC sessions." But loop_driver::drive takes a `&mut RtcSession`, NOT a `&mut dyn AudioPipe`. So either the spec is wrong, OR `RegisterSim` needs its own dispatch path. The cleanest fix (parallel to step-5's `MediaLeg` enum solution): introduce `MediaCmd::RegisterSim` that wraps the SimAudioPipe in a SimSession struct whose `tick(now)` method is `pipe.on_pcm_frame(synthetic_caller_frame); let _ = pipe.next_pcm_frame();` — but the synthetic caller frame + reply routing is the SimCall's responsibility, not the SimSession's. Actually, the simplest approach the spec implies: `RegisterSim` is a TEST-only path. It doesn't run through the str0m/loop_driver machinery at all — the MediaThread inserts the SimPipe into a SEPARATE list (not the same HashMap) and ticks it via a direct call to `pipe.on_pcm_frame()` + `pipe.next_pcm_frame()` per meta-tick. But the spec's intent is "the harness drives the SAME media-leg path real callers use" — meaning the slice-4 Reflex<...> wrapper stack + the TapEngine spawn. To get THAT, the SimPipe needs to plug into the session mapleton as a Box just like TapAudioPipe does in slice-4 Task 6's composition site. This is too complex to resolve cleanly in a tactical plan revision. The pragmatic answer: the SimCall wires it SELF — it constructs the TapAudioPipe + Reflex<...> + LocalVadReflex<...> composition directly (REUSING slice-4's stack composition code from Task 6, just outside the MediaThread), drives the wrapped pipe manually via a tight loop, captures brain replies via mpsc from the TapEngine. This means `MediaCmd::RegisterSim` is NOT needed. The SimCall runs entirely in tokio: spawns its own TapEngine against the MockRealtimeBrain URL, wraps the TapAudioPipe in the Reflex stack (REUSED), drives the SimAudioPipe's on_pcm_frame (caller speech) by submitting frames to the inner tap pipe's tx_pcm_in, and feeds brain replies back via the tap pipe's tx_audio_out drain. This is simpler than RegisterSim — no MediaThread extension required, no seam change. Update the Spec §3.5 (and §1.1 mentions of RegisterSim) to reflect this decision. For the plan, the harness wires itself standalone. - [ ] **Step 1: Read existing slice-4 Task 6 spawn-composition pattern** Read `crates/rutster/src/media_thread.rs` for the existing spawn_tap_engine composition site (the `Connected` transition that wraps TapAudioPipe in Reflex + LocalVadReflex). The SimCall mirrors this composition but in tokio, not on the std thread. - [ ] **Step 2: Implement `SimCall` + `ScenarioRunner`** Pseudocode for the wiring: ```rust //! # SimCall — one synthetic caller through the FOB reflex loop //! //! See spec §3.4. The SimCall stands up its own TapEngine (REUSED from //! slice-2) + the slice-4 Reflex<...> composition in tokio, then drives //! a `SimAudioPipe` against it. The measurement captures live inside //! the SimAudioPipe (the caller's clock — spec §2.2). use std::time::{Duration, Instant}; use tokio::sync::mpsc; use rutster_media::{PcmFrame, Reflex, LocalVadReflex, ReflexMetrics, AdvisoryEvent}; use rutster_tap::TapAudioPipe; use crate::scenario::Scenario; use crate::sim_audio_pipe::SimAudioPipe; use crate::latency::LatencyProbe; pub struct SimCall { scenario: Scenario, /// The brain's WS URL (e.g. MockRealtimeBrain URL or a real brain WS). brain_url: url::Url, } impl SimCall { pub fn new(scenario: Scenario, brain_url: url::Url) -> Self { Self { scenario, brain_url } } /// Drive the scenario against the FOB reflex loop. Returns the /// LatencyProbe with the captured timeline. pub async fn run(self) -> LatencyProbe { // 1. Construct the TapAudioPipe + spawn_tap_engine against brain_url. // (Reuses slice-2/slice-4's spawn_tap_engine wiring; capture the // returned TapConn; the advisory_tx/Reflex stack wires as slice-4 // Task 5 + Task 6 composition.) let (tap_pipe, tap_conn) /* = spawn_tap_engine(...) */; let (advisory_tx, advisory_rx) = mpsc::channel::(16); let metrics = ReflexMetrics::new(); let reflex = Reflex::new(tap_pipe, advisory_rx, metrics.clone()); let wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx); // 2. The SimAudioPipe drives the scenario's caller side; the wrapped_pipe // is the FOB-brain side they interact with. let mut sim_pipe = SimAudioPipe::new(self.scenario.clone(), 16); // 3. Drive loop: per 20 ms tick: // a. If the SimPipe's scenario says "speak loud" → call wrapped_pipe // .on_pcm_frame(PcmFrame::loud()) — that hits LocalVadReflex → // Reflex → TapAudioPipe → tx_pcm_in → brain WS. // b. Pull the wrapped_pipe's next_pcm_frame() output → push to // sim_pipe.push_reply(frame) — that's the brain's reply landing // at the caller's ear (the SimPipe's source path). // c. (Sink path: the wrapped_pipe itself produces PCM frames the // TapEngine sends to the brain. The SimPipe's reply_ring is // filled manually with what the wrapped_pipe produces — that's // the simulation of "what the FOB's media loop would have // returned to the WebRTC peer.") let tick = Duration::from_millis(20); loop { // Drive the sink: emit the next caller frame. // (The SimPipe's on_pcm_frame is what determines caller behavior.) // We bypass calling SimPipe::on_pcm_frame in this loop — // its on_pcm_frame is the SINK path consumed INTERNALLY when // the wrapped_pipe's on_pcm_frame routes back into it. // // Per spec §3.2.1: sim_pipe.on_pcm_frame is for the harness to // feed caller-side signals. We use it to read scenario state: // the SimPipe emits CallerLoudOnset on step transition; we don't // need to actually push PcmFrames into it because its job is to // TIME the scenario steps + capture timestamps. // For "speak loud" step: emit a loud PcmFrame to the wrapped_pipe. if sim_pipe.current_step_is_speak_loud() { wrapped_pipe.on_pcm_frame(PcmFrame::loud()); } // Pull the FOB's outbound frame (if any) + feed back to the // SimPipe's reply_ring. if let Some(reply) = wrapped_pipe.next_pcm_frame() { sim_pipe.push_reply(reply); } // Drive the SimPipe's internal state advance (e.g. countdown // step frames). sim_pipe.tick(); // Termination: scenario reached End. if sim_pipe.scenario_done() { break; } tokio::time::sleep(tick).await; } // 4. Tear down + return the LatencyProbe. let _ = tap_conn.close_tx.send(()); tap_conn.join.abort(); let captures = sim_pipe.take_captures(); LatencyProbe::from_captures(captures) } } pub struct ScenarioRunner { brain_url: url::Url, } impl ScenarioRunner { pub fn new(brain_url: url::Url) -> Self { Self { brain_url } } pub async fn run(&self, scenario: Scenario) -> LatencyProbe { SimCall::new(scenario, self.brain_url.clone()).run().await } } ``` (The exact shape — particularly how the SimAudioPipe + the wrapped_slice-4 Reflex stack share PcmFrames — needs more design than I can resolve in this plan revision. The dev-a implementing this should: (a) trade off details once they read the slice-4 `media_thread.rs` Connected spawn seam, (b) emit a STATUS UPDATE if they hit a design fork worth PM input. The KEY invariant is: captured `Instant::now()` timestamps in SimAudioPipe, slicing through the harness, NOT in the FOB itself.) - [ ] **Step 3: Write tests asserting the SimCall drives a scenario to completion against an in-process MockRealtimeBrain + measures latency** ```rust #[tokio::test] async fn sim_call_completes_trivial_scenario_against_mock_brain() { // Stand up MockRealtimeBrain (slice-3 merged). let mock = MockRealtimeBrain::start().await.unwrap(); let scenario = Scenario::from_toml(r#" name = "trivial" [[steps]] kind = "speak_loud" frames = 3 [[steps]] kind = "end" "#).unwrap(); let probe = SimCall::new(scenario, mock.url()).run().await; // Probe should have at least one kill_time capture. let kills = probe.kill_times(); assert!(!kills.is_empty(), "expected barge-in to fire on 3 loud frames"); } ``` - [ ] **Step 4: fmt + clippy + test + commit** ```bash cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings && cargo test --all git add crates/rutster-sim/src/runner.rs git commit -s -m "feat(sim): SimCall + ScenarioRunner — drives scenario against FOB reflex loop (slice-4½ S4) SimCall stands up its own TapEngine + composes slice-4's Reflex + LocalVadReflex stack against it. Drives a SimAudioPipe through the scenario: emit loud/quiet PcmFrames into the wrapped pipe's sink; pull next_pcm_frame outputs + push into the SimAudioPipe's reply_ring. The captured Instant::now() timestamps in the SimAudioPipe are the caller's clock (spec §2.2) — the harness can't lie about latency." ``` --- ### Task S5: `ConcurrencyRunner` **Files:** - Modify: `crates/rutster-sim/src/concurrency.rs` - Test: inline **Interfaces:** - Consumes: `SimCall` (S4) + `LatencyProbe` (S3). - Produces: `pub struct ConcurrencyRunner { ... }` with `pub async fn run(&self, scenario) -> SweepReport` + `pub struct SweepReport { per_concurrency: Vec }` + `pub struct PerConcurrencyReport { ... }`. - [ ] **Step 1: Implement the sweep driver** For each N ∈ `[1, 10, 50]`: - Spawn N `SimCall`s against the SAME MockRealtimeBrain URL (or N mock brains — TBD; prefer ONE mock brain for determinism; the mock is designed to handle multiple WS clients in its accept loop). - `tokio::join!`-style await all. - Aggregate the per-call `LatencyProbe` results into a `PerConcurrencyReport`: - Collect all kill_times + mouth_to_ear_times across all N calls. - Compute p50/p99 over the merged sample. - Return the `SweepReport`. Tick-lag gauge: S6 attaches; for S5 alone, the `PerConcurrencyReport` leaves `max_tick_lag_micros` + `tick_overrun_pct` as 0 (the S6 integration fills them in). - [ ] **Step 2: Tests** — `concurrency_run_at_1_produces_report`, `concurrency_run_at_10_reports_10_calls`. - [ ] **Step 3: fmt + clippy + test + commit** — message `feat(sim): ConcurrencyRunner — N concurrent SimCalls + SweepReport aggregation (slice-4½ S5)`. --- ### Task S6: `TickLagGauge` **Files:** - Modify: `crates/rutster-sim/src/tick_lag.rs` - Test: inline **Interfaces:** - Consumes: slice-5/seams `MediaCmd::Stats` (returns `MediaStats { tick_overruns, last_tick_micros }`). - Produces: `pub struct TickLagGauge { stats_tx: mpsc::Sender, samples: Vec }` + `pub struct TickLagSample { at: Instant, last_tick_micros: u64, tick_overruns_cumulative: u64 }` + `pub fn poll_periodically(&mut self, period: Duration, stop_rx: oneshot::Receiver<()>)`. - [ ] **Step 1: Implement the gauge** The poll loop: spawn a tokio task that, every `period` (default 1 sec), sends `MediaCmd::Stats` to the binary's MediaThread, awaits the `MediaStats` reply, pushes the sample into `samples`. When the SimCall's run loop completes, signal `stop_rx` to terminate the gauge. After `run()` returns: drain `samples`; the PerConcurrencyReport's `max_tick_lag_micros` = max sample's `last_tick_micros`; `tick_overrun_pct` = the differential between the first and last sample's cumulative overruns, divided by total tick count (computed from the elapsed wallclock duration × 100 ticks/sec). - [ ] **Step 2: Tests** — `gauge_polls_periodically`, `gauge_aggregates_max_tick_lag`, `gauge_computes_overrun_pct`. - [ ] **Step 3: fmt + clippy + test + commit** — message `feat(sim): TickLagGauge — reads slice-5/seams MediaCmd::Stats during sweep (slice-4½ S6)`. --- ### Task S7: `cargo test --features=sim-bench` CI job + threshold constants + assertion tests **Files:** - Modify: `crates/rutster-sim/src/thresholds.rs` - Modify: `.github/workflows/ci.yml` - Test: under `#[cfg(feature = "sim-bench")] #[tokio::test]` in `thresholds.rs` (the CI gate). **Interfaces:** - Consumes: `ConcurrencyRunner` (S5) + `TickLagGauge` (S6). - Produces: - `pub const BARGE_IN_KILL_TIME_P99_MS: f64 = 80.0;` - `pub const MOUTH_TO_EAR_P99_MS: f64 = 700.0;` - `pub const TICK_LAG_MAX_MS: f64 = 10.0;` - `pub const TICK_OVERRUN_PCT_MAX: f64 = 1.0;` - `pub const SWEEP_CONCURRENCIES: &[usize] = &[1, 10, 50];` - `#[cfg(feature = "sim-bench")] #[tokio::test]` tests asserting per-concurrency thresholds. - [ ] **Step 1: Write the threshold consts + the assertion test bodies** ```rust // crates/rutster-sim/src/thresholds.rs //! CI threshold constants + the assertion tests under `--features=sim-bench`. //! See spec §5.1 + §5.5. A regression-failing threshold is a Rust `assert!`, //! not a tracing metric — failure here is a red X on the PR (ADR-0010). pub const BARGE_IN_KILL_TIME_P99_MS: f64 = 80.0; pub const MOUTH_TO_EAR_P99_MS: f64 = 700.0; pub const TICK_LAG_MAX_MS: f64 = 10.0; pub const TICK_OVERRUN_PCT_MAX: f64 = 1.0; pub const SWEEP_CONCURRENCIES: &[usize] = &[1, 10, 50]; #[cfg(all(test, feature = "sim-bench"))] mod bench_assertions { use super::*; use crate::concurrency::ConcurrencyRunner; use crate::scenario::Scenario; #[tokio::test] async fn loud_barge_at_each_concurrency_passes_thresholds() { let scenario = Scenario::load("../scenarios/loud-barge.toml").unwrap(); for &n in SWEEP_CONCURRENCIES { let report = ConcurrencyRunner::in_process(n).run(scenario.clone()).await; let row = report.per_concurrency.iter() .find(|r| r.concurrency == n).expect("concurrency row"); assert!(row.p99_kill_ms <= BARGE_IN_KILL_TIME_P99_MS, "p99 kill-time at N={}: {}ms > {}ms (budget overflow)", n, row.p99_kill_ms, BARGE_IN_KILL_TIME_P99_MS); assert!(row.p99_mouth_to_ear_ms <= MOUTH_TO_EAR_P99_MS, "p99 mouth-to-ear at N={}: {}ms > {}ms", n, row.p99_mouth_to_ear_ms, MOUTH_TO_EAR_P99_MS); assert!((row.max_tick_lag_micros as f64) / 1000.0 <= TICK_LAG_MAX_MS, "max tick-lag at N={}: {}us > {}ms", n, row.max_tick_lag_micros, TICK_LAG_MAX_MS); assert!(row.tick_overrun_pct <= TICK_OVERRUN_PCT_MAX, "tick overrun % at N={}: {}% > {}%", n, row.tick_overrun_pct, TICK_OVERRUN_PCT_MAX); } } #[tokio::test] async fn quiet_advisory_at_1_concurrency_passes_thresholds() { let scenario = Scenario::load("../scenarios/quiet-advisory.toml").unwrap(); let report = ConcurrencyRunner::in_process(1).run(scenario).await; let row = &report.per_concurrency[0]; // Advisory-path: kill-time can be longer than local-VAD (brain round-trip ~300ms). // Use a relaxed ceiling for advisory kills: assert!(row.p99_kill_ms <= 400.0, // ~brain advisory latency + slack "advisory kill-time {}ms > 400ms", row.p99_kill_ms); } #[tokio::test] async fn sustained_call_multibarge_does_not_drift() { let scenario = Scenario::load("../scenarios/sustained-call.toml").unwrap(); let report = ConcurrencyRunner::in_process(1).run(scenario).await; let row = &report.per_concurrency[0]; // 3-barge drift check: the per-barge kill_times should be within 1.5× // of each other (anti-fatigue). // (This test relies on the LatencyProbe surfacing per-barge captures in // sequence; the assertion reads them in temporal order.) // Implementation note: kill_times.len() should be >= 3; the third bar's // kill-time should be <= 1.5× the first's. The exact bound is "drift detect" // — defensible threshold per spec §7.9. } } ``` - [ ] **Step 2: Add the CI job to `.github/workflows/ci.yml`** ```yaml sim-bench: name: sim-bench (stable) runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 - name: Install libopus run: sudo apt-get update && sudo apt-get install -y libopus-dev - name: Run sim-bench threshold sweep run: cargo test --all --features=sim-bench -- --test-threads=1 ``` `--test-threads=1` is load-bearing — concurrent sim-bench tests would contaminate each other's `MediaStats` polling (the tick-lag gauge measures the SHARED media thread; concurrent runs of the sweep would all see each other's load). - [ ] **Step 3: Run locally** `cargo test --all --features=sim-bench -- --test-threads=1` and verify it passes. If a threshold fails: investigate whether it's an actual regression OR a budget too-tight for the CI runner's variance. If the latter: surface as `question` to PM (relay) with the failed number + the proposed adjustment; do NOT just bump it silently. - [ ] **Step 4: fmt + clippy + test + commit**: ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings # note: clippy on sim-bench code paths needs --features=sim-bench cargo clippy --all --all-targets --features=sim-bench -- -D warnings cargo test --all cargo test --all --features=sim-bench -- --test-threads=1 git add crates/rutster-sim/src/thresholds.rs .github/workflows/ci.yml git commit -s -m "ci(sim): sim-bench CI job + threshold consts + assertion tests (slice-4½ S7) cargo test --all --features=sim-bench runs the threshold-assertion sweep; a separate CI job executes it on every PR + nightly. A latency regression fails the build the same way a broken test does (ADR-0010). Single-threaded test execution to avoid MediaStats cross-contamination across concurrent runs." ``` --- ### Task S8: Scenario pack + LEARNING.md pointer (filler task: any time after S4) **Files:** - Create: `crates/rutster-sim/scenarios/loud-barge.toml` - Create: `crates/rutster-sim/scenarios/quiet-advisory.toml` - Create: `crates/rutster-sim/scenarios/sustained-call.toml` - Modify: `LEARNING.md` - [ ] **Step 1: Author the three scenario TOMLs** Each scenario asserts a different property of the FOB reflex loop. Per spec §5.3. - [ ] **Step 2: Add a LEARNING.md pointer** ```markdown ## Slice 4½ (benchmark + simulation harness) To learn how the harness measures latency without lying to itself: - ([`crates/rutster-sim/src/sim_audio_pipe.rs`](crates/rutster-sim/src/sim_audio_pipe.rs)) — the AudioPipe that IS the caller; captures both clocks. - ([`crates/rutster-sim/src/latency.rs`](crates/rutster-sim/src/latency.rs)) — the post-hoc p50/p99 computer. - ([`crates/rutster-sim/src/concurrency.rs`](crates/rutster-sim/src/concurrency.rs)) — the 1/10/50 sweep + doctrine-drift detector for the timing-thread debt. ``` - [ ] **Step 3: fmt + test + commit**: ```bash cargo fmt --all --check && cargo test --all git add crates/rutster-sim/scenarios/ LEARNING.md git commit -s -m "docs(sim): scenario pack + LEARNING.md pointers (slice-4½ S8) Three shipped scenarios assert distinct FOB reflex properties: loud-barge (primary VAD path), quiet-advisory (secondary brain-advisory path), and sustained-call (multi-barge anti-fatigue). LEARNING.md indexes the new crate's measurement-discipline curriculum." ``` --- ## Final acceptance checklist After all 8 tasks merge: - [ ] `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all`, `cargo deny check` all clean (stable + 1.85). - [ ] `cargo test --all --features=sim-bench` clean on stable. - [ ] Seam gate unchanged: `loop_driver.rs` + `rtc_session.rs` byte-identical (CI pinned-blob). - [ ] `cargo doc --no-deps` renders the new `crates/rutster-sim/` modules cleanly. - [ ] All 3 shipped scenarios pass their threshold assertions across 1/10/50 concurrency. - [ ] PR opened via `tea pulls create --head slice-4-half/sim-harness-dev-a --base main --title "slice-4½: benchmark + sim harness — rutster-sim seed + CI-regressed thresholds (S1-S8)" --description ""`. - [ ] Do NOT merge the PR — the maintainer (user) merges after reviewing the sim-bench CI run's numbers on the runner.