# 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<dyn AudioPipe>, reply: oneshot::Sender<ChannelId> }` 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<ScenarioStep> }`
  - `pub enum ScenarioStep { SpeakLoud { frames: u32 }, SpeakQuiet { frames: u32 }, Pause { frames: u32 }, AwaitReply { frames: u32 }, End }`
  - `impl Scenario { pub fn load(path: impl AsRef<std::path::Path>) -> Result<Self, ScenarioError> }`
  - `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<ScenarioStep>,
}

/// 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<Path>) -> Result<Self, ScenarioError> {
        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<Self, ScenarioError> {
        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<Capture> }`

- [ ] **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<Capture>,
    /// Pre-allocated reply frames pushed externally by the SimCall wiring
    /// (S4). The next_pcm_frame call pops from here.
    reply_ring: VecDeque<PcmFrame>,
}

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<Capture> {
        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<PcmFrame> {
        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<Capture> }`, 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<Capture>`, pairings done via linear
scan with cursor-on-onset. Output: `Vec<Duration>`.

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<Capture>,
}

impl LatencyProbe {
    pub fn from_captures(captures: Vec<Capture>) -> Self {
        Self { captures }
    }

    /// Barge-in kill-times: pair each `CallerLoudOnset` with the next
    /// `BargeKillObserved` thereafter. Per-call measurement.
    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() {
                        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<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 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<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;
    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<dyn AudioPipe>` 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<ChannelId, RtcSession>) 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<dyn AudioPipe> 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::<AdvisoryEvent>(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<TapAudioPipe> + 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<PerConcurrencyReport> }` + `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<MediaCmd>, samples: Vec<TickLagSample> }` +
  `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 "<full §7 done-criteria from spec>"`.
- [ ] Do NOT merge the PR — the maintainer (user) merges after reviewing the sim-bench CI run's
      numbers on the runner.
