fix: cancel and bound plot_ir work

ac-rs/ZMQ.md

@@ -773,11 +773,15 @@ a clean slate (e.g. issuing `transfer_stream` immediately after
 
 **Reply**
 ```json
-{ "ok": true, "stopped": ["<worker-name>", ...] }
+{ "ok": true, "stopped": ["<worker-name>", ...], "stimulus": "silent" }
 ```
 
 `stopped` lists the workers that were actually joined during this call —
-empty if no matching worker was running.
+empty if no matching worker was running. `stimulus: "silent"` is present only
+when no workers remain after the selected handles have joined; cancellable
+stimulus workers have silenced their backend before the reply is constructed.
+It is omitted from a named-stop reply when another worker remains because that
+worker may still be driving output.
 
 **DATA** — after stop, the worker emits a terminal frame:
 ```json
@@ -1101,6 +1105,17 @@ only the default trait impl bails.
 }
 ```
 
+Request budgets are enforced before port resolution or worker spawn:
+
+- `duration`: greater than 0 and at most 60 seconds
+- `tail_s`: from 0 through 60 seconds
+- `n_harmonics`: from 1 through 32
+- `window_len`: from 1 through 1048576 samples
+
+An out-of-budget request returns `ok: false`, confirms that the stimulus is
+silent, and emits no audio. `stop` cancels both the sweep and tail portions of
+an accepted `plot_ir` request.
+
 `level_dbfs` is clamped to the config's `drive_max_dbfs` ceiling (#360) —
 before #360 this was the one command besides `calibrate` that emitted
 whatever was asked for with nothing bounding it.
@@ -1214,6 +1229,17 @@ captures + analyses the loopback. Emits one `measurement/frequency_response/poin
 }
 ```
 
+`duration` must be greater than 0 and at most 60 seconds. The derived sweep
+grid may contain at most 10000 points; the daemon validates the checked point
+count before allocating it or spawning a worker.
+
+Each point's actual capture is floored at `3.0 / freq` seconds (enough cycles
+to analyse a low frequency); since the sweep grid is non-decreasing, `start_hz`
+carries the largest such floor. The daemon also rejects the request,
+before resolving ports or spawning a worker, when `max(duration, 3.0 /
+start_hz)` is non-finite or exceeds the same 60-second ceiling — so a very low
+`start_hz` can be rejected even when `duration` itself is within range.
+
 `level_dbfs` is clamped to the config's `drive_max_dbfs` ceiling (#360).
 
 **Reply**
@@ -1273,6 +1299,9 @@ at each level step. Emits one `measurement/frequency_response/point` frame per l
 }
 ```
 
+`duration` must be greater than 0 and at most 60 seconds. `steps` must be from
+1 through 10000. Both are validated before worker spawn.
+
 `start_dbfs`/`stop_dbfs` are clamped to the config's `drive_max_dbfs`
 ceiling (#360), each computed step individually — a range whose top end
 exceeds the ceiling flattens there rather than running unclamped.

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

@@ -1,10 +1,27 @@
 use crate::client::AcClient;
 
+fn render_success(ack: &serde_json::Value) -> Vec<String> {
+    let mut lines = ack
+        .get("stopped")
+        .and_then(|v| v.as_array())
+        .into_iter()
+        .flatten()
+        .filter_map(|v| v.as_str())
+        .map(|name| format!("  stopped     {name}"))
+        .collect::<Vec<_>>();
+    if let Some(stimulus) = ack.get("stimulus").and_then(|v| v.as_str()) {
+        lines.push(format!("  stimulus    {stimulus}"));
+    }
+    lines
+}
+
 pub fn run(client: &mut AcClient) {
     let ack = client.send_cmd(&serde_json::json!({"cmd": "stop"}), None);
     match ack {
         Some(ref v) if v.get("ok").and_then(|v| v.as_bool()) == Some(true) => {
-            println!("  Stopped.");
+            for line in render_success(v) {
+                println!("{line}");
+            }
         }
         Some(ref v) => {
             let err = v
@@ -18,3 +35,33 @@ pub fn run(client: &mut AcClient) {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn renders_stopped_command_and_confirmed_silence() {
+        assert_eq!(
+            render_success(&json!({
+                "ok": true,
+                "stopped": ["plot_ir"],
+                "stimulus": "silent"
+            })),
+            vec!["  stopped     plot_ir", "  stimulus    silent"]
+        );
+    }
+
+    #[test]
+    fn empty_stop_does_not_invent_a_command() {
+        assert_eq!(
+            render_success(&json!({
+                "ok": true,
+                "stopped": [],
+                "stimulus": "silent"
+            })),
+            vec!["  stimulus    silent"]
+        );
+    }
+}

ac-rs/crates/ac-daemon/src/audio/cpal_backend.rs

@@ -361,6 +361,15 @@ impl AudioEngine for CpalEngine {
     }
 
     fn play_and_capture(&mut self, samples: &[f32], tail_s: f64) -> Result<Vec<f32>> {
+        self.play_and_capture_cancellable(samples, tail_s, &AtomicBool::new(false))
+    }
+
+    fn play_and_capture_cancellable(
+        &mut self,
+        samples: &[f32],
+        tail_s: f64,
+        stop: &AtomicBool,
+    ) -> Result<Vec<f32>> {
         if samples.is_empty() {
             anyhow::bail!("play_and_capture: empty stimulus");
         }
@@ -381,6 +390,11 @@ impl AudioEngine for CpalEngine {
         let mut ok = false;
         loop {
             std::thread::sleep(Duration::from_millis(10));
+            if stop.load(Ordering::Relaxed) {
+                self.state.silence.store(true, Ordering::Relaxed);
+                self.state.one_shot_active.store(false, Ordering::Release);
+                anyhow::bail!("play_and_capture cancelled");
+            }
             if self.state.ring.lock().unwrap().len() >= n_total {
                 ok = true;
                 break;

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

@@ -295,6 +295,27 @@ impl AudioEngine for FakeEngine {
         Ok(out)
     }
 
+    fn play_and_capture_cancellable(
+        &mut self,
+        samples: &[f32],
+        tail_s: f64,
+        stop: &std::sync::atomic::AtomicBool,
+    ) -> Result<Vec<f32>> {
+        let out = self.play_and_capture(samples, tail_s)?;
+        // The on-demand fake backend has no hardware clock. Pace this path
+        // in 10 ms chunks so protocol tests exercise cancellation during
+        // both stimulus and tail rather than completing instantaneously.
+        let chunk = (self.sample_rate as usize / 100).max(1);
+        for _ in (0..out.len()).step_by(chunk) {
+            if stop.load(std::sync::atomic::Ordering::Relaxed) {
+                self.set_silence();
+                anyhow::bail!("play_and_capture cancelled");
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        Ok(out)
+    }
+
     fn capture_stereo(&mut self, duration: f64) -> Result<(Vec<f32>, Vec<f32>)> {
         let n = self.samples_in(duration);
         if let Some(out) = self.ring_capture(n, duration, RingDrain::Stereo) {

ac-rs/crates/ac-daemon/src/audio/jack_backend.rs

@@ -21,11 +21,34 @@ use ringbuf::{HeapCons, HeapProd, HeapRb};
 
 use super::rings::CaptureRings;
 use super::AudioEngine;
+use crate::handlers::MAX_STIMULUS_DURATION_S;
 use ac_core::shared::generator::{generate_pink_noise, generate_sine_1s};
 
-/// 16 s at 192 kHz — comfortably larger than any single capture request.
-/// Fixed at construction so neither thread ever reallocates.
-const RING_CAPACITY: usize = 16 * 192_000;
+/// Capture-ring capacity, in samples, needed to hold the full `plot_ir`
+/// budget (`handlers::MAX_STIMULUS_DURATION_S` applies independently to
+/// both `duration` and `tail_s`, so `play_and_capture_cancellable` can be
+/// asked for up to twice that combined) at a given sample rate.
+///
+/// Before #437's rig verification the ring was a fixed `16 * 192_000` —
+/// "comfortably larger than any single capture request" was true only
+/// because nothing yet validated a request up to the 60 s budget; a rig run
+/// at 96 kHz (`rig-2026-09-14-pr437-plot-budget`, finding 1) showed a
+/// within-budget `plot_ir` request play its full stimulus and then time out
+/// with no IR, because 60 s alone already exceeded that ring's 32 s
+/// capacity at that rate. The fix after that (`120 * 192_000`, sized to the
+/// budget at the project's then-assumed highest rate) reintroduced the same
+/// class of bug one level up: `start()` accepts JACK's actual live sample
+/// rate with no ceiling, and a rig running above 192 kHz (e.g. the 384 kHz
+/// path `ac_core::visualize::mtw::ladder` already exercises) again exceeds
+/// a fixed capacity sized only for the assumed worst case (codex-qa, PR
+/// #437 at 942c0e27). Computing capacity from the *actual* live rate at
+/// `start()` instead of any fixed assumption closes both bugs at once: the
+/// ring always fits the accepted budget, at whatever rate JACK reports. See
+/// `meas_ring_capacity_fits_stimulus_duration_and_tail_budget_at_every_rate`
+/// below.
+fn meas_ring_capacity(sample_rate: u32) -> usize {
+    ((MAX_STIMULUS_DURATION_S * 2.0) * sample_rate as f64).ceil() as usize
+}
 
 /// 4 s at 192 kHz — ref inputs are only used by (multi-pair) transfer_stream
 /// whose `capture_duration(4, sr)` ≈ 2.5 s, so this leaves a comfortable
@@ -288,7 +311,9 @@ impl AudioEngine for JackEngine {
         self.state.silence.store(true, Ordering::Relaxed);
 
         // Split SPSC rings: producer → RT callback, consumer → worker thread.
-        let rb = HeapRb::<f32>::new(RING_CAPACITY);
+        // Sized from the live rate JACK just reported, not a fixed
+        // assumption (#437, codex-qa at 942c0e27).
+        let rb = HeapRb::<f32>::new(meas_ring_capacity(self.sample_rate));
         let (ring_prod, ring_cons) = rb.split();
         self.rings.set_meas(ring_cons);
 
@@ -373,6 +398,15 @@ impl AudioEngine for JackEngine {
     }
 
     fn play_and_capture(&mut self, samples: &[f32], tail_s: f64) -> Result<Vec<f32>> {
+        self.play_and_capture_cancellable(samples, tail_s, &AtomicBool::new(false))
+    }
+
+    fn play_and_capture_cancellable(
+        &mut self,
+        samples: &[f32],
+        tail_s: f64,
+        stop: &AtomicBool,
+    ) -> Result<Vec<f32>> {
         if samples.is_empty() {
             anyhow::bail!("play_and_capture: empty stimulus");
         }
@@ -393,8 +427,22 @@ impl AudioEngine for JackEngine {
         self.state.one_shot_active.store(true, Ordering::Release);
 
         let duration_s = n_total as f64 / sr;
-        let mut waiter = park_waiter(self.state.clone());
-        let wait = waiter(&self.rings, n_total, duration_s + 2.0);
+        let timeout = Instant::now() + Duration::from_secs_f64(duration_s + 2.0);
+        *self.state.waker.lock().unwrap() = Some(std::thread::current());
+        let wait = loop {
+            if stop.load(Ordering::Relaxed) {
+                self.state.silence.store(true, Ordering::Relaxed);
+                break Err(anyhow::anyhow!("play_and_capture cancelled"));
+            }
+            if self.rings.occupied() >= n_total {
+                break Ok(());
+            }
+            if Instant::now() > timeout {
+                break Err(anyhow::anyhow!("capture timeout after {duration_s:.1}s"));
+            }
+            std::thread::park_timeout(Duration::from_millis(10));
+        };
+        *self.state.waker.lock().unwrap() = None;
 
         // Ensure RT stops consuming one-shot even if we bailed early.
         self.state.one_shot_active.store(false, Ordering::Release);
@@ -611,6 +659,32 @@ mod tests {
     use ringbuf::traits::Observer;
     use ringbuf::HeapRb;
 
+    // ---- capture ring vs. protocol budget (#437 rig finding 1; codex-qa
+    // live-sample-rate finding at 942c0e27) ----
+
+    #[test]
+    fn meas_ring_capacity_fits_stimulus_duration_and_tail_budget_at_every_rate() {
+        // `meas_ring_capacity` derives capacity from the *live* rate at
+        // `start()` rather than a fixed assumption, so this isn't a
+        // coupled-constant check against a hardcoded ceiling any more (the
+        // fixed `120 * 192_000` this replaced broke silently above 192 kHz,
+        // e.g. the 384 kHz path `mtw::ladder` already exercises — codex-qa,
+        // PR #437 at 942c0e27). What's still worth asserting is that the
+        // *formula* actually covers the full accepted budget at a rate,
+        // including rates above the old fixed ceiling, rather than trusting
+        // the arithmetic by inspection alone.
+        let max_combined_s = MAX_STIMULUS_DURATION_S * 2.0;
+        for sr in [44_100_u32, 48_000, 96_000, 192_000, 384_000] {
+            let cap = meas_ring_capacity(sr);
+            let n_total = (max_combined_s * sr as f64) as usize;
+            assert!(
+                cap >= n_total,
+                "sr={sr}: {max_combined_s}s combined duration+tail_s needs \
+                 {n_total} samples, meas_ring_capacity returned only {cap}"
+            );
+        }
+    }
+
     // ---- fill_one_shot ----
 
     #[test]

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

@@ -16,6 +16,8 @@ pub mod jack_backend;
 #[cfg(feature = "cpal-audio")]
 pub mod cpal_backend;
 
+use std::sync::atomic::AtomicBool;
+
 use anyhow::{bail, Result};
 
 /// Minimal trait for audio playback + capture, matching Python's JackEngine duck-type contract.
@@ -56,6 +58,20 @@ pub trait AudioEngine: Send + 'static {
         )
     }
 
+    /// Cancellable one-shot playback/capture for stimulus commands whose
+    /// worker can be stopped over CTRL. Implementations must silence output
+    /// before returning a cancellation error. The default preserves source
+    /// compatibility for non-hardware test engines; every shipped backend
+    /// overrides it.
+    fn play_and_capture_cancellable(
+        &mut self,
+        samples: &[f32],
+        tail_s: f64,
+        _stop: &AtomicBool,
+    ) -> Result<Vec<f32>> {
+        self.play_and_capture(samples, tail_s)
+    }
+
     /// Non-blocking drain of up to `max_samples` from the capture ring,
     /// without the pre-clear that `capture_block` performs. Returns whatever
     /// has accumulated since the last call (possibly empty on backends that

ac-rs/crates/ac-daemon/src/handlers/admin.rs

@@ -55,7 +55,7 @@ pub fn stop(state: &ServerState, cmd: &Value) -> Value {
     // receive on the REP socket is guaranteed to see an empty workers map
     // and can start an `Exclusive`-group worker like `transfer_stream`.
     let mut joined: Vec<(String, crate::workers::WorkerHandle)> = Vec::new();
-    {
+    let no_workers_remain = {
         let mut workers = state.workers.lock().unwrap();
         if let Some(name) = target {
             if let Some(w) = workers.get(name) {
@@ -72,10 +72,15 @@ pub fn stop(state: &ServerState, cmd: &Value) -> Value {
                 joined.push((name, handle));
             }
         }
-    }
+        workers.is_empty()
+    };
     let stopped: Vec<String> = joined.iter().map(|(n, _)| n.clone()).collect();
     drop(joined); // runs Drop → joins the worker threads
-    json!({"ok": true, "stopped": stopped})
+    let mut reply = json!({"ok": true, "stopped": stopped});
+    if no_workers_remain {
+        reply["stimulus"] = json!("silent");
+    }
+    reply
 }
 
 pub fn devices(state: &ServerState) -> Value {