diff --git a/ac-rs/ZMQ.md b/ac-rs/ZMQ.md index 7bccafab..c2ddf8d2 100644 --- a/ac-rs/ZMQ.md +++ b/ac-rs/ZMQ.md @@ -773,11 +773,15 @@ a clean slate (e.g. issuing `transfer_stream` immediately after **Reply** ```json -{ "ok": true, "stopped": ["", ...] } +{ "ok": true, "stopped": ["", ...], "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. diff --git a/ac-rs/crates/ac-cli/src/commands/stop.rs b/ac-rs/crates/ac-cli/src/commands/stop.rs index 5864e601..ebed3f53 100644 --- a/ac-rs/crates/ac-cli/src/commands/stop.rs +++ b/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 { + 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::>(); + 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"] + ); + } +} diff --git a/ac-rs/crates/ac-daemon/src/audio/cpal_backend.rs b/ac-rs/crates/ac-daemon/src/audio/cpal_backend.rs index f99e9b18..ba1badbf 100644 --- a/ac-rs/crates/ac-daemon/src/audio/cpal_backend.rs +++ b/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> { + 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> { 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; diff --git a/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs b/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs index 1ea6342b..123e2e8d 100644 --- a/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs +++ b/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> { + 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, Vec)> { let n = self.samples_in(duration); if let Some(out) = self.ring_capture(n, duration, RingDrain::Stereo) { diff --git a/ac-rs/crates/ac-daemon/src/audio/jack_backend.rs b/ac-rs/crates/ac-daemon/src/audio/jack_backend.rs index d60618f4..d56dd500 100644 --- a/ac-rs/crates/ac-daemon/src/audio/jack_backend.rs +++ b/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::::new(RING_CAPACITY); + // Sized from the live rate JACK just reported, not a fixed + // assumption (#437, codex-qa at 942c0e27). + let rb = HeapRb::::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> { + 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> { 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] diff --git a/ac-rs/crates/ac-daemon/src/audio/mod.rs b/ac-rs/crates/ac-daemon/src/audio/mod.rs index 07d56e43..39b5ef5f 100644 --- a/ac-rs/crates/ac-daemon/src/audio/mod.rs +++ b/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> { + 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 diff --git a/ac-rs/crates/ac-daemon/src/handlers/admin.rs b/ac-rs/crates/ac-daemon/src/handlers/admin.rs index fa62cd7a..e6a7da6f 100644 --- a/ac-rs/crates/ac-daemon/src/handlers/admin.rs +++ b/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 = 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 { diff --git a/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs b/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs index 9b1d7326..2d5f35ed 100644 --- a/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs +++ b/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs @@ -27,9 +27,87 @@ use crate::server::ServerState; use super::super::{ apply_drive_ceiling, busy_guard, cal_guard, cfg_guard, make_engine_for_state, resolve_input, resolve_output, send_pub, snapshot_from_cal, spawn_worker, sweep_point_frame, Tier1Ctx, + MAX_IR_HARMONICS, MAX_IR_WINDOW_SAMPLES, MAX_STIMULUS_DURATION_S, MAX_SWEEP_POINTS, }; use crate::handlers::mic; +fn request_error(cmd: &str, message: impl std::fmt::Display) -> Value { + json!({ + "ok": false, + "error": format!("{cmd} not started — {message}\n stimulus silent"), + }) +} + +fn point_budget_error(cmd: &str, count: usize) -> Value { + json!({ + "ok": false, + "error": format!( + "{cmd} not started — request expands to {count} points\n maximum {MAX_SWEEP_POINTS} points\n stimulus silent" + ), + }) +} + +fn bounded_duration( + request: &Value, + field: &str, + default: f64, + zero_allowed: bool, + cmd: &str, +) -> Result { + let Some(raw) = request.get(field) else { + return Ok(default); + }; + let Some(value) = raw.as_f64() else { + return Err(request_error( + cmd, + format!("{field} must be a finite number"), + )); + }; + if !value.is_finite() || value < 0.0 || (!zero_allowed && value == 0.0) { + let lower = if zero_allowed { + "at least 0" + } else { + "greater than 0" + }; + return Err(request_error( + cmd, + format!("{field} must be finite and {lower} seconds"), + )); + } + if value > MAX_STIMULUS_DURATION_S { + return Err(request_error( + cmd, + format!("{field} {value:.3} s exceeds {MAX_STIMULUS_DURATION_S:.3} s maximum"), + )); + } + Ok(value) +} + +fn bounded_usize( + request: &Value, + field: &str, + default: usize, + min: usize, + max: usize, + cmd: &str, +) -> Result { + let Some(raw) = request.get(field) else { + return Ok(default); + }; + let Some(value) = raw.as_u64() else { + return Err(request_error(cmd, format!("{field} must be an integer"))); + }; + let value = usize::try_from(value) + .map_err(|_| request_error(cmd, format!("{field} overflows usize")))?; + if !(min..=max).contains(&value) { + return Err(request_error( + cmd, + format!("{field} {value} is outside {min}–{max}"), + )); + } + Ok(value) +} + pub fn plot(state: &ServerState, cmd: &Value) -> Value { busy_guard!(state, "plot"); cfg_guard!(state); @@ -42,8 +120,40 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { .get("level_dbfs") .and_then(Value::as_f64) .unwrap_or(-10.0); - let ppd = cmd.get("ppd").and_then(Value::as_u64).unwrap_or(10) as usize; - let duration = cmd.get("duration").and_then(Value::as_f64).unwrap_or(1.0); + let ppd = match bounded_usize(cmd, "ppd", 10, 1, usize::MAX, "plot") { + Ok(v) => v, + Err(e) => return e, + }; + let duration = match bounded_duration(cmd, "duration", 1.0, false, "plot") { + Ok(v) => v, + Err(e) => return e, + }; + let n_points = match super::super::checked_log_freq_point_count(start_hz, stop_hz, ppd) { + Ok(n) if n <= MAX_SWEEP_POINTS => n, + Ok(n) => return point_budget_error("plot", n), + Err(e) => return request_error("plot", e), + }; + // #437 codex-qa: the worker floors each point's capture at `3.0 / + // freq` so low frequencies get enough cycles to analyse — but the log + // grid is non-decreasing (`checked_log_freq_point_count` above already + // refused `stop < start`), so `start_hz` is always the smallest point + // and therefore the largest `3.0 / freq` floor. Reject here, before + // ports are resolved or the worker spawned, whenever that floor alone + // would blow the same ceiling `duration` is already bounded by — + // otherwise a tiny `start_hz` bypasses the duration budget entirely + // and can hand the backend a non-finite `Duration`. + let max_point_duration = f64::max(duration, 3.0 / start_hz); + if !max_point_duration.is_finite() || max_point_duration > MAX_STIMULUS_DURATION_S { + // Scientific notation: `start_hz` near the low end of this check + // (e.g. the codex-qa repro's `1e-300`) makes `3.0 / start_hz` a + // several-hundred-digit decimal in fixed-point form. + return request_error( + "plot", + format!( + "start_hz {start_hz:e} forces a per-point duration of {max_point_duration:e} s (max(duration, 3/start_hz)), exceeding {MAX_STIMULUS_DURATION_S:.3} s maximum" + ), + ); + } let bpo = cmd.get("bpo").and_then(Value::as_u64).map(|v| v as usize); let cfg = state.cfg.lock().unwrap().clone(); // #360: `plot` puts a stimulus on a physical output, so it is clamped @@ -94,6 +204,7 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { let mic_curve_opt = cal.as_ref().and_then(|c| c.mic_response.clone()); let spl_offset = cal.as_ref().and_then(Calibration::spl_offset_db); let freqs = super::super::log_freq_points(start_hz, stop_hz, ppd); + debug_assert!(freqs.len() <= n_points); let amplitude = ac_core::shared::generator::dbfs_to_amplitude(level_dbfs); if let Err(e) = eng.start(&[out_port], Some(&in_port)) { @@ -341,8 +452,14 @@ pub fn plot_level(state: &ServerState, cmd: &Value) -> Value { .and_then(Value::as_f64) .unwrap_or(-40.0); let stop_dbfs = cmd.get("stop_dbfs").and_then(Value::as_f64).unwrap_or(0.0); - let steps = cmd.get("steps").and_then(Value::as_u64).unwrap_or(26) as usize; - let duration = cmd.get("duration").and_then(Value::as_f64).unwrap_or(1.0); + let steps = match bounded_usize(cmd, "steps", 26, 1, MAX_SWEEP_POINTS, "plot_level") { + Ok(v) => v, + Err(e) => return e, + }; + let duration = match bounded_duration(cmd, "duration", 1.0, false, "plot_level") { + Ok(v) => v, + Err(e) => return e, + }; let cfg = state.cfg.lock().unwrap().clone(); let ceiling = cfg.drive_max_dbfs; @@ -667,13 +784,22 @@ pub fn plot_ir(state: &ServerState, cmd: &Value) -> Value { cfg_guard!(state); let f1_hz = cmd.get("f1_hz").and_then(Value::as_f64).unwrap_or(20.0); let f2_hz = cmd.get("f2_hz").and_then(Value::as_f64).unwrap_or(20_000.0); - let duration = cmd.get("duration").and_then(Value::as_f64).unwrap_or(1.0); + let duration = match bounded_duration(cmd, "duration", 1.0, false, "plot_ir") { + Ok(v) => v, + Err(e) => return e, + }; let level_dbfs = cmd .get("level_dbfs") .and_then(Value::as_f64) .unwrap_or(-6.0); - let tail_s = cmd.get("tail_s").and_then(Value::as_f64).unwrap_or(0.5); - let n_harmonics = cmd.get("n_harmonics").and_then(Value::as_u64).unwrap_or(5) as usize; + let tail_s = match bounded_duration(cmd, "tail_s", 0.5, true, "plot_ir") { + Ok(v) => v, + Err(e) => return e, + }; + let n_harmonics = match bounded_usize(cmd, "n_harmonics", 5, 1, MAX_IR_HARMONICS, "plot_ir") { + Ok(v) => v, + Err(e) => return e, + }; // 4096 is a request, not a promise: `extract_irs` clamps each order's // gate down to the spacing of its own nearest neighbour, so the linear // IR keeps the full 4096 (its neighbour, order 2, sits ~4816 samples @@ -681,10 +807,11 @@ pub fn plot_ir(state: &ServerState, cmd: &Value) -> Value { // 1999 / 1551 / 1551. Those lengths are not silent — they ride out in // the `measurement/impulse_response` envelope and, when any order was // shortened, in the report notes. See issue #278. - let window_len = cmd - .get("window_len") - .and_then(Value::as_u64) - .unwrap_or(4096) as usize; + let window_len = + match bounded_usize(cmd, "window_len", 4096, 1, MAX_IR_WINDOW_SAMPLES, "plot_ir") { + Ok(v) => v, + Err(e) => return e, + }; let cfg = state.cfg.lock().unwrap().clone(); // #360: `plot_ir` had no clamp at all — the module doc on @@ -718,7 +845,7 @@ pub fn plot_ir(state: &ServerState, cmd: &Value) -> Value { }; let backend = eng.backend_name(); - let worker = spawn_worker(state, "plot_ir", move |_stop| { + let worker = spawn_worker(state, "plot_ir", move |stop| { // Calibration snapshot. The linear IR itself is never mic-curve // corrected — arrival estimation and gating (`extract_irs`, // `gated_frequency_response` below) run on the raw, uncorrected @@ -777,8 +904,12 @@ pub fn plot_ir(state: &ServerState, cmd: &Value) -> Value { let amp = ac_core::shared::generator::dbfs_to_amplitude(level_dbfs) as f32; let scaled: Vec = sweep.iter().map(|&s| s * amp).collect(); - let captured = match eng.play_and_capture(&scaled, tail_s) { + let capture = eng.play_and_capture_cancellable(&scaled, tail_s, &stop); + eng.set_silence(); + eng.stop(); + let captured = match capture { Ok(c) => c, + Err(_) if stop.load(Ordering::Relaxed) => return, Err(e) => { send_pub( &pub_tx, @@ -1059,3 +1190,56 @@ pub fn plot_ir(state: &ServerState, cmd: &Value) -> Value { } json!({"ok": true, "out_port": out_port_reply, "level_dbfs": level_dbfs, "backend": backend}) } + +#[cfg(test)] +mod request_budget_tests { + use super::*; + + #[test] + fn documented_duration_boundaries_are_accepted() { + assert_eq!( + bounded_duration( + &json!({"duration": MAX_STIMULUS_DURATION_S}), + "duration", + 1.0, + false, + "plot_ir" + ), + Ok(MAX_STIMULUS_DURATION_S) + ); + assert_eq!( + bounded_duration(&json!({"tail_s": 0.0}), "tail_s", 0.5, true, "plot_ir"), + Ok(0.0) + ); + } + + #[test] + fn documented_integer_boundaries_are_accepted() { + assert_eq!( + bounded_usize( + &json!({"n_harmonics": MAX_IR_HARMONICS}), + "n_harmonics", + 5, + 1, + MAX_IR_HARMONICS, + "plot_ir" + ), + Ok(MAX_IR_HARMONICS) + ); + assert_eq!( + bounded_usize( + &json!({"window_len": MAX_IR_WINDOW_SAMPLES}), + "window_len", + 4096, + 1, + MAX_IR_WINDOW_SAMPLES, + "plot_ir" + ), + Ok(MAX_IR_WINDOW_SAMPLES) + ); + assert_eq!( + super::super::super::checked_log_freq_point_count(100.0, 1000.0, 10_000), + Ok(MAX_SWEEP_POINTS) + ); + } +} diff --git a/ac-rs/crates/ac-daemon/src/handlers/mod.rs b/ac-rs/crates/ac-daemon/src/handlers/mod.rs index 6548a47c..40c4510b 100644 --- a/ac-rs/crates/ac-daemon/src/handlers/mod.rs +++ b/ac-rs/crates/ac-daemon/src/handlers/mod.rs @@ -483,6 +483,29 @@ pub(super) fn send_pub(tx: &crossbeam_channel::Sender>, topic: &str, fra // Sweep math helpers // --------------------------------------------------------------------------- +pub(super) const MAX_STIMULUS_DURATION_S: f64 = 60.0; +pub(super) const MAX_SWEEP_POINTS: usize = 10_000; +pub(super) const MAX_IR_HARMONICS: usize = 32; +pub(super) const MAX_IR_WINDOW_SAMPLES: usize = 1_048_576; + +pub(super) fn checked_log_freq_point_count( + start: f64, + stop: f64, + ppd: usize, +) -> Result { + if !start.is_finite() || !stop.is_finite() || start <= 0.0 || stop < start { + return Err("frequency range must be finite, positive, and non-decreasing".into()); + } + if ppd == 0 { + return Err("ppd must be at least 1".into()); + } + let raw = ((stop / start).log10() * ppd as f64).round(); + if !raw.is_finite() || raw > usize::MAX as f64 { + return Err("derived sweep point count overflows usize".into()); + } + Ok((raw as usize).max(2)) +} + pub(super) fn log_freq_points(start: f64, stop: f64, ppd: usize) -> Vec { let n_decades = (stop / start).log10(); let n_points = (n_decades * ppd as f64).round() as usize; diff --git a/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs b/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs index d8fe712d..17a452b3 100644 --- a/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs +++ b/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs @@ -145,15 +145,113 @@ fn in_range_channels_are_unaffected() { c.call(json!({"cmd":"stop"})); } -/// #428: `duration: 0` makes `plot`'s per-point capture length +#[test] +fn named_stop_does_not_claim_silence_while_output_remains() { + let d = Daemon::spawn(); + let c = Client::new(&d); + + let generate = c.call(json!({ + "cmd": "generate", + "freq_hz": 1000.0, + "level_dbfs": -40.0, + })); + assert_eq!(generate["ok"], json!(true), "generate rejected: {generate}"); + let monitor = c.call(json!({ + "cmd": "monitor_spectrum", + "interval": 0.2, + "fft_n": 8192, + })); + assert_eq!(monitor["ok"], json!(true), "monitor rejected: {monitor}"); + + let stopped_monitor = c.call(json!({"cmd": "stop", "name": "monitor_spectrum"})); + assert_eq!( + stopped_monitor["stopped"], + json!(["monitor_spectrum"]), + "{stopped_monitor}" + ); + assert!( + stopped_monitor.get("stimulus").is_none(), + "generate still drives output, so silence must not be attested: {stopped_monitor}" + ); + + let stopped_generate = c.call(json!({"cmd": "stop", "name": "generate"})); + assert_eq!(stopped_generate["stopped"], json!(["generate"])); + assert_eq!(stopped_generate["stimulus"], json!("silent")); +} + +fn assert_budget_rejection(c: &Client<'_>, request: Value, field: &str) { + let reply = c.call(request); + assert_eq!(reply["ok"], json!(false), "request must fail: {reply}"); + let error = reply["error"].as_str().unwrap_or_default(); + assert!(error.contains(field), "error must name {field}: {error:?}"); + assert!( + error.contains("not started") && error.contains("stimulus silent"), + "rejection must confirm no audio was emitted: {error:?}" + ); + let status = c.call(json!({"cmd": "status"})); + assert_eq!(status["busy"], json!(false), "worker spawned: {status}"); +} + +#[test] +fn plot_family_rejects_resource_budgets_before_spawn() { + let d = Daemon::spawn(); + let c = Client::new(&d); + + assert_budget_rejection( + &c, + json!({"cmd":"plot", "start_hz":100.0, "stop_hz":1000.0, "ppd":10001}), + "point", + ); + assert_budget_rejection(&c, json!({"cmd":"plot", "duration":60.001}), "duration"); + assert_budget_rejection(&c, json!({"cmd":"plot_level", "steps":10001}), "steps"); + assert_budget_rejection(&c, json!({"cmd":"plot_level", "duration":0.0}), "duration"); + assert_budget_rejection(&c, json!({"cmd":"plot_ir", "tail_s":60.001}), "tail_s"); + assert_budget_rejection( + &c, + json!({"cmd":"plot_ir", "n_harmonics":33}), + "n_harmonics", + ); + assert_budget_rejection( + &c, + json!({"cmd":"plot_ir", "window_len":1048577}), + "window_len", + ); + assert_budget_rejection(&c, json!({"cmd":"plot_ir", "duration":"NaN"}), "duration"); + assert_budget_rejection(&c, json!({"cmd":"plot", "ppd":u64::MAX}), "point"); + // #437 codex-qa: `plot`'s worker floors each point's capture at + // `3.0 / freq`, which the request-level `duration` bound alone does + // not cover — a low enough `start_hz` (the smallest point on the + // non-decreasing log grid) must be rejected before spawn even though + // `duration` itself is within budget. + assert_budget_rejection( + &c, + json!({"cmd":"plot", "start_hz":0.001, "stop_hz":0.001, "ppd":1}), + "start_hz", + ); + // The codex-qa repro itself: unrejected, this request used to reach + // the worker and panic converting the resulting `3.0 / freq` seconds + // (well past `Duration`'s representable range) into a `Duration`. + assert_budget_rejection( + &c, + json!({"cmd":"plot", "start_hz":1e-300, "stop_hz":1e-300, "ppd":1, "duration":60.0}), + "start_hz", + ); +} + +/// #428: a tiny `duration` (0.001 s) makes `plot`'s per-point capture length /// `max(duration, 3.0 / freq)`, which falls under `analyze`'s 256-sample /// minimum for any point above 562.5 Hz at the fake backend's 48 kHz rate /// (`3.0 / 562.5 * 48_000 == 256`). A sweep that fails partway through /// must not archive the successful prefix as a complete measurement: it /// must terminate on `error`, carrying how much of the request actually /// completed, and never reach `done` or `measurement/report` for this run. +/// +/// Not `duration: 0`: the request budget rejects it before spawn (#437; +/// ZMQ.md documents `duration` as greater than 0). 0.001 s is below every +/// point's `3.0 / freq` floor, so the per-point lengths — and the failure — +/// are the same as with 0. #[test] -fn plot_duration_zero_fails_atomically_instead_of_archiving_partial_sweep() { +fn plot_tiny_duration_fails_atomically_instead_of_archiving_partial_sweep() { let d = Daemon::spawn(); let c = Client::new(&d); let r = c.call(json!({ @@ -162,7 +260,7 @@ fn plot_duration_zero_fails_atomically_instead_of_archiving_partial_sweep() { "stop_hz": 2000.0, "level_dbfs": -20.0, "ppd": 5, - "duration": 0.0, + "duration": 0.001, })); assert_eq!(r["ok"], json!(true), "plot ack: {r}"); diff --git a/ac-rs/crates/ac-daemon/tests/it_protocol/plot_ir.rs b/ac-rs/crates/ac-daemon/tests/it_protocol/plot_ir.rs index 2415f941..10bb08ad 100644 --- a/ac-rs/crates/ac-daemon/tests/it_protocol/plot_ir.rs +++ b/ac-rs/crates/ac-daemon/tests/it_protocol/plot_ir.rs @@ -79,6 +79,80 @@ fn plot_ir_emits_impulse_response_with_expected_delay_peak() { assert!(got_report, "never saw measurement/report frame"); } +fn assert_plot_ir_stops_promptly(duration: f64, tail_s: f64, settle: Duration) { + let d = Daemon::spawn(); + let c = Client::new(&d); + let reply = c.call(json!({ + "cmd":"plot_ir", + "f1_hz":200.0, + "f2_hz":8000.0, + "duration":duration, + "tail_s":tail_s, + "window_len":1024, + "n_harmonics":3 + })); + assert_eq!(reply["ok"], json!(true), "plot_ir rejected: {reply}"); + + std::thread::sleep(settle); + let started = Instant::now(); + let stopped = c.call(json!({"cmd":"stop", "name":"plot_ir"})); + assert!( + started.elapsed() < Duration::from_secs(1), + "stop blocked for {:?}: {stopped}", + started.elapsed() + ); + assert_eq!(stopped["stopped"], json!(["plot_ir"]), "{stopped}"); + assert_eq!(stopped["stimulus"], json!("silent"), "{stopped}"); + let status = c.call(json!({"cmd":"status"})); + assert_eq!(status["busy"], json!(false), "{status}"); + + // #437 codex-qa: a prompt `stop` reply plus `busy:false` also holds when + // `plot_ir` simply ran to completion before `stop` was ever sent — a + // finished worker's handle stays in the workers map until `stop` removes + // it, so those two assertions alone cannot distinguish "cancelled + // mid-run" from "already done, and `stop` just reaped it". Prove + // cancellation directly: drain every PUB frame already queued and assert + // none of them is this request's `measurement/impulse_response`, + // `measurement/report`, or `done` frame. A regression that reverts the + // handler to the non-cancellable `play_and_capture` would publish all + // three almost immediately (that path has no pacing sleep at all on the + // fake backend), so they would already be sitting on the SUB socket by + // the time this drain runs, well before `stop` was sent. + let mut drained = 0; + while let Some((topic, payload)) = c.recv_pub(50) { + drained += 1; + assert!( + drained <= 1000, + "runaway PUB drain while checking for a post-cancel completion frame" + ); + if payload["cmd"] != json!("plot_ir") { + continue; + } + assert_ne!( + topic, "measurement/impulse_response", + "plot_ir published an impulse response after stop cancelled it: {payload}" + ); + assert_ne!( + topic, "measurement/report", + "plot_ir published a report after stop cancelled it: {payload}" + ); + assert_ne!( + topic, "done", + "plot_ir published a done frame after stop cancelled it: {payload}" + ); + } +} + +#[test] +fn plot_ir_stop_cancels_during_stimulus() { + assert_plot_ir_stops_promptly(5.0, 0.1, Duration::from_millis(200)); +} + +#[test] +fn plot_ir_stop_cancels_during_tail() { + assert_plot_ir_stops_promptly(0.1, 5.0, Duration::from_millis(300)); +} + // --------------------------------------------------------------------- // Drive ceiling (#360) — plot_ir and calibrate previously emitted an // unclamped level; both are commands whose whole point is to put a