feat(sim): SimCall + ScenarioRunner -- drives scenario against FOB reflex loop (slice-4½ S4)

The standalone-path SimCall: composes slice-4's Reflex<TapAudioPipe> + outer
LocalVadReflex in tokio (mirrors slice-4 barge_in_integration.rs primary-path
test composition), then drives a SimAudioPipe's scenario on the 20 ms tick.
Captures Instant::now() timestamps inside the SimAudioPipe -- the harness
cannot lie about latency because the only clock it uses is the caller's
(spec section 2.2).

A fake-brain tokio task pushes PcmFrame::zeroed replies to TapAudioPipe's
tx_audio_out channel every 20 ms, mimicking slice-3 MockRealtimeBrain's audio
echo (without the WS server + translator pipeline orchestration cost). This
exercises the mouth-to-ear reply path so the S7 threshold assertions have
non-NaN data to assert against.

S4 fix surfaced by the SimCall driving loop: SimAudioPipe::scenario_done()
now returns true when the cursor enters the End step (was previously only
gtrue past step_idx >= steps.len(); since End's on_pcm_frame is a no-op with
no countdown, the cursor stops advancing on End and the SimCall would loop
forever). Patched in S2's sim_audio_pipe.rs as part of this commit because
S2's unit tests didn't exercise the driving loop.

No MediaCmd::RegisterSim variant added (per kickoff hard rule + plan S4
standalone-path conclusion). The seam files loop_driver.rs + rtc_session.rs
remain byte-identical; media_thread.rs is untouched by slice 4½.

Signed-off-by: Aaron D. Lee <himself@adlee.work>
This commit is contained in:
2026-07-05 03:15:13 -04:00
parent ab3ba9c11d
commit ca30bcd48d
5 changed files with 336 additions and 10 deletions

1
Cargo.lock generated
View File

@@ -1618,6 +1618,7 @@ version = "0.0.0"
dependencies = [
"rutster",
"rutster-media",
"rutster-tap",
"serde",
"thiserror 1.0.69",
"tokio",

View File

@@ -13,6 +13,7 @@ description = "Self-hostable benchmark + simulation harness (ADR-0010 spearhead
[dependencies]
rutster-media = { path = "../rutster-media" }
rutster = { path = "../rutster" }
rutster-tap = { path = "../rutster-tap" }
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
serde = { workspace = true, features = ["derive"] }
toml = { workspace = true }

View File

@@ -59,6 +59,7 @@ pub mod thresholds;
pub mod tick_lag;
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::{

View File

@@ -1,12 +1,326 @@
//! # runner — `SimCall` + `ScenarioRunner`: drive one synthetic caller
//! end-to-end
//! end-to-end through the FOB reflex loop
//!
//! **Stub — lands in S4.**
//! See spec §3.4 + §4.1 for the design + plan Task S4 for the
//! implementation. The SimCall wires itself STANDALONE in tokio (per
//! the plan's S4 standalone-path conclusion): it composes slice-4's
//! `Reflex<TapAudioPipe>` + `LocalVadReflex` stack itself rather than
//! registering a sim session with the binary's `MediaThread` via a new
//! `MediaCmd` variant. The seam files (`loop_driver.rs` +
//! `rtc_session.rs`) stay byte-identical; `media_thread.rs` is
//! untouched by slice 4½.
//!
//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`
//! §3.4 + §4.1 for the design + `docs/superpowers/plans/2026-07-05-slice-4-half-benchmark-sim.md`
//! Task S4 for the implementation. The SimCall wires itself standalone in
//! tokio (no `MediaCmd::RegisterSim` — per the plan's S4 standalone-path
//! conclusion, the File Structure table is stale on this point; S4
//! supersedes). Composes slice-4's `Reflex<TapAudioPipe>` +
//! `LocalVadReflex` stack directly against an in-process `MockRealtimeBrain`.
//! # Why standalone (no `MediaCmd::RegisterSim`)
//!
//! The spec's §3.5 sketches a `MediaCmd::RegisterSim { pipe: Box<dyn
//! AudioPipe>, reply }` variant that would let the harness register a
//! sim session against the binary's `MediaThread`. The plan's S4
//! reasoning concludes this is unnecessary: `loop_driver::drive` expects
//! an `&mut RtcSession` (str0m) — a `&mut dyn AudioPipe` synthetic
//! session wouldn't fit the existing dispatch without either a
//! separate driver-path OR a `MediaLeg` enum wrapper (the step-5
//! approach). Both options change `media_thread.rs` in ways the
//! seam-discipline + the kickoff's hard rule forbid this slice.
//! Simpler: the SimCall composes the `Reflex<TapAudioPipe>` + outer
//! `LocalVadReflex` stack itself in tokio (the same composition site
//! the binary's `Connected` transition performs in slice-4), drives
//! the wrapped stack via direct method calls on the 20 ms tick, and
//! captures `Instant::now()` timestamps inside the `SimAudioPipe`
//! (the caller's clock — spec §2.2). The harness measures the FOB
//! reflex loop's behavior under load without going through the
//! binary's `MediaThread` dispatch.
//!
//! # The fake-brain task (mimics `MockRealtimeBrain`)
//!
//! Spec §8.6 says "in-process measurement against `MockRealtimeBrain`,
//! not client-server against the binary's HTTP surface." The literal
//! composition path (WS `MockRealtimeBrain` + `spawn_tap_engine` + the
//! translator pipeline) is the integration slice-3 + slice-4 already
//! proved. For slice 4½'s threshold assertions, the S4 SimCall mimics
//! the brain side with an in-runtime tokio task that pushes
//! `PcmFrame::zeroed()` replies to `TapAudioPipe`'s `tx_audio_out`
//! channel every 20 ms. This gives the reply-path traffic the
//! mouth-to-ear metric needs (some `CallerHeardReply` captures) without
//! the WS-round-trip orchestration cost. A future slice (post-spearhead
//! refinement) replaces the fake-brain task with the real WS
//! `MockRealtimeBrain` for network-realism latency measurement.
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use rutster_media::{
AdvisoryEvent, AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics,
};
use rutster_tap::{TapAudioPipe, TapMetrics};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use crate::latency::LatencyProbe;
use crate::scenario::Scenario;
use crate::sim_audio_pipe::SimAudioPipe;
/// One synthetic call: a `SimAudioPipe` (the caller-side recorder +
/// scenario driver) + the wiring to drive it against an in-process
/// `Reflex<TapAudioPipe>` + `LocalVadReflex` stack in tokio.
///
/// Single binary; no separate process. The `SimCall::run` method
/// returns a `LatencyProbe` carrying the capture stream for the
/// `ConcurrencyRunner` (S5) to aggregate.
pub struct SimCall {
scenario: Scenario,
}
impl SimCall {
pub fn new(scenario: Scenario) -> Self {
Self { scenario }
}
/// Drive the scenario against the FOB reflex loop. Returns the
/// `LatencyProbe` with the captured timeline.
///
/// # Hot-path policy (per AGENTS.md)
///
/// The 20 ms tick loop is the slice-4½ hot path. Failures here
/// are match-and-continue + observed, never `?`-propagated:
/// `try_send` on a full channel drops + observes; `next_pcm_frame`
/// returning `None` (muted/empty ring) captures a `BargeKillObserved`
/// + continues.
pub async fn run(self) -> LatencyProbe {
// 1. Build the Reflex stack — mirrors slice-4's
// `primary_path_local_vad_kills_playout_without_brain`
// test composition (crates/rutster/tests/barge_in_integration.rs:158):
// the inner pipe is `TapAudioPipe` (the production AudioPipe);
// the inner Reflex drains `AdvisoryEvent`s from a tokio mpsc;
// the outer `LocalVadReflex` is the PRIMARY barge-in trigger
// (slice-4 §3.4 — local RMS/energy VAD with zero brain round-trip).
//
// `tx_pcm_in` would forward caller audio to the brain WS in
// production wiring; here it's owned-but-unused (no brain WS to
// forward to). The `_rx_pcm_in` receiver is dropped (the channel
// fills to its bound of 32, then `TapAudioPipe::on_pcm_frame`'s
// `try_send` drops + observes per the hot-path policy).
let (tx_pcm_in, _rx_pcm_in) = mpsc::channel::<PcmFrame>(32);
let (tx_audio_out, rx_audio_out) = mpsc::channel::<PcmFrame>(32);
let tap_metrics = TapMetrics::new();
let inner_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);
let (advisory_tx, advisory_rx) = mpsc::channel::<AdvisoryEvent>(16);
let reflex_metrics = ReflexMetrics::new();
let reflex = Reflex::new(inner_pipe, advisory_rx, reflex_metrics);
let mut wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx);
// 2. The `SimAudioPipe` — the recorder + scenario driver.
// `SimAudioPipe::new` calls `enter_step` on `steps[0]` immediately,
// capturing `CallerLoudOnset` synchronously if the scenario starts
// with `SpeakLoud` (the loud-barge shape does).
let mut sim_pipe = SimAudioPipe::new(self.scenario.clone(), 16);
// 3. The fake-brain task — a tokio task that periodically pushes
// replies to `tx_audio_out`. Mimics slice-3's `MockRealtimeBrain`
// sending audio_out frames ≈ every 20 ms (the slice-3 mock echoes
// audio back). Exercise the mouth-to-ear path: without brain-side
// traffic, the `Reflex::next_pcm_frame` would always return `None`
// (ring empty), and `mouth_to_ear_times()` would be empty → the
// `p99_mouth_to_ear_ms` assertion in S7 would panic on NaN.
//
// The `AtomicBool` stop flag is the simplest cross-task signal: the
// SimCall's tick loop sets it when scenario_done; the brain task
// reads it on each 20 ms interval. `Arc<AtomicBool>` over a
// `tokio::sync::Notify` because the brain task is a polling loop
// (already sleeping 20 ms each iteration) — the AtomicBool is cheaper
// than a Notify that would need wake-up coordination.
let brain_stop = Arc::new(AtomicBool::new(false));
let brain_stop_clone = brain_stop.clone();
let brain_task: JoinHandle<()> = tokio::spawn(async move {
// Seed the reply ring synchronously so tick 1 (which races
// the brain task's first `interval.tick()` due to async task
// scheduling) has a reply to consume. Without this seed,
// tick 1's `next_pcm_frame` would capture `BargeKillObserved`
// even though the barge hasn't fired (VAD hasn't tripped on
// one loud frame yet) — that capture is noise the
// LatencyProbe would dedup, but the seed keeps the
// measurement timeline clean: the first kill observed
// corresponds to the actual barge.
let _ = tx_audio_out.try_send(PcmFrame::zeroed());
loop {
tokio::time::sleep(Duration::from_millis(20)).await;
if brain_stop_clone.load(Ordering::Relaxed) {
break;
}
// try_send: drop + observe on full channel (hot-path policy).
let _ = tx_audio_out.try_send(PcmFrame::zeroed());
}
});
// 4. The 20 ms tick loop. Each iteration:
// (a) SINK: if the scenario says "speak loud," push a loud frame
// into the wrapped stack — simulating the caller speaking.
// `LocalVadReflex::on_pcm_frame` observes the loud frame's RMS,
// increments `above_threshold_streak`, and after
// `VAD_DEBOUNCE_FRAMES` consecutive loud frames, sends
// `AdvisoryEvent::SpeechStarted` on the advisory channel.
// (b) SOURCE: drain the wrapped stack's `next_pcm_frame` — which
// drains advisories (applying the Reflex state table) + pulls
// brain replies from `TapAudioPipe`'s ring. If `Some`, push
// into the SimPipe's reply ring.
// (c) Drain the SimPipe's reply ring → captures
// `CallerHeardReply` on `Some`; `BargeKillObserved` on `None`
// (the LatencyProbe dedups captures without prior onset).
// (d) Advance the SimPipe's scenario cursor via `on_pcm_frame`.
// (e) Termination: `scenario_done()` checks for `End` step.
let tick = Duration::from_millis(20);
loop {
if sim_pipe.current_step_is_speak_loud() {
wrapped_pipe.on_pcm_frame(loud_pcm_frame());
}
if let Some(reply) = wrapped_pipe.next_pcm_frame() {
sim_pipe.push_reply(reply);
}
// Drain the SimPipe's reply ring → one `CallerHeardReply`
// capture per `Some` (typically one reply per tick, but the
// drain loop handles bursts). Loop exits on `None` — one
// `BargeKillObserved` capture then. The LatencyProbe pairs
// each `CallerLoudOnset` with the next `BargeKillObserved`
// (kill metric) AND the next `CallerHeardReply` (m2e metric)
// independently — both pairs can share the same onset.
while sim_pipe.next_pcm_frame().is_some() {
// drained + captured
}
sim_pipe.on_pcm_frame(PcmFrame::zeroed());
if sim_pipe.scenario_done() {
break;
}
tokio::time::sleep(tick).await;
}
// 5. Cleanup: signal the fake-brain task + await termination.
// Await avoids leaking the task after the SimCall returns
// (otherwise the brain task would race the runtime shutdown +
// could log warnings on test teardown).
brain_stop.store(true, Ordering::Relaxed);
let _ = brain_task.await;
let captures = sim_pipe.take_captures();
LatencyProbe::from_captures(captures)
}
}
/// Construct a loud PcmFrame for the SimCall's sink path. Sample value
/// 1000 — well above `VAD_RMS_THRESHOLD` (500.0) per slice-4 §3.4.
///
/// The same construction pattern appears inline in slice-4's
/// `barge_in_integration.rs`. A `PcmFrame::loud()` factory on
/// `rutster_media::PcmFrame` would centralize this; that's deferred
/// (no public-API churn this slice — the slice-5 trunk slice that
/// also needs loud frames can add the factory).
fn loud_pcm_frame() -> PcmFrame {
let mut f = PcmFrame::zeroed();
for s in f.samples.iter_mut() {
*s = 1000;
}
f
}
/// Single-call driver. A convenience wrapper around `SimCall` that
/// consumes the scenario + returns the `LatencyProbe`. The
/// `ConcurrencyRunner` (S5) constructs `SimCall`s directly per
/// concurrency level rather than going through `ScenarioRunner` — but
/// `ScenarioRunner` is the public API surface for one-off manual
/// measurement.
pub struct ScenarioRunner;
impl ScenarioRunner {
pub fn new() -> Self {
Self
}
pub async fn run(&self, scenario: Scenario) -> LatencyProbe {
SimCall::new(scenario).run().await
}
}
impl Default for ScenarioRunner {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
/// The canonical loud-barge scenario shape (spec §5.3 entry #1):
/// 20 loud frames → barrier-await one reply → end. The await_reply
/// barrier ensures the SimPipe's leave-the-SpeakLoud-step transition
/// happens CLEANLY (with a reply in the ring to consume) rather than
/// racing the barge-in state machine.
fn loud_barge_scenario() -> Scenario {
Scenario::from_toml(
r#"
name = "loud-barge"
[[steps]]
kind = "speak_loud"
frames = 20
[[steps]]
kind = "await_reply"
frames = 0
[[steps]]
kind = "end"
"#,
)
.unwrap()
}
#[tokio::test]
async fn sim_call_drives_loud_barge_scenario_to_completion() {
// The barge must fire: after `VAD_DEBOUNCE_FRAMES` (3) consecutive
// loud frames, the LocalVadReflex trips → sends SpeechStarted →
// the Reflex drains + mutes + flushes the inner ring on the next
// `next_pcm_frame` call → None returned + captured as
// BargeKillObserved → paired by LatencyProbe with the construct-time
// CallerLoudOnset → kill_time sample.
let scenario = loud_barge_scenario();
let probe = SimCall::new(scenario).run().await;
let kills = probe.kill_times();
assert!(
!kills.is_empty(),
"expected barge-in to fire on 20 loud frames (got {} kills)",
kills.len()
);
}
#[tokio::test]
async fn sim_call_short_trivial_scenario_completes() {
// Smoke test: 3 loud frames + End (no barrier). The SimCall must
// terminate cleanly. The `scenario_done()` check is what the
// SimCall's tick loop reads — this test ensures the End-step detection
// works (the S2 SimAudioPipe's `scenario_done` fix surfaced by the
// S4 driving loop: returns true when the cursor enters End step).
let scenario = Scenario::from_toml(
r#"
name = "trivial"
[[steps]]
kind = "speak_loud"
frames = 3
[[steps]]
kind = "end"
"#,
)
.unwrap();
let probe = SimCall::new(scenario).run().await;
// 3 loud frames ≥ VAD_DEBOUNCE_FRAMES (=3), so the VAD trips on
// the 3rd → kill captures should be non-empty.
assert!(
!probe.kill_times().is_empty(),
"expected kill on 3 consecutive loud frames"
);
}
}

View File

@@ -143,8 +143,17 @@ impl SimAudioPipe {
/// True iff the scenario cursor is at end (no more steps to advance).
/// Used by the `SimCall` driver in S4 to terminate its tick loop.
///
/// The `End` step's `on_pcm_frame` is a no-op (no countdown decrement),
/// so checking `step_idx >= steps.len()` alone wouldn't terminate the
/// tick loop — the cursor stops advancing on entering `End`. The done
/// condition is therefore "cursor at `End` step OR past the last step"
/// (covers both the in-end + post-array-bounds cases).
pub fn scenario_done(&self) -> bool {
self.step_idx >= self.scenario.steps.len()
matches!(
self.scenario.steps.get(self.step_idx),
Some(ScenarioStep::End) | None
)
}
/// True iff the current step is `SpeakLoud`. Used by the `SimCall`