agent: code-atomiser-fix — surface capture-thread drops + bypass validation requeue cap

dropped_chunks was incremented on cpal-callback channel-full and
validation requeue overflow but never read by the live session, so
the UI's dropped_audio_ms missed callback-level losses entirely.
Architecture doc had flagged this as a TODO. Also: the 350ms
validation buffer was requeued via try_send into the same 32-slot
channel, silently dropping past the cap on small-buffer audio hosts
(WASAPI exclusive, low-latency ALSA at 256 frames -> ~65 chunks).

Fix: live runtime reads MicrophoneCapture::dropped_chunks() on each
recv_audio tick (LiveSessionRuntime::poll_capture_drops) and converts
the per-chunk-duration delta into the dropped_audio_ms surfaced to
the UI overload status. Per-chunk duration is derived from the most
recent AudioChunk's sample_rate + samples-per-channel so it adapts
to whatever rate cpal is delivering at. Validation requeue moved
from try_send into the bounded channel onto a VecDeque<AudioChunk>
returned alongside the Receiver; ActiveCapture drains the replay
buffer before reading rx in recv_audio, bypassing the 32-slot cap
entirely. Architecture doc updated to remove the TODO and document
the new pre-roll path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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<AudioChunk>)> {
+    ///
+    /// 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<AudioChunk>, mpsc::Receiver<AudioChunk>)> {
         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<AudioChunk>)> {
+    pub fn start() -> Result<(Self, VecDeque<AudioChunk>, mpsc::Receiver<AudioChunk>)> {
         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<AudioChunk>)> {
+) -> Result<(MicrophoneCapture, VecDeque<AudioChunk>, mpsc::Receiver<AudioChunk>)> {
     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::<AudioChunk>(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<AudioChunk> = 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,
     ))
 }

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::<T>` 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::<T>` (`capture.rs:505`) is generic over `T: Sample + SizedSample` with `f32: FromSample<T>` 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<AudioChunk>
+                   └─ on accept: return MicrophoneCapture + VecDeque<AudioChunk> pre-roll + Receiver<AudioChunk>
 ```
 
 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 `<unnamed>` 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<AudioChunk>` 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
 

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<AudioChunk>,
     mic_error_rx: Option<std::sync::mpsc::Receiver<CaptureRuntimeError>>,
+    /// 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<AudioChunk>,
 }
 
 impl ActiveCapture {
     fn start(config: &StartLiveTranscriptionConfig) -> Result<Self, String> {
-        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<RecentTranscriptSegment>,
+    /// 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<Option<AudioChunk>, 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;