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 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 <himself@adlee.work>

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,

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`
+//! # Why this is THE measurement boundary (spec §2.2)
-//! §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`
+//! Both clocks live INSIDE this pipe. The wall-clock the *caller* started
-//! (sink: caller speaks); captures `Instant::now()` at every meaningful
+//! speaking is captured here (we decided when to "speak"); the wall-clock
-//! event for the `LatencyProbe`.
+//! 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<Capture>,
+    /// Pre-allocated reply frames pushed externally by the SimCall wiring
+    /// (S4). The `next_pcm_frame` call pops from here.
+    reply_ring: VecDeque<PcmFrame>,
+}
+
+/// 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<Capture> {
+        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<PcmFrame> {
+        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<P>::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"
+        );
+    }
+}