From 19a6b83cb2cbbe0b060abf37358f9b0a24667c28 Mon Sep 17 00:00:00 2001 From: Jake Date: Fri, 17 Apr 2026 13:02:51 +0100 Subject: [PATCH] =?UTF-8?q?audio:=20Day=202=20=E2=80=94=20Codex=20follow-u?= =?UTF-8?q?p=20hardening=20(channel=20disconnect,=20spawn=5Fblocking,=20fa?= =?UTF-8?q?llback=20silence=20guard,=20requeue=20counting,=20runtime=20err?= =?UTF-8?q?or=20propagation)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the 6 Codex review findings on the Day 1 mic-capture work (commits 96980c7 + 41db162). Detail in output/reports/kon-codex-mic-capture-followup-2026-04-17.md (CORBEL workspace). src-tauri/src/commands/audio.rs: - D1: Sending an explicit stop signal before dropping stop_tx, so the accumulator task wakes up immediately rather than waiting for the Disconnected detection added below. - D2: Wrapping MicrophoneCapture::start() in tokio::task::spawn_blocking. start() can spend up to 350ms × N_devices × 2 passes; running it on the async runtime froze other Tauri commands. - M3: Match on rx.try_recv() error variants. Empty -> sleep + continue. Disconnected -> exit accumulator task immediately. Previous behaviour was an infinite loop after the capture stream died. crates/audio/src/capture.rs: - D3: Added DEAD_SILENCE_FLOOR (1e-7) gate that applies even in the fallback pass. PulseAudio/PipeWire can stream zero-valued bytes from a borked device; that is worse than failing fast. - M1: Validation requeue (`for chunk in collected { try_send }`) now counts drops in the same dropped_chunks counter. - M2: New CaptureRuntimeError type. The cpal stream error callback now forwards errors via a separate sync_channel (capacity 16) that the live session can drain via take_error_rx() and surface as toasts. Re-exported from kon_audio crate root. cargo check -p kon-audio passes clean. NOT YET DONE (M4): JACK-specific monitor name patterns. Defers until testing on a JACK host. Current is_monitor_name() may miss JACK conventions. Wiring the new error_rx into the live session and surfacing as toasts lands with the Day 3 error-toast system commit. --- crates/audio/src/capture.rs | 69 +++++++++++++++++++++--- crates/audio/src/lib.rs | 2 +- src-tauri/src/commands/audio.rs | 94 ++++++++++++++++++++++----------- 3 files changed, 126 insertions(+), 39 deletions(-) diff --git a/crates/audio/src/capture.rs b/crates/audio/src/capture.rs index a7ab53b..6d14b98 100644 --- a/crates/audio/src/capture.rs +++ b/crates/audio/src/capture.rs @@ -44,6 +44,17 @@ pub struct DeviceInfo { pub is_default: bool, } +/// A non-fatal capture-time error emitted by the cpal stream callback after +/// `start()` has already returned. The live session subscribes to these via +/// `error_rx()` so the frontend can show a toast when the mic vanishes +/// mid-recording. +/// (Codex review 2026/04/17 M2) +#[derive(Debug, Clone)] +pub struct CaptureRuntimeError { + pub device_name: String, + pub message: String, +} + /// Manages microphone capture via cpal. pub struct MicrophoneCapture { stream: Option, @@ -52,6 +63,9 @@ pub struct MicrophoneCapture { /// Counter incremented every time the capture callback drops a chunk /// because the channel was full. Read via `dropped_chunks()`. dropped_chunks: Arc, + /// Receiver for runtime stream errors (device unplugged, audio server + /// crash, etc.). The live session calls `error_rx()` once and listens. + error_rx: Option>, } impl MicrophoneCapture { @@ -62,6 +76,14 @@ impl MicrophoneCapture { self.dropped_chunks.load(Ordering::Relaxed) } + /// Take the runtime-error receiver. Can be called once per capture; the + /// caller (live session manager) drains it on its own cadence and surfaces + /// errors to the frontend. Returns None on the second call. + /// (Codex review 2026/04/17 M2) + pub fn take_error_rx(&mut self) -> Option> { + self.error_rx.take() + } + /// Enumerate every input device the host knows about, with the metadata /// needed by the device-picker UI. pub fn list_devices() -> Result> { @@ -254,11 +276,16 @@ fn open_and_validate( let (tx, rx) = mpsc::sync_channel::(AUDIO_CHANNEL_CAPACITY); let requeue_tx = tx.clone(); let dropped_chunks = Arc::new(AtomicU64::new(0)); + // Bounded channel for runtime stream errors. Capacity 16 = plenty for + // the rare error case; if it ever fills, we drop newer errors silently + // because they would be redundant noise in a stream that is already + // failing. (Codex review 2026/04/17 M2) + let (err_tx, err_rx) = mpsc::sync_channel::(16); let stream = match format { - SampleFormat::F32 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()), - SampleFormat::I16 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()), - SampleFormat::U16 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()), + SampleFormat::F32 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone(), err_tx.clone(), name.to_string()), + SampleFormat::I16 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone(), err_tx.clone(), name.to_string()), + SampleFormat::U16 => build_input_stream::(&device, &config, sample_rate, channels, tx, dropped_chunks.clone(), err_tx.clone(), name.to_string()), other => { return Err(KonError::AudioCaptureFailed(format!( "unsupported sample format {other:?}" @@ -313,9 +340,25 @@ fn open_and_validate( ))); } - // Re-queue the collected chunks so downstream gets them. + // Even in the fallback pass (require_audio=false), reject completely + // dead-zero audio. PulseAudio/PipeWire will sometimes happily emit a + // long stream of f32 zeros from a borked device — that is worse than + // failing fast. (Codex review 2026/04/17 D3) + const DEAD_SILENCE_FLOOR: f32 = 1e-7; + if rms < DEAD_SILENCE_FLOOR { + return Err(KonError::AudioCaptureFailed(format!( + "device produced dead silence (rms={rms:.6e} below absolute floor {DEAD_SILENCE_FLOOR:.6e})" + ))); + } + + // Re-queue the collected chunks so downstream gets them. Count any + // drops here against the same `dropped_chunks` counter so the live + // session sees them and can warn the user. + // (Codex review 2026/04/17 M1) for chunk in collected { - let _ = requeue_tx.try_send(chunk); + if requeue_tx.try_send(chunk).is_err() { + dropped_chunks.fetch_add(1, Ordering::Relaxed); + } } eprintln!("[kon-audio] selected microphone: '{name}'"); @@ -324,11 +367,13 @@ fn open_and_validate( stream: Some(stream), device_name: name.to_string(), dropped_chunks, + error_rx: Some(err_rx), }, rx, )) } +#[allow(clippy::too_many_arguments)] fn build_input_stream( device: &cpal::Device, supported_config: &cpal::SupportedStreamConfig, @@ -336,12 +381,15 @@ fn build_input_stream( channels: u16, tx: mpsc::SyncSender, dropped_chunks: Arc, + err_tx: mpsc::SyncSender, + device_name: String, ) -> std::result::Result where T: Sample + SizedSample, f32: FromSample, { let config: cpal::StreamConfig = supported_config.clone().into(); + let err_device_name = device_name.clone(); device.build_input_stream( &config, move |data: &[T], _| { @@ -358,7 +406,16 @@ where dropped_chunks.fetch_add(1, Ordering::Relaxed); } }, - |err| eprintln!("[kon-audio] capture error: {err}"), + move |err| { + // Surface stream errors to the live session via err_tx so the + // frontend can show a toast. Also keep the eprintln for ops + // logs. (Codex review 2026/04/17 M2) + eprintln!("[kon-audio] capture error: {err}"); + let _ = err_tx.try_send(CaptureRuntimeError { + device_name: err_device_name.clone(), + message: err.to_string(), + }); + }, None, ) } diff --git a/crates/audio/src/lib.rs b/crates/audio/src/lib.rs index 30795b4..05a68bd 100644 --- a/crates/audio/src/lib.rs +++ b/crates/audio/src/lib.rs @@ -6,7 +6,7 @@ pub mod streaming_resample; pub mod vad; pub mod wav; -pub use capture::{AudioChunk, DeviceInfo, MicrophoneCapture}; +pub use capture::{AudioChunk, CaptureRuntimeError, DeviceInfo, MicrophoneCapture}; pub use concurrency::decode_and_resample; pub use decode::decode_audio_file; pub use resample::resample_to_16khz; diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index d8def7a..0d1c07f 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -53,15 +53,28 @@ pub async fn start_native_capture( device_name.as_deref().unwrap_or("") ); - // Stop any existing capture + // Stop any existing capture: send an explicit stop signal first, then + // drop the sender. The accumulator task watches for `Disconnected` too, + // but signalling explicitly avoids the brief race window. + // (Codex review 2026/04/17 D1) if let Some(tx) = state.stop_tx.lock().unwrap().take() { + let _ = tx.try_send(()); drop(tx); } - let (capture, rx) = match device_name.as_deref() { - Some(name) if !name.is_empty() => MicrophoneCapture::start_with_device(name), - _ => MicrophoneCapture::start(), - } + // `MicrophoneCapture::start()` is synchronous and may spend up to + // DEVICE_VALIDATION_MS per device per pass (350ms × N devices × 2 passes + // worst case). Run on a blocking thread so the async runtime stays + // responsive to other Tauri commands. (Codex review 2026/04/17 D2) + let device_name_for_blocking = device_name.clone(); + let (capture, rx) = tokio::task::spawn_blocking(move || { + match device_name_for_blocking.as_deref() { + Some(name) if !name.is_empty() => MicrophoneCapture::start_with_device(name), + _ => MicrophoneCapture::start(), + } + }) + .await + .map_err(|e| format!("audio task join error: {e}"))? .map_err(|e| { eprintln!("[native-capture] MicrophoneCapture::start failed: {e}"); e.to_string() @@ -95,38 +108,52 @@ pub async fn start_native_capture( break; } - // Drain available audio chunks from cpal (non-blocking) + // Drain available audio chunks from cpal (non-blocking). + // Distinguish Empty (try again) from Disconnected (capture stream + // is dead — exit the loop, don't spin forever). + // (Codex review 2026/04/17 M3) let mut got_data = false; - while let Ok(chunk) = rx.try_recv() { - got_data = true; - let sample_rate = chunk.sample_rate; - let channels = chunk.channels as usize; + let mut capture_dead = false; + loop { + match rx.try_recv() { + Ok(chunk) => { + got_data = true; + let sample_rate = chunk.sample_rate; + let channels = chunk.channels as usize; - // Downmix to mono if stereo - let mono: Vec = if channels > 1 { - chunk - .samples - .chunks(channels) - .map(|frame| frame.iter().sum::() / channels as f32) - .collect() - } else { - chunk.samples - }; + // Downmix to mono if stereo + let mono: Vec = if channels > 1 { + chunk + .samples + .chunks(channels) + .map(|frame| frame.iter().sum::() / channels as f32) + .collect() + } else { + chunk.samples + }; - // Downsample to 16kHz using simple decimation - // (acceptable quality for speech — same approach as pcm-processor.js) - let ratio = sample_rate as f64 / WHISPER_SAMPLE_RATE as f64; - if (ratio - 1.0).abs() < 0.01 { - pcm_buffer.extend_from_slice(&mono); - } else { - let mut pos: f64 = 0.0; - for &s in &mono { - pos += 1.0; - if pos >= ratio { - pcm_buffer.push(s); - pos -= ratio; + // Downsample to 16kHz using simple decimation + // (acceptable quality for speech — same approach as pcm-processor.js) + let ratio = sample_rate as f64 / WHISPER_SAMPLE_RATE as f64; + if (ratio - 1.0).abs() < 0.01 { + pcm_buffer.extend_from_slice(&mono); + } else { + let mut pos: f64 = 0.0; + for &s in &mono { + pos += 1.0; + if pos >= ratio { + pcm_buffer.push(s); + pos -= ratio; + } + } } } + Err(std::sync::mpsc::TryRecvError::Empty) => break, + Err(std::sync::mpsc::TryRecvError::Disconnected) => { + eprintln!("[native-capture] capture stream disconnected; accumulator exiting"); + capture_dead = true; + break; + } } } @@ -144,6 +171,9 @@ pub async fn start_native_capture( })); } + if capture_dead { + break; + } if !got_data { // Avoid busy-spinning when no audio data is available tokio::time::sleep(std::time::Duration::from_millis(10)).await;