fix: plot/plot_level fail atomically on analyzer error, report xrun session delta

ac-rs/ZMQ.md

@@ -1230,12 +1230,30 @@ same applied value.
 // topic: data  (measurement/frequency_response/point frame, see Shared types)
 ```
 
-**DATA** — terminal:
+**DATA** — terminal, success:
 ```json
 // topic: done
 { "cmd": "plot", "n_points": <int>, "xruns": <int> }
 ```
 
+`xruns` is the session delta — `AudioEngine::xruns()` sampled once at
+engine creation and once at sweep completion, then subtracted
+(wrapping-safe) — not a sum of that per-point cumulative reading (#428).
+
+**DATA** — terminal, analyzer failure (#428):
+```json
+// topic: error
+{ "cmd": "plot", "message": "<analyzer error>", "requested_points": <int>, "completed_points": <int> }
+```
+
+An analyzer failure at any point aborts the sweep atomically: the engine
+stops, this `error` is published, and none of `measurement/frequency_
+response/complete`, `measurement/report`, the report file, or `done`
+follow — a failed sweep never archives its completed prefix as a
+successful measurement. `requested_points` is the full sweep's point
+count; `completed_points` is how many points had already published a
+`measurement/frequency_response/point` frame before the failure.
+
 ---
 
 ### `plot_level`
@@ -1270,12 +1288,22 @@ exceeds the ceiling flattens there rather than running unclamped.
 includes `"freq_hz"` and `"drive_db"` fields — `drive_db` is the applied,
 post-clamp level for that step).
 
-**DATA** — terminal:
+**DATA** — terminal, success:
 ```json
 // topic: done
 { "cmd": "plot_level", "n_points": <int>, "xruns": <int> }
 ```
 
+`xruns` is the session delta, same accounting as `plot`'s (#428).
+
+**DATA** — terminal, analyzer failure (#428): same shape and same
+atomic-failure guarantee as `plot`'s, above, with `"cmd": "plot_level"`
+and `requested_points` the level-step count (`steps`).
+```json
+// topic: error
+{ "cmd": "plot_level", "message": "<analyzer error>", "requested_points": <int>, "completed_points": <int> }
+```
+
 ---
 
 ### `monitor_spectrum`
@@ -2963,6 +2991,8 @@ When the guard fires:
 // topic: error
 { "cmd": "<name>", "message": "<exception string>" }
 ```
+`plot`/`plot_level` add `requested_points`/`completed_points` to this
+shape on an analyzer failure (#428) — see their sections above.
 
 ### Unparseable config.json (#370)
 ```json

ac-rs/crates/ac-cli/src/commands/plot.rs

@@ -68,8 +68,8 @@ pub fn run(
         launch_ui(LaunchKind::SweepFreq, cfg, None);
     }
 
-    let results = collect_sweep(client, "plot");
-    if results.is_empty() {
+    let (results, outcome) = collect_sweep(client, "plot");
+    if outcome != SweepOutcome::Done || results.is_empty() {
         return;
     }
     io::print_summary(&results, "DUT", have_cal);
@@ -140,8 +140,8 @@ pub fn run_level(
         launch_ui(LaunchKind::SweepLevel, cfg, None);
     }
 
-    let results = collect_sweep(client, "plot_level");
-    if results.is_empty() {
+    let (results, outcome) = collect_sweep(client, "plot_level");
+    if outcome != SweepOutcome::Done || results.is_empty() {
         return;
     }
     io::print_summary(&results, "DUT", have_cal);
@@ -415,11 +415,34 @@ fn print_ir_notes(report_frame: Option<&serde_json::Value>) {
     }
 }
 
-fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> Vec<serde_json::Value> {
+/// Whether a sweep reached its terminal `done` frame. Anything else — a
+/// terminal `error` (analyzer failure, #428) or a timeout — leaves
+/// `results` holding only a prefix that must never be treated as a
+/// complete artifact: no summary printed, no CSV written.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SweepOutcome {
+    Done,
+    Failed,
+}
+
+fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> (Vec<serde_json::Value>, SweepOutcome) {
+    collect_sweep_frames(|| client.recv_data(300_000), cmd_name)
+}
+
+/// Core of `collect_sweep`, generic over the frame source so the
+/// atomic-failure gating (a terminal `error` must never leave `outcome ==
+/// Done`, however many `measurement/frequency_response/point` frames
+/// preceded it) can be unit-tested without a real `AcClient`/socket —
+/// see `tests::error_after_points_is_failed_not_done` below (#428 QA).
+fn collect_sweep_frames(
+    mut next_frame: impl FnMut() -> Option<(String, serde_json::Value)>,
+    cmd_name: &str,
+) -> (Vec<serde_json::Value>, SweepOutcome) {
     let mut results = Vec::new();
+    let mut outcome = SweepOutcome::Failed;
 
     loop {
-        let frame = match client.recv_data(300_000) {
+        let frame = match next_frame() {
             Some(f) => f,
             None => {
                 eprintln!("\n  error: timeout waiting for {cmd_name} data");
@@ -441,17 +464,27 @@ fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> Vec<serde_json::Value
                     println!("\n  !! {xruns} xrun(s) during {cmd_name}");
                 }
             }
+            outcome = SweepOutcome::Done;
             break;
         } else if topic == "error" {
             let msg = data
                 .get("message")
                 .and_then(|v| v.as_str())
                 .unwrap_or("error");
-            eprintln!("\n  !! {msg}");
+            let partial = match (
+                data.get("requested_points").and_then(|v| v.as_u64()),
+                data.get("completed_points").and_then(|v| v.as_u64()),
+            ) {
+                (Some(requested), Some(completed)) => {
+                    format!(" ({completed} of {requested} points completed; no report written)")
+                }
+                _ => String::new(),
+            };
+            eprintln!("\n  !! {msg}{partial}");
             break;
         }
     }
-    results
+    (results, outcome)
 }
 
 fn save_results(results: &[serde_json::Value], label: &str, cfg: &ac_core::config::Config) {
@@ -462,6 +495,69 @@ fn save_results(results: &[serde_json::Value], label: &str, cfg: &ac_core::confi
     io::save_csv(results, &path);
 }
 
+#[cfg(test)]
+mod tests {
+    use super::{collect_sweep_frames, SweepOutcome};
+    use std::collections::VecDeque;
+
+    fn point(freq_hz: f64) -> serde_json::Value {
+        serde_json::json!({
+            "type": "measurement/frequency_response/point",
+            "freq_hz": freq_hz,
+        })
+    }
+
+    /// PR #451 QA finding (#428): a terminal `error` after some points had
+    /// already streamed must report `SweepOutcome::Failed` and only the
+    /// completed prefix — `run`/`run_level` gate `print_summary`/
+    /// `save_results` on `outcome == Done`, so this is what makes the
+    /// atomic-failure guarantee reach the CLI's own summary/CSV output,
+    /// not just the daemon's wire frames.
+    #[test]
+    fn error_after_points_is_failed_not_done() {
+        let mut frames: VecDeque<(String, serde_json::Value)> = VecDeque::from([
+            ("data".to_string(), point(100.0)),
+            ("data".to_string(), point(200.0)),
+            (
+                "error".to_string(),
+                serde_json::json!({
+                    "cmd": "plot",
+                    "message": "capture at 1000 Hz has 48 samples; minimum is 256",
+                    "requested_points": 5,
+                    "completed_points": 2,
+                }),
+            ),
+            // Must never be reached: a real daemon does not publish a
+            // point or `done` after a terminal `error`, and the loop must
+            // not either.
+            ("done".to_string(), serde_json::json!({"xruns": 0})),
+        ]);
+
+        let (results, outcome) = collect_sweep_frames(|| frames.pop_front(), "plot");
+
+        assert_eq!(outcome, SweepOutcome::Failed);
+        assert_eq!(
+            results.len(),
+            2,
+            "only the pre-failure points should be retained: {results:?}"
+        );
+    }
+
+    #[test]
+    fn done_after_points_is_done() {
+        let mut frames: VecDeque<(String, serde_json::Value)> = VecDeque::from([
+            ("data".to_string(), point(100.0)),
+            ("data".to_string(), point(200.0)),
+            ("done".to_string(), serde_json::json!({"xruns": 0})),
+        ]);
+
+        let (results, outcome) = collect_sweep_frames(|| frames.pop_front(), "plot");
+
+        assert_eq!(outcome, SweepOutcome::Done);
+        assert_eq!(results.len(), 2);
+    }
+}
+
 /// What `launch_ui` should do post-command. The GPU viewer this used to
 /// spawn is gone; `Monitor` now always renders via the
 /// terminal (`monitor_tui`), and the sweep variants just note that no

ac-rs/crates/ac-daemon/src/audio/fake/hooks.rs

@@ -107,11 +107,13 @@ pub(super) fn tau_noise_amplitude_override() -> f32 {
 /// go red under `--fake-audio`.
 ///
 /// `AC_FAKE_XRUNS_OVERRIDE`: comma-separated delta list, one value
-/// consumed per `play_and_capture` call in this process (0-based — same
-/// call indexing as [`TAU_DELAY_CALL_COUNT`] above, so slot *N* of this
-/// list and slot *N* of the delay override line up with the same
-/// `measure_tau_twice` lifecycle). A call past the end of the list adds 0.
-/// Unset ⇒ every call adds 0, byte-identical to today's hardcoded-0 count.
+/// consumed per `play_and_capture` call in this process (0-based: the
+/// first call gets the first value); a call past the end of the list adds
+/// 0. Unset ⇒ every call adds 0, byte-identical to today's hardcoded-0
+/// count. Deliberately scoped to `play_and_capture` alone — sharing this
+/// counter with `capture_block` (below) would shift call indices for
+/// every unrelated calibrate/monitor path that also captures via
+/// `capture_block`, breaking the fixed indexing this doc promises.
 static XRUNS_CALL_COUNT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
 
 fn xruns_override_list() -> &'static [u32] {
@@ -130,3 +132,41 @@ pub(super) fn next_xruns_delta() -> u32 {
     let call_idx = XRUNS_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     xruns_override_list().get(call_idx).copied().unwrap_or(0)
 }
+
+/// Opt-in, fake-only test hook (#428): lets a test drive a `plot`/
+/// `plot_level` sweep's `capture_block` calls across a nonzero xrun count,
+/// independent of [`next_xruns_delta`] above. Without this, a sweep that
+/// only calls `capture_block` (`plot`, `plot_level` — never
+/// `play_and_capture`) has no way to exercise a nonzero session xrun
+/// delta under `--fake-audio`, and the #428 fix (report the delta since
+/// baseline, not a per-point cumulative sum) has no reproduction outside
+/// unit tests.
+///
+/// `AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE`: comma-separated delta list, one
+/// value consumed per `capture_block` call in this process (0-based). A
+/// `plot`/`plot_level` point issues two calls — a discarded 0.1 s warm-up,
+/// then the real capture — so both consume a slot. A call past the end of
+/// the list adds 0. Unset ⇒ every call adds 0, unchanged from before #428.
+static CAPTURE_BLOCK_XRUNS_CALL_COUNT: std::sync::atomic::AtomicUsize =
+    std::sync::atomic::AtomicUsize::new(0);
+
+fn capture_block_xruns_override_list() -> &'static [u32] {
+    static LIST: std::sync::OnceLock<Vec<u32>> = std::sync::OnceLock::new();
+    LIST.get_or_init(|| {
+        std::env::var("AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE")
+            .ok()
+            .map(|s| s.split(',').filter_map(|v| v.trim().parse().ok()).collect())
+            .unwrap_or_default()
+    })
+}
+
+/// Next `capture_block` xrun delta, consuming one slot of the override
+/// list (see [`CAPTURE_BLOCK_XRUNS_CALL_COUNT`] doc above).
+pub(super) fn next_capture_block_xruns_delta() -> u32 {
+    let call_idx =
+        CAPTURE_BLOCK_XRUNS_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    capture_block_xruns_override_list()
+        .get(call_idx)
+        .copied()
+        .unwrap_or(0)
+}

ac-rs/crates/ac-daemon/src/audio/fake/mod.rs

@@ -39,8 +39,8 @@ use anyhow::Result;
 use std::time::Duration;
 
 use self::hooks::{
-    next_loopback_delay_samples, next_xruns_delta, period_size_override, tau_gain_override,
-    tau_noise_amplitude_override,
+    next_capture_block_xruns_delta, next_loopback_delay_samples, next_xruns_delta,
+    period_size_override, tau_gain_override, tau_noise_amplitude_override,
 };
 use self::ring_mode::{FakeRings, RingDrain};
 use self::stimulus::{Stimulus, StimulusGen, Synth};
@@ -234,6 +234,10 @@ impl AudioEngine for FakeEngine {
         if let Some(out) = self.ring_capture(n, duration, RingDrain::Block) {
             return Ok(out?.into_iter().next().unwrap_or_default());
         }
+        // Opt-in xrun injection (#428) — see
+        // `hooks::next_capture_block_xruns_delta`'s doc. Inert (adds 0)
+        // unless `AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE` is set.
+        self.xruns += next_capture_block_xruns_delta();
         std::thread::sleep(Duration::from_secs_f64(duration));
         let port = self.input_port.clone();
         let gain = tau_gain_override();