diff --git a/crates/audio/src/capture.rs b/crates/audio/src/capture.rs index 05c838a..3d85c12 100644 --- a/crates/audio/src/capture.rs +++ b/crates/audio/src/capture.rs @@ -1,3 +1,4 @@ +use std::collections::VecDeque; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc; use std::sync::Arc; @@ -140,7 +141,18 @@ impl MicrophoneCapture { /// Start capturing from the device whose name matches `device_name` exactly. /// If no match is found, returns an error rather than silently falling back. - pub fn start_with_device(device_name: &str) -> Result<(Self, mpsc::Receiver)> { + /// + /// The returned tuple is `(capture, replay_buffer, rx)`: + /// - `replay_buffer` holds chunks observed during the 350ms + /// validation pre-roll. Consumers MUST drain it before reading + /// from `rx` so the head of the recording isn't lost on hosts + /// whose cpal buffer is small enough to overflow the 32-slot + /// channel during validation (WASAPI exclusive, low-latency + /// ALSA at 256 frames). + /// - `rx` is the live cpal callback channel. + pub fn start_with_device( + device_name: &str, + ) -> Result<(Self, VecDeque, mpsc::Receiver)> { let host = cpal::default_host(); let devices = host .input_devices() @@ -169,7 +181,7 @@ impl MicrophoneCapture { /// a short window — this is what defeats the "silent monitor source wins" bug. /// 4. If no non-monitor device produces real audio, fall back to monitor sources /// as a last resort (with a clear log line). Never accept dead silence. - pub fn start() -> Result<(Self, mpsc::Receiver)> { + pub fn start() -> Result<(Self, VecDeque, mpsc::Receiver)> { let host = cpal::default_host(); let default_name = host .default_input_device() @@ -358,7 +370,7 @@ fn open_and_validate( device: cpal::Device, name: &str, require_audio: bool, -) -> Result<(MicrophoneCapture, mpsc::Receiver)> { +) -> Result<(MicrophoneCapture, VecDeque, mpsc::Receiver)> { let config = device .default_input_config() .map_err(|e| Error::AudioCaptureFailed(format!("default_input_config: {e}")))?; @@ -376,7 +388,6 @@ 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 32 = plenty for // the rare error case; if it ever fills, drops are reported via stderr @@ -490,16 +501,21 @@ fn open_and_validate( ))); } - // 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. - for chunk in collected { - if requeue_tx.try_send(chunk).is_err() { - dropped_chunks.fetch_add(1, Ordering::Relaxed); - } - } + // Hand the validation pre-roll back to the consumer as a separate + // VecDeque rather than try_send-requeuing into the 32-slot channel. + // On small-buffer audio hosts (WASAPI exclusive at ~256 frames / + // low-latency ALSA) the 350ms window collects ~65 chunks; the old + // requeue path silently dropped roughly half of them, losing ~150ms + // from the head of every recording. The consumer-side drain + // bypasses the channel cap entirely. + let replay_buffer: VecDeque = collected.into_iter().collect(); - tracing::info!(target: "lumotia_audio", device = %name, "selected microphone"); + tracing::info!( + target: "lumotia_audio", + device = %name, + replay_chunks = replay_buffer.len(), + "selected microphone" + ); Ok(( MicrophoneCapture { stream: Some(stream), @@ -507,6 +523,7 @@ fn open_and_validate( dropped_chunks, error_rx: Some(err_rx), }, + replay_buffer, rx, )) } diff --git a/docs/architecture-map/03-audio-transcription/audio-capture-pipeline.md b/docs/architecture-map/03-audio-transcription/audio-capture-pipeline.md index 900946c..0d0ce80 100644 --- a/docs/architecture-map/03-audio-transcription/audio-capture-pipeline.md +++ b/docs/architecture-map/03-audio-transcription/audio-capture-pipeline.md @@ -74,7 +74,7 @@ Monitor detection (`is_monitor_name`, `capture.rs:258`) catches the standard Pul 4. Dispatches to `build_input_stream::` for the device's sample format (F32 / I16 / U16). Anything else returns `MagnotiaError::AudioCaptureFailed`. 5. Calls `stream.play()`. 6. Sniffs samples for `DEVICE_VALIDATION_MS`, sums squared samples, derives RMS. -7. Rejects below floor (or dead silence). Otherwise re-queues the validation chunks back into the channel so downstream consumers do not lose the first 350 ms. +7. Rejects below floor (or dead silence). Otherwise hands the validation chunks back to the consumer as a `VecDeque` pre-roll alongside the live `mpsc::Receiver`, so downstream consumers do not lose the first 350 ms even on small-buffer hosts that would overflow the bounded channel. `build_input_stream::` (`capture.rs:505`) is generic over `T: Sample + SizedSample` with `f32: FromSample` so the same body handles all three sample formats. The data callback maps every sample to `f32`, packages an `AudioChunk`, and `try_send`s on the channel; failure increments the `dropped_chunks` atomic. The error callback ships a `CaptureRuntimeError` on `err_tx` (channel-full path increments `dropped_errors` and logs to stderr). @@ -96,20 +96,20 @@ cpal default_host() │ └─ error callback: cpal::StreamError → CaptureRuntimeError → err_mpsc └─ stream.play() └─ 350 ms sniff → RMS → accept | reject - └─ on accept: re-queue collected chunks, return MicrophoneCapture + Receiver + └─ on accept: return MicrophoneCapture + VecDeque pre-roll + Receiver ``` Output: one `AudioChunk` per cpal callback period at the device's native rate. The live session is responsible for downmixing channels (if `channels > 1`) and feeding `StreamingResampler` to reach 16 kHz mono. Native rate is *not* normalised here. ## Watch-outs -- **Channel capacity is 32 chunks.** At a 1024-frame cpal buffer at 48 kHz that's roughly 700 ms. A blocked consumer for longer than that means dropped audio. `dropped_chunks()` is the visibility hook; the live-session command must surface it. +- **Channel capacity is 32 chunks.** At a 1024-frame cpal buffer at 48 kHz that's roughly 700 ms. A blocked consumer for longer than that means dropped audio. `dropped_chunks()` is the visibility hook; the live-session command reads it on every `recv_audio` tick (`commands/live.rs::poll_capture_drops`) and converts the delta into the `dropped_audio_ms` surfaced to the UI overlay. - **Default-device first works against the safest pick on Linux Pulse setups** where the default sink monitor sneaks in. The four-tier sort handles this, but only because monitor names match the patterns in `is_monitor_name`. New PipeWire schemes that don't include `.monitor` / `loopback` would slip through. - **350 ms validation window adds a startup latency floor.** Slice 2 needs to know about this when wiring "click record". - **`stop()` is `pause`, not `drop`.** The stream object is kept alive until `Drop`. A subsequent `start()` on the same `MicrophoneCapture` is not supported (signature returns a fresh instance). - **Sample format dispatch is closed-set.** Anything not F32 / I16 / U16 is a hard error. cpal can in principle expose I8 / I32 / F64 on exotic devices. - **`device_display_name` swallows errors.** `cpal::Device::description()` errors silently become `None`, then `` downstream. Acceptable for a UI list, surprising for debugging. -- **Re-queue uses `try_send` on a channel of capacity 32.** If the sniff produced more than 32 chunks (≈64 ms at 48 kHz 256-frame buffers — uncommon but possible), the early ones are dropped against the same `dropped_chunks` counter. Documented at `capture.rs:486`. +- **Validation pre-roll is returned out-of-band, not requeued.** `open_and_validate` returns a `VecDeque` alongside the live `mpsc::Receiver`; the live-session consumer drains the deque before reading the channel. This bypasses the 32-slot cap entirely so small-buffer hosts (WASAPI exclusive, low-latency ALSA at 256 frames) don't lose ~150ms from the head of every recording. Earlier versions used `try_send` to requeue and silently dropped half of the pre-roll on those hosts. ## See also diff --git a/src-tauri/src/commands/live.rs b/src-tauri/src/commands/live.rs index bf08061..040ea6e 100644 --- a/src-tauri/src/commands/live.rs +++ b/src-tauri/src/commands/live.rs @@ -1,6 +1,6 @@ #![allow(clippy::too_many_arguments)] -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::path::PathBuf; use std::sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, @@ -165,26 +165,40 @@ struct LiveSessionSummary { /// explicit and locally structured. struct ActiveCapture { /// Keeping the capture handle alive keeps the underlying cpal stream alive. - _capture: MicrophoneCapture, + capture: MicrophoneCapture, rx: std::sync::mpsc::Receiver, mic_error_rx: Option>, + /// Pre-roll chunks collected during the 350ms device validation + /// window. Drained BEFORE reading from `rx` so the head of the + /// recording survives small-buffer hosts that would otherwise + /// overflow the 32-slot capture channel during validation. + replay_buffer: VecDeque, } impl ActiveCapture { fn start(config: &StartLiveTranscriptionConfig) -> Result { - let (mut capture, rx) = match config.microphone_device.as_deref() { + let (mut capture, replay_buffer, rx) = match config.microphone_device.as_deref() { Some(name) if !name.is_empty() => MicrophoneCapture::start_with_device(name), _ => MicrophoneCapture::start(), } .map_err(|e| e.to_string())?; let mic_error_rx = capture.take_error_rx(); Ok(Self { - _capture: capture, + capture, rx, mic_error_rx, + replay_buffer, }) } + /// Cumulative chunks the cpal callback (and the validation + /// pre-roll, when it overflows the channel cap) has dropped. + /// The live runtime samples this on each recv tick to convert + /// the delta into `dropped_audio_ms`. + fn dropped_chunks(&self) -> u64 { + self.capture.dropped_chunks() + } + fn drain_runtime_errors( &mut self, session_id: u64, @@ -217,6 +231,19 @@ struct LiveLoopState { resampler_flushed: bool, result_listener_lost: bool, recent_segments: Vec, + /// Cumulative value of `MicrophoneCapture::dropped_chunks()` last + /// observed by `poll_capture_drops`. Subtracted from the live + /// reading to compute the per-tick delta added to + /// `dropped_audio_ms`. Counts callback-level losses that the + /// session's own overflow path can't see. + last_dropped_chunks: u64, + /// Most-recent chunk dimensions used to convert a `dropped_chunks` + /// delta into milliseconds. Populated on the first received chunk + /// and refreshed thereafter. Zeroed sentinel means "no chunk yet"; + /// drops observed before any chunk arrives are deferred until we + /// know the device's native rate. + last_chunk_samples_per_chan: u32, + last_chunk_sample_rate: u32, } impl LiveLoopState { @@ -276,6 +303,11 @@ impl LiveSessionRuntime { if let Some(chunk) = self.recv_audio()? { self.process_audio_chunk(chunk)?; } + // Pick up cpal-callback drops AND validation-window + // requeue drops that the live session's own buffer + // overflow path can't see. Called after recv_audio so the + // most recent chunk dimensions inform the ms conversion. + self.poll_capture_drops(); self.drop_pending_overflow(); self.flush_tail_if_stopping()?; if self.dispatch_inference_if_ready() { @@ -306,6 +338,14 @@ impl LiveSessionRuntime { } fn recv_audio(&mut self) -> Result, String> { + // Drain the validation pre-roll first. This is the head-of-the + // -recording audio that was collected during the 350ms device + // validation window; replaying it from a consumer-side VecDeque + // bypasses the 32-slot channel cap that previously dropped half + // of it silently on small-buffer hosts. + if let Some(chunk) = self.capture.replay_buffer.pop_front() { + return Ok(Some(chunk)); + } match self.capture.rx.recv_timeout(Duration::from_millis(25)) { Ok(chunk) => Ok(Some(chunk)), Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Ok(None), @@ -321,6 +361,14 @@ impl LiveSessionRuntime { } fn process_audio_chunk(&mut self, chunk: AudioChunk) -> Result<(), String> { + // Remember native-rate dimensions so `poll_capture_drops` can + // turn a chunks-dropped delta into milliseconds without + // hard-coding host-buffer assumptions. + let channels = chunk.channels.max(1) as u32; + let total_samples = chunk.samples.len() as u32; + self.state.last_chunk_sample_rate = chunk.sample_rate; + self.state.last_chunk_samples_per_chan = total_samples / channels; + let mono = downmix_chunk(chunk.samples, chunk.channels as usize); let resampler = match &mut self.state.resampler { Some(resampler) => resampler, @@ -341,6 +389,48 @@ impl LiveSessionRuntime { Ok(()) } + /// Reconcile the live session's `dropped_audio_ms` against the + /// capture thread's cpal-callback + validation-requeue drop + /// counter. Without this the UI's reported drops only ever + /// reflected the live runtime's own buffer overflow path, missing + /// every callback-level loss caused by transient back-pressure or + /// the 350ms validation pre-roll overflowing the channel cap on + /// small-buffer audio hosts. + fn poll_capture_drops(&mut self) { + let now = self.capture.dropped_chunks(); + if now == self.state.last_dropped_chunks { + return; + } + let delta = now.saturating_sub(self.state.last_dropped_chunks); + self.state.last_dropped_chunks = now; + + // Defer the conversion until we've seen at least one chunk so + // the dimensions are real. Validation-window drops can fire + // before any chunk reaches `process_audio_chunk`; they get + // attributed at the next reconciliation once we know the rate. + if self.state.last_chunk_sample_rate == 0 + || self.state.last_chunk_samples_per_chan == 0 + { + // Roll back the consumed delta so we re-observe it once + // we have dimensions to convert it with. + self.state.last_dropped_chunks = self.state.last_dropped_chunks.saturating_sub(delta); + return; + } + + let per_chunk_ms = (self.state.last_chunk_samples_per_chan as u64 * 1000) + / self.state.last_chunk_sample_rate.max(1) as u64; + let added_ms = delta.saturating_mul(per_chunk_ms); + if added_ms == 0 { + return; + } + self.state.dropped_audio_ms = self.state.dropped_audio_ms.saturating_add(added_ms); + let _ = self.status_channel.send(LiveStatusMessage::Overload { + session_id: self.session_id, + dropped_audio_ms: self.state.dropped_audio_ms, + message: "Microphone capture dropped audio chunks (downstream back-pressure)".into(), + }); + } + fn drop_pending_overflow(&mut self) { if self.state.inflight.is_none() || self.state.capture_buffer.len() <= MAX_PENDING_SAMPLES { return;