//! # sim_audio_pipe — the test-double AudioPipe that simulates a caller
//!
//! 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.
//!
//! # 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<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.
    ///
    /// 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 {
        matches!(
            self.scenario.steps.get(self.step_idx),
            Some(ScenarioStep::End) | None
        )
    }

    /// 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"
        );
    }
}
