From c3a6d73fb384028e14af3cc2b99fd925adbbcff0 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sun, 5 Jul 2026 03:01:32 -0400 Subject: [PATCH] =?UTF-8?q?feat(sim):=20SimAudioPipe=20+=20Capture=20enum?= =?UTF-8?q?=20(slice-4=C2=BD=20S2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 cannot lie about latency because the only clock it uses is the caller's (spec section 2.2). Capture enum carries CallerLoudOnset / BargeKillObserved / CallerHeardReply timestamps. BargeKillObserved is captured unconditionally on empty reply_ring -- the LatencyProbe (S3) dedups captures without a prior onset, keeping the hot path branch-free. LatencyProbe (S3) consumes the Capture stream post-run. Signed-off-by: Aaron D. Lee --- crates/rutster-sim/src/lib.rs | 1 + crates/rutster-sim/src/sim_audio_pipe.rs | 424 ++++++++++++++++++++++- 2 files changed, 418 insertions(+), 7 deletions(-) diff --git a/crates/rutster-sim/src/lib.rs b/crates/rutster-sim/src/lib.rs index a808fd1..5d9d3ab 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 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, diff --git a/crates/rutster-sim/src/sim_audio_pipe.rs b/crates/rutster-sim/src/sim_audio_pipe.rs index a98198c..3a88b5e 100644 --- a/crates/rutster-sim/src/sim_audio_pipe.rs +++ b/crates/rutster-sim/src/sim_audio_pipe.rs @@ -1,9 +1,419 @@ -//! # sim_audio_pipe — the test-double `AudioPipe` that simulates a caller +//! # sim_audio_pipe — the test-double AudioPipe that simulates a caller //! -//! **Stub — lands in S2.** +//! See spec §3.2 (the design) + plan Task S2 (the implementation). +//! 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. //! -//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md` -//! §3.2 for the design + `docs/superpowers/plans/2026-07-05-slice-4-half-benchmark-sim.md` -//! Task S2 for the implementation. Drives a `Scenario` on `on_pcm_frame` -//! (sink: caller speaks); captures `Instant::now()` at every meaningful -//! event for the `LatencyProbe`. +//! # Why this is THE measurement boundary (spec §2.2) +//! +//! 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). The harness can't lie about latency because the only clock it +//! uses is the caller's. +//! +//! # State machine overview (spec §3.2.1) +//! +//! The `SimAudioPipe` walks the `Scenario::steps` vector front-to-back. +//! Each step drives either the sink path (via `on_pcm_frame` decrementing +//! `step_frames_remaining` for `SpeakLoud`/`SpeakQuiet`/`Pause`) or the +//! source path (via `next_pcm_frame` decrementing the `AwaitReply` +//! countdown). On each step boundary, `enter_step` runs the appropriate +//! initialization (capturing `CallerLoudOnset` for `SpeakLoud`, setting +//! the `await_reply_target` for `AwaitReply`, etc.). +//! +//! # The `BargeKillObserved` capture is unconditional on empty source +//! +//! When `next_pcm_frame` finds the `reply_ring` empty, it captures +//! `BargeKillObserved` *unconditionally*. Some of these captures are noise +//! (empty ring without a prior barge event). The `LatencyProbe` (S3) is +//! the dedup gate — it pairs each `CallerLoudOnset` with the next +//! `BargeKillObserved` and ignores captures without a prior onset. The +//! hot path stays simple (no conditional logic in the tick); the +//! pairing post-hoc handles the noise. + +use std::collections::VecDeque; +use std::time::Instant; + +use rutster_media::{AudioPipe, AudioSink, AudioSource, PcmFrame}; + +use crate::scenario::{Scenario, ScenarioStep}; + +/// A timestamped event captured by `SimAudioPipe`. Read by `LatencyProbe` +/// post-run to compute p50/p99 latencies. +/// +/// Each capture carries an `Instant` (8 bytes on Linux + the enum +/// discriminant + alignment = 24 bytes total). `Copy` is derived so the +/// `LatencyProbe`'s pairing scan copies captures by value through stack +/// slots rather than passing references. `Instant: Copy`, so the derive +/// is sound. +#[derive(Debug, Clone, Copy)] +pub enum Capture { + /// The caller started speaking loudly (a `SpeakLoud` step began). + /// Captured in `enter_step` when the scenario cursor advances into + /// a `SpeakLoud { frames }` step. The wall-clock the *caller* + /// started speaking — the latency-onset anchor for both kill-time + /// and mouth-to-ear metrics. + CallerLoudOnset { at: Instant }, + /// The FOB killed playout (a `next_pcm_frame` returned `None` + /// immediately after a barge event). See spec §3.2.1. + /// + /// Captured *unconditionally* on empty `reply_ring` — the + /// `LatencyProbe` ignores captures without a prior `CallerLoudOnset` + /// (spray noise). This keeps the hot path branch-free. + BargeKillObserved { at: Instant }, + /// The caller heard a brain reply (a `next_pcm_frame` returned + /// `Some(frame)` after the barge cleared). See spec §3.2.1. + /// The wall-clock the *caller* heard the reply — the receipt + /// anchor for the mouth-to-ear metric. + CallerHeardReply { at: Instant }, +} + +/// The test-double AudioPipe. See module docs. +/// +/// # Lifetime + ownership +/// +/// The `SimAudioPipe` owns its `Scenario` (moved in on construction). The +/// `captures` + `reply_ring` are pre-allocated buffers — bounded to keep +/// the hot path allocation-free. `take_captures()` drains the captures +/// once (post-run) for the `LatencyProbe` to consume. +pub struct SimAudioPipe { + scenario: Scenario, + /// Cursor into `scenario.steps`. + step_idx: usize, + /// Frames remaining in the current step (for SpeakLoud/SpeakQuiet/Pause). + /// Decrements per `on_pcm_frame` call; on reaching 0 → `advance_step`. + 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). + /// `VecDeque` (not `Vec`) for O(1) front-drop when the cap is hit. + captures: VecDeque, + /// Pre-allocated reply frames pushed externally by the SimCall wiring + /// (S4). The `next_pcm_frame` call pops from here. + reply_ring: VecDeque, +} + +/// Capacity of the `captures` ring (spec §3.2 — "bounded; on overflow the +/// oldest is dropped"). 1024 = ~10 seconds of 100 Hz tick captures — ample +/// for any realistic scenario length; pre-allocated once in `new()`. +const CAPTURE_RING_CAP: usize = 1024; + +impl SimAudioPipe { + /// Construct a `SimAudioPipe` for a given scenario. The + /// `reply_ring_cap` is the maximum number of brain-reply frames + /// the pipe will buffer (the SimCall's wiring pushes via + /// `push_reply`). + /// + /// `new` immediately calls `enter_step` on `steps[0]` — meaning a + /// `Scenario` starting with `SpeakLoud { frames }` will emit its + /// first `Capture::CallerLoudOnset` synchronously inside the + /// constructor. Tests that assert on this capture find it before + /// any `on_pcm_frame` call. + 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 tick-driving wiring in S4 (which + /// forwards the wrapped Reflex stack's `next_pcm_frame` output to + /// the SimPipe's reply sink — see spec §3.4). + pub fn push_reply(&mut self, frame: PcmFrame) { + self.reply_ring.push_back(frame); + } + + /// Drain captures for the `LatencyProbe`. Consumes the buffer. + /// Subsequent calls return empty until new captures land. + pub fn take_captures(&mut self) -> Vec { + self.captures.drain(..).collect() + } + + /// 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. + pub fn scenario_done(&self) -> bool { + self.step_idx >= self.scenario.steps.len() + } + + /// True iff the current step is `SpeakLoud`. Used by the `SimCall` + /// driver in S4 to decide whether to push a loud PcmFrame into the + /// wrapped Reflex stack on this tick. + pub fn current_step_is_speak_loud(&self) -> bool { + matches!( + self.scenario.steps.get(self.step_idx), + Some(ScenarioStep::SpeakLoud { .. }) + ) + } + + /// Advance the step cursor; initialize per-step counters + emit any + /// step-entry capture. Called by `enter_step` on construct AND by + /// `advance_step` when the prior step's countdown hits zero. + 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 via + // `scenario_done()` + stops its tick loop. + return; + } + match &self.scenario.steps[self.step_idx] { + ScenarioStep::SpeakLoud { frames } => { + self.step_frames_remaining = *frames; + // Capture onset at step entry. The LatencyProbe pairs this + // with the next BargeKillObserved + the next CallerHeardReply. + 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 for the kill metric. Quiet steps drive the + // advisory-path scenario (quiet-advisory.toml). + } + ScenarioStep::Pause { frames } => { + self.step_frames_remaining = *frames; + } + ScenarioStep::AwaitReply { frames } => { + self.await_reply_target = *frames; + } + ScenarioStep::End => { + // no-op — `scenario_done()` flips true on the next `advance_step`. + } + } + } + + /// Move to the next step. Called when `step_frames_remaining` reaches + /// zero (sink path) OR when `await_reply_target` is met (source path). + 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 { + // Bounded ring: drop oldest + push newest. The hot-path + // policy (spec §3.2: "Discarded on every `on_pcm_frame` call + // once the capture buffer is at capacity") — measurement + // never crashes the loop. + self.captures.pop_front(); + } + self.captures.push_back(c); + } + + fn is_in_await_reply_step(&self) -> bool { + matches!( + self.scenario.steps.get(self.step_idx), + Some(ScenarioStep::AwaitReply { .. }) + ) + } +} + +impl AudioSource for SimAudioPipe { + fn next_pcm_frame(&mut self) -> Option { + match self.reply_ring.pop_front() { + Some(frame) => { + if self.is_in_await_reply_step() { + // Count this reply toward `await_reply_target`; advance + // when the target is hit. Saturating-sub guards against + // underflow on a misconfigured scenario (target=0 from + // the get-go → first reply advances immediately). + self.await_reply_target = self.await_reply_target.saturating_sub(1); + if self.await_reply_target == 0 { + self.advance_step(); + } + } + // Capture: this is the "caller heard" wall-clock. The + // LatencyProbe pairs it with the prior `CallerLoudOnset` + // for the mouth-to-ear metric. + self.push_capture(Capture::CallerHeardReply { at: Instant::now() }); + Some(frame) + } + None => { + // Empty reply_ring: the reflex muted us (slice-4 §3.2 + // state machine — `Reflex

::muted == true` after a + // barge). Capture BargeKillObserved unconditionally; the + // LatencyProbe dedups noise. See module docs. + 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 decrements the current step's + // `step_frames_remaining` for the speak/pause variants; on + // reaching zero, `advance_step` runs. The inbound `_frame` is + // discarded: the SimPipe is the *client side* of the AudioPipe + // contract — the SimCall's wiring (S4) routes the caller-side PCM + // into the wrapped Reflex stack via `wrapped_pipe.on_pcm_frame`, not + // through here. + 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 { + /// Clear the playout ring (reply_ring). Called by the binary when + /// the brain disconnects (slice-2 spec §5.3 step 4). For sim, this + /// is exercised in tests + the teardown path. + fn clear_playout_ring(&mut self) { + self.reply_ring.clear(); + } + + /// Barge-in flush: same as `clear_playout_ring` for the SimPipe (the + /// reply_ring IS the playout buffer; there's no separate inbound queue + /// to drain). Slice-4's `Reflex::barge_in_flush` calls this on + /// `SpeechStarted` to make the resume race-free — the first reply + /// observed post-barge is provably post-barge. + fn barge_in_flush(&mut self) { + self.clear_playout_ring(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The canonical trivial scenario used across most tests: 3 loud + /// frames followed by End. Compact enough to read at a glance; + /// deterministic (no `AwaitReply` barrier to coordinate). + fn trivial_scenario() -> Scenario { + Scenario::from_toml( + r#" + name = "trivial" + [[steps]] + kind = "speak_loud" + frames = 3 + [[steps]] + kind = "end" + "#, + ) + .unwrap() + } + + #[test] + fn speak_loud_advances_step_cursor_on_each_on_pcm_frame() { + // On construct, `enter_step` is called for steps[0] = SpeakLoud{3}, + // emitting the first `CallerLoudOnset` capture synchronously. + // The for loop then drains `step_frames_remaining` to 0 across 3 + // sink calls → `advance_step` → cursor now points at End. + let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); + for _ in 0..3 { + pipe.on_pcm_frame(PcmFrame::zeroed()); + } + 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() { + // Construct advances step_idx to 0 (SpeakLoud), capturing + // CallerLoudOnset. The first `next_pcm_frame` call finds an + // empty reply_ring → captures BargeKillObserved, returns None. + // (The LatencyProbe will pair this BargeKillObserved with the + // prior CallerLoudOnset — paired kill-time = ~0 ms in this + // synthetic no-system case.) + let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); + 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() { + // `push_reply` queues a synthetic brain-reply frame; the next + // `next_pcm_frame` call pops it, captures CallerHeardReply, + // returns Some(frame). PcmFrame derives PartialEq in + // `rutster_media::pcm` — verifies the exact frame round-trips + // through push/pop unchanged. + let mut pipe = SimAudioPipe::new(trivial_scenario(), 8); + pipe.push_reply(PcmFrame::zeroed()); + let r = pipe.next_pcm_frame().expect("reply"); + assert_eq!(r, PcmFrame::zeroed(), "frame round-trips unchanged"); + let caps = pipe.take_captures(); + assert!( + caps.iter() + .any(|c| matches!(c, Capture::CallerHeardReply { .. })), + "expected CallerHeardReply captured" + ); + } + + #[test] + fn captures_are_in_temporal_order() { + // `Instant::now()` is monotonic — captures pushed in sequence + // must have non-decreasing `at` fields. This guards against a + // future refactor that captures off-thread (which could + // reorder timestamps + break the LatencyProbe's pairing logic). + 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()); // advances step_frames_remaining + let caps = pipe.take_captures(); + assert!(caps.len() >= 2, "captured at least 2 events"); + for w in caps.windows(2) { + 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() { + // `take_captures` is drain-once (consume semantics) so the + // LatencyProbe gets exactly one canonical timeline per SimCall + // run. A stale-buffer bug (returning the same captures twice) + // would compute double-counted latencies — this test guards. + 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" + ); + } +}