jake/Lumotia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audio: Day 2 — Codex follow-up hardening (channel disconnect, spawn_blocking, fallback silence guard, requeue counting, runtime error propagation)

Browse Source 
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.
This commit is contained in:
Jake
2026-04-17 13:02:51 +01:00
parent 41db162041
commit 19a6b83cb2
3 changed files with 126 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
crates/audio/src/capture.rs
View File

@@ -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<cpal::Stream>,
@@ -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<AtomicU64>,
/// Receiver for runtime stream errors (device unplugged, audio server
/// crash, etc.). The live session calls `error_rx()` once and listens.
error_rx: Option<mpsc::Receiver<CaptureRuntimeError>>,
}
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<mpsc::Receiver<CaptureRuntimeError>> {
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<Vec<DeviceInfo>> {
@@ -254,11 +276,16 @@ fn open_and_validate(
let (tx, rx) = mpsc::sync_channel::<AudioChunk>(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::<CaptureRuntimeError>(16);
let stream = match format {
SampleFormat::F32 => build_input_stream::<f32>(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()),
SampleFormat::I16 => build_input_stream::<i16>(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()),
SampleFormat::U16 => build_input_stream::<u16>(&device, &config, sample_rate, channels, tx, dropped_chunks.clone()),
SampleFormat::F32 => build_input_stream::<f32>(&device, &config, sample_rate, channels, tx, dropped_chunks.clone(), err_tx.clone(), name.to_string()),
SampleFormat::I16 => build_input_stream::<i16>(&device, &config, sample_rate, channels, tx, dropped_chunks.clone(), err_tx.clone(), name.to_string()),
SampleFormat::U16 => build_input_stream::<u16>(&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<T>(
device: &cpal::Device,
supported_config: &cpal::SupportedStreamConfig,
@@ -336,12 +381,15 @@ fn build_input_stream<T>(
channels: u16,
tx: mpsc::SyncSender<AudioChunk>,
dropped_chunks: Arc<AtomicU64>,
err_tx: mpsc::SyncSender<CaptureRuntimeError>,
device_name: String,
) -> std::result::Result<cpal::Stream, cpal::BuildStreamError>
where
T: Sample + SizedSample,
f32: FromSample<T>,
{
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,
)
}

2
crates/audio/src/lib.rs
View File

@@ -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;

94
src-tauri/src/commands/audio.rs
View File

@@ -53,15 +53,28 @@ pub async fn start_native_capture(
device_name.as_deref().unwrap_or("<auto>")
);
// 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<f32> = if channels > 1 {
chunk
.samples
.chunks(channels)
.map(|frame| frame.iter().sum::<f32>() / channels as f32)
.collect()
} else {
chunk.samples
};
// Downmix to mono if stereo
let mono: Vec<f32> = if channels > 1 {
chunk
.samples
.chunks(channels)
.map(|frame| frame.iter().sum::<f32>() / 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;