From ca30bcd48dceecfc8bfc3f5a4f28ff15453c71e3 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sun, 5 Jul 2026 03:15:13 -0400 Subject: [PATCH] =?UTF-8?q?feat(sim):=20SimCall=20+=20ScenarioRunner=20--?= =?UTF-8?q?=20drives=20scenario=20against=20FOB=20reflex=20loop=20(slice-4?= =?UTF-8?q?=C2=BD=20S4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The standalone-path SimCall: composes slice-4's Reflex + 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 --- Cargo.lock | 1 + crates/rutster-sim/Cargo.toml | 1 + crates/rutster-sim/src/lib.rs | 1 + crates/rutster-sim/src/runner.rs | 332 ++++++++++++++++++++++- crates/rutster-sim/src/sim_audio_pipe.rs | 11 +- 5 files changed, 336 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1cbddbf..aadfcc2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1618,6 +1618,7 @@ version = "0.0.0" dependencies = [ "rutster", "rutster-media", + "rutster-tap", "serde", "thiserror 1.0.69", "tokio", diff --git a/crates/rutster-sim/Cargo.toml b/crates/rutster-sim/Cargo.toml index a6762e7..541bf8d 100644 --- a/crates/rutster-sim/Cargo.toml +++ b/crates/rutster-sim/Cargo.toml @@ -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 } diff --git a/crates/rutster-sim/src/lib.rs b/crates/rutster-sim/src/lib.rs index be7f979..80b0a55 100644 --- a/crates/rutster-sim/src/lib.rs +++ b/crates/rutster-sim/src/lib.rs @@ -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::{ diff --git a/crates/rutster-sim/src/runner.rs b/crates/rutster-sim/src/runner.rs index 62397a1..b0930df 100644 --- a/crates/rutster-sim/src/runner.rs +++ b/crates/rutster-sim/src/runner.rs @@ -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` + `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` + -//! `LocalVadReflex` stack directly against an in-process `MockRealtimeBrain`. +//! # Why standalone (no `MediaCmd::RegisterSim`) +//! +//! The spec's §3.5 sketches a `MediaCmd::RegisterSim { pipe: Box, 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` + 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` + `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::(32); + let (tx_audio_out, rx_audio_out) = mpsc::channel::(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::(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` 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" + ); + } +} diff --git a/crates/rutster-sim/src/sim_audio_pipe.rs b/crates/rutster-sim/src/sim_audio_pipe.rs index 3a88b5e..5a16333 100644 --- a/crates/rutster-sim/src/sim_audio_pipe.rs +++ b/crates/rutster-sim/src/sim_audio_pipe.rs @@ -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`