feat(A.2 #19): progressive WAV write during live capture

The Vec<f32> in-memory accumulator on run_live_session had three
failure modes: (a) a crash during transcription took the recording
with it, (b) RAM grew linearly with session length, (c) OOM killed
the capture thread silently.

New kon_audio::WavWriter wraps hound::WavWriter<BufWriter<File>> with
an append-friendly API and a 500 ms-granularity header flush. On any
abort after a flush the on-disk file is a valid, playable WAV. Unit
test (brief item #19 acceptance) simulates the abort with
std::mem::forget and asserts the pre-flush samples are recoverable.

Live capture now:
- resolves the destination path at start_live_transcription_session
  time via a new resolve_recording_path helper extracted from
  persist_audio_samples,
- opens a WavWriter before any samples arrive, sample rate taken from
  LocalEngine::capabilities() (#13 wiring) with 16 kHz fallback,
- feeds the resampler output through WavWriter::append inside
  append_resampled_audio — drops the writer with a user-visible
  warning if a write fails mid-session,
- calls flush() at stop (after resampler tail), finalise() on clean
  exit, and drops-to-last-flushed state on abort.

LiveSessionSummary.audio_samples → audio_path: the path is already
written by the time stop_live_transcription_session runs; no
post-session write step remains for live capture.
persist_audio_samples is kept for the offline save_audio command.
This commit is contained in:
2026-04-22 04:39:44 +01:00
parent 8b49d0fe9c
commit f9b396a966
4 changed files with 300 additions and 38 deletions

View File

@@ -12,4 +12,4 @@ pub use decode::decode_audio_file;
pub use resample::resample_to_16khz; pub use resample::resample_to_16khz;
pub use streaming_resample::StreamingResampler; pub use streaming_resample::StreamingResampler;
pub use vad::SpeechDetector; pub use vad::SpeechDetector;
pub use wav::{read_wav, write_wav}; pub use wav::{read_wav, write_wav, WavWriter};

View File

@@ -1,8 +1,101 @@
use std::io::BufWriter;
use std::path::Path; use std::path::Path;
use kon_core::error::{KonError, Result}; use kon_core::error::{KonError, Result};
use kon_core::types::AudioSamples; use kon_core::types::AudioSamples;
/// Append-friendly WAV writer for long-running captures.
///
/// The in-memory `Vec<f32>` used by `run_live_session` to persist audio
/// on session end (brief item #19) has three failure modes: (a) a crash
/// during transcription takes the recording with it; (b) RAM bloat at
/// long session lengths; (c) an OOM kills the capture loop. `WavWriter`
/// replaces that pattern with an on-disk writer that periodically
/// flushes the WAV header so the file on disk is a valid, playable WAV
/// at any point the process is interrupted.
///
/// The writer samples at the rate / channel count supplied at
/// construction; callers read those from
/// `LocalEngine::capabilities()` (brief item #13 wiring) rather than
/// hardcoding 16 kHz / mono.
pub struct WavWriter {
inner: hound::WavWriter<BufWriter<std::fs::File>>,
samples_since_flush: usize,
flush_every: usize,
}
impl WavWriter {
/// Sample count between automatic header flushes. Flushing costs
/// two seeks per call; 8000 samples at 16 kHz = 500 ms, so the
/// worst-case "last half second is lost on crash" bound holds.
const DEFAULT_FLUSH_EVERY_SAMPLES: usize = 8_000;
/// Create a new WAV file at `path`, truncating any previous content.
/// Header reflects zero samples until the first `flush` or
/// `finalize`.
pub fn create(path: &Path, sample_rate: u32, channels: u16) -> Result<Self> {
let spec = hound::WavSpec {
channels,
sample_rate,
bits_per_sample: 16,
sample_format: hound::SampleFormat::Int,
};
let file = std::fs::File::create(path).map_err(KonError::Io)?;
let buffered = BufWriter::new(file);
let inner = hound::WavWriter::new(buffered, spec).map_err(|e| {
KonError::Io(std::io::Error::other(format!("WAV create failed: {e}")))
})?;
Ok(Self {
inner,
samples_since_flush: 0,
flush_every: Self::DEFAULT_FLUSH_EVERY_SAMPLES,
})
}
/// Append f32 samples in `[-1.0, 1.0]`. Samples outside that range
/// are clamped (matching `write_wav`). Automatically flushes the
/// header every `flush_every` samples so the on-disk file stays a
/// valid WAV even if the process is killed between appends.
pub fn append(&mut self, samples: &[f32]) -> Result<()> {
for &sample in samples {
let clamped = sample.clamp(-1.0, 1.0);
let int_sample = (clamped * i16::MAX as f32) as i16;
self.inner.write_sample(int_sample).map_err(|e| {
KonError::Io(std::io::Error::other(format!("WAV write failed: {e}")))
})?;
}
self.samples_since_flush += samples.len();
if self.samples_since_flush >= self.flush_every {
self.flush()?;
}
Ok(())
}
/// Force an immediate header flush. Leaves the file in a valid-WAV
/// state up to the current sample count. Callers do not need to
/// call this explicitly — `append` flushes every
/// `Self::DEFAULT_FLUSH_EVERY_SAMPLES` — but may do so at natural
/// boundaries (end-of-utterance, UI events) for tighter recovery.
pub fn flush(&mut self) -> Result<()> {
self.inner.flush().map_err(|e| {
KonError::Io(std::io::Error::other(format!("WAV flush failed: {e}")))
})?;
self.samples_since_flush = 0;
Ok(())
}
/// Finalise the WAV: writes the terminal header state and closes
/// the file. Call on clean session end. A dropped-without-finalize
/// writer leaves a playable file up to the last flush; callers
/// that care about the unflushed tail should always finalise.
pub fn finalize(self) -> Result<()> {
self.inner.finalize().map_err(|e| {
KonError::Io(std::io::Error::other(format!("WAV finalize failed: {e}")))
})?;
Ok(())
}
}
/// Write f32 PCM samples to a 16-bit WAV file. /// Write f32 PCM samples to a 16-bit WAV file.
pub fn write_wav(path: &Path, audio: &AudioSamples) -> Result<()> { pub fn write_wav(path: &Path, audio: &AudioSamples) -> Result<()> {
let spec = hound::WavSpec { let spec = hound::WavSpec {
@@ -58,6 +151,72 @@ pub fn read_wav(path: &Path) -> Result<AudioSamples> {
mod tests { mod tests {
use super::*; use super::*;
#[test]
fn wav_writer_survives_crash() {
// Property under test: a `WavWriter` that has been flushed but
// never finalised leaves a valid, readable WAV on disk. This
// is the crash-safety guarantee — if the kon process aborts
// mid-session, the on-disk file up to the last flush is
// recoverable.
//
// `std::mem::forget` is the canonical way to simulate an
// abort inside a unit test: it skips the Drop impl (which
// would otherwise finalise the hound writer for us) and
// mirrors what happens when the OS reaps the process without
// giving Rust a chance to run destructors.
let temp_dir = std::env::temp_dir();
let path = temp_dir.join("kon_test_wav_writer_survives_crash.wav");
let _ = std::fs::remove_file(&path);
let mut writer = WavWriter::create(&path, 16_000, 1).unwrap();
let flushed_samples = vec![0.1_f32; 16_000]; // 1s
writer.append(&flushed_samples).unwrap();
writer.flush().unwrap();
// Post-flush, append another second that will NOT be reflected
// in the header if the writer dies before the next flush.
let unflushed_tail = vec![0.2_f32; 16_000];
writer.append(&unflushed_tail).unwrap();
// Abort — Drop does not run, the hound finaliser is skipped.
std::mem::forget(writer);
let loaded = read_wav(&path).unwrap();
assert_eq!(loaded.sample_rate(), 16_000);
assert!(
loaded.samples().len() >= 16_000,
"expected at least the flushed 16000 samples, got {}",
loaded.samples().len()
);
// The flushed portion is readable and approximately correct.
for s in &loaded.samples()[..16_000] {
assert!(
(s - 0.1).abs() < 0.01,
"flushed sample {s} deviates from 0.1 beyond 16-bit quantisation slack",
);
}
let _ = std::fs::remove_file(&path);
}
#[test]
fn wav_writer_append_then_finalize_roundtrips() {
let temp_dir = std::env::temp_dir();
let path = temp_dir.join("kon_test_wav_writer_finalize.wav");
let _ = std::fs::remove_file(&path);
let mut writer = WavWriter::create(&path, 16_000, 1).unwrap();
writer.append(&vec![0.0_f32; 8_000]).unwrap();
writer.append(&vec![0.5_f32; 8_000]).unwrap();
writer.finalize().unwrap();
let loaded = read_wav(&path).unwrap();
assert_eq!(loaded.sample_rate(), 16_000);
assert_eq!(loaded.samples().len(), 16_000);
let _ = std::fs::remove_file(&path);
}
#[test] #[test]
fn wav_roundtrip() { fn wav_roundtrip() {
let temp_dir = std::env::temp_dir(); let temp_dir = std::env::temp_dir();

View File

@@ -228,25 +228,22 @@ pub async fn stop_native_capture(
Ok(samples) Ok(samples)
} }
pub async fn persist_audio_samples( /// Resolve the destination path for a new live-capture recording,
/// ensuring the parent directory exists. Extracted from
/// `persist_audio_samples` so `start_live_transcription_session` can
/// hand the path to the progressive WAV writer before any samples
/// arrive (brief item #19).
pub fn resolve_recording_path(
app: &tauri::AppHandle, app: &tauri::AppHandle,
samples: Vec<f32>, output_folder: Option<&str>,
output_folder: Option<String>, ) -> Result<PathBuf, String> {
) -> Result<String, String> { let recordings_dir = match output_folder.map(str::trim).filter(|s| !s.is_empty()) {
let recordings_dir = if let Some(ref folder) = output_folder { Some(folder) => PathBuf::from(folder),
if !folder.is_empty() { None => app
PathBuf::from(folder) .path()
} else {
app.path()
.app_local_data_dir()
.map_err(|e: tauri::Error| e.to_string())?
.join("recordings")
}
} else {
app.path()
.app_local_data_dir() .app_local_data_dir()
.map_err(|e: tauri::Error| e.to_string())? .map_err(|e: tauri::Error| e.to_string())?
.join("recordings") .join("recordings"),
}; };
std::fs::create_dir_all(&recordings_dir) std::fs::create_dir_all(&recordings_dir)
@@ -256,8 +253,15 @@ pub async fn persist_audio_samples(
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(); .as_secs();
let filename = format!("kon-{timestamp}.wav"); Ok(recordings_dir.join(format!("kon-{timestamp}.wav")))
let path = recordings_dir.join(&filename); }
pub async fn persist_audio_samples(
app: &tauri::AppHandle,
samples: Vec<f32>,
output_folder: Option<String>,
) -> Result<String, String> {
let path = resolve_recording_path(app, output_folder.as_deref())?;
let path_clone = path.clone(); let path_clone = path.clone();
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {

View File

@@ -1,6 +1,7 @@
#![allow(clippy::too_many_arguments)] #![allow(clippy::too_many_arguments)]
use std::collections::HashMap; use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{ use std::sync::{
atomic::{AtomicBool, AtomicU64, Ordering}, atomic::{AtomicBool, AtomicU64, Ordering},
Arc, Mutex, Arc, Mutex,
@@ -11,13 +12,13 @@ use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tauri::ipc::Channel; use tauri::ipc::Channel;
use crate::commands::audio::persist_audio_samples; use crate::commands::audio::resolve_recording_path;
use crate::commands::build_initial_prompt; use crate::commands::build_initial_prompt;
use crate::commands::models::{default_model_id_for_engine, ensure_model_loaded}; use crate::commands::models::{default_model_id_for_engine, ensure_model_loaded};
use crate::commands::power::PowerAssertion; use crate::commands::power::PowerAssertion;
use crate::AppState; use crate::AppState;
use kon_ai_formatting::{post_process_segments, FormatMode, PostProcessOptions}; use kon_ai_formatting::{post_process_segments, FormatMode, PostProcessOptions};
use kon_audio::{MicrophoneCapture, StreamingResampler}; use kon_audio::{MicrophoneCapture, StreamingResampler, WavWriter};
use kon_core::constants::WHISPER_SAMPLE_RATE; use kon_core::constants::WHISPER_SAMPLE_RATE;
use kon_core::types::{AudioSamples, Segment, TranscriptionOptions}; use kon_core::types::{AudioSamples, Segment, TranscriptionOptions};
use kon_transcription::LocalEngine; use kon_transcription::LocalEngine;
@@ -61,7 +62,6 @@ pub struct LiveTranscriptionState {
struct RunningLiveSession { struct RunningLiveSession {
id: u64, id: u64,
output_folder: Option<String>,
stop_flag: Arc<AtomicBool>, stop_flag: Arc<AtomicBool>,
handle: tokio::task::JoinHandle<Result<LiveSessionSummary, String>>, handle: tokio::task::JoinHandle<Result<LiveSessionSummary, String>>,
status_channel: Channel<LiveStatusMessage>, status_channel: Channel<LiveStatusMessage>,
@@ -147,7 +147,11 @@ pub enum LiveStatusMessage {
struct LiveSessionSummary { struct LiveSessionSummary {
session_id: u64, session_id: u64,
dropped_audio_ms: u64, dropped_audio_ms: u64,
audio_samples: Option<Vec<f32>>, /// Absolute path of the progressively-written WAV file. `Some` iff
/// `save_audio` was requested and the writer opened successfully;
/// the file on disk is a valid, playable WAV even if the session
/// was killed mid-append (brief item #19 crash-safety guarantee).
audio_path: Option<String>,
} }
struct InferenceTask { struct InferenceTask {
@@ -188,6 +192,7 @@ struct SpeechGateDecision {
#[tauri::command] #[tauri::command]
pub async fn start_live_transcription_session( pub async fn start_live_transcription_session(
app: tauri::AppHandle,
state: tauri::State<'_, AppState>, state: tauri::State<'_, AppState>,
live_state: tauri::State<'_, LiveTranscriptionState>, live_state: tauri::State<'_, LiveTranscriptionState>,
mut config: StartLiveTranscriptionConfig, mut config: StartLiveTranscriptionConfig,
@@ -247,18 +252,31 @@ pub async fn start_live_transcription_session(
.saturating_add(1); .saturating_add(1);
let stop_flag = Arc::new(AtomicBool::new(false)); let stop_flag = Arc::new(AtomicBool::new(false));
let engine = pick_engine(&state, &config.engine)?; let engine = pick_engine(&state, &config.engine)?;
let output_folder = config.output_folder.clone();
// Resolve the WAV destination up front so the progressive writer
// (brief item #19) can open it before any samples arrive. Failure
// to create the recordings directory is fatal — the user asked
// for save_audio=true and silently dropping the recording would
// surprise them worse.
let audio_path = if config.save_audio {
Some(resolve_recording_path(&app, config.output_folder.as_deref())?)
} else {
None
};
let worker_stop = stop_flag.clone(); let worker_stop = stop_flag.clone();
let worker_status = status_channel.clone(); let worker_status = status_channel.clone();
let worker_results = result_channel.clone(); let worker_results = result_channel.clone();
let dictionary_terms = profile_terms.clone(); let dictionary_terms = profile_terms.clone();
let worker_audio_path = audio_path.clone();
let handle = tokio::task::spawn_blocking(move || { let handle = tokio::task::spawn_blocking(move || {
run_live_session( run_live_session(
session_id, session_id,
engine, engine,
config, config,
worker_audio_path,
dictionary_terms, dictionary_terms,
worker_results, worker_results,
worker_status, worker_status,
@@ -268,7 +286,6 @@ pub async fn start_live_transcription_session(
*live_state.running.lock().unwrap() = Some(RunningLiveSession { *live_state.running.lock().unwrap() = Some(RunningLiveSession {
id: session_id, id: session_id,
output_folder,
stop_flag, stop_flag,
handle, handle,
status_channel, status_channel,
@@ -300,11 +317,11 @@ pub async fn stop_live_transcription_session(
.await .await
.map_err(|e| format!("Live session task failed: {e}"))??; .map_err(|e| format!("Live session task failed: {e}"))??;
let audio_path = if let Some(samples) = summary.audio_samples { // Progressive WAV writer (brief item #19) wrote samples to disk
Some(persist_audio_samples(&app, samples, running.output_folder.clone()).await?) // throughout the session; the path is already finalised. Nothing
} else { // further to persist.
None let _ = app;
}; let audio_path = summary.audio_path;
let response = StopLiveTranscriptionResponse { let response = StopLiveTranscriptionResponse {
session_id: summary.session_id, session_id: summary.session_id,
@@ -333,6 +350,7 @@ fn run_live_session(
session_id: u64, session_id: u64,
engine: Arc<LocalEngine>, engine: Arc<LocalEngine>,
config: StartLiveTranscriptionConfig, config: StartLiveTranscriptionConfig,
audio_path: Option<PathBuf>,
dictionary_terms: Vec<String>, dictionary_terms: Vec<String>,
result_channel: Channel<LiveResultMessage>, result_channel: Channel<LiveResultMessage>,
status_channel: Channel<LiveStatusMessage>, status_channel: Channel<LiveStatusMessage>,
@@ -358,11 +376,42 @@ fn run_live_session(
let mut resampler: Option<StreamingResampler> = None; let mut resampler: Option<StreamingResampler> = None;
let mut capture_buffer: Vec<f32> = Vec::new(); let mut capture_buffer: Vec<f32> = Vec::new();
let mut kept_audio = if config.save_audio {
Some(Vec::new()) // Progressive WAV writer (brief item #19). Sample rate comes from
// the loaded backend's capabilities (#13 wiring) so a future
// non-16kHz backend records at its native rate without further
// plumbing. The writer flushes its header every ~500 ms, so the
// file on disk is a playable WAV even if the process is killed.
let sample_rate = engine
.capabilities()
.map(|c| c.sample_rate)
.unwrap_or(WHISPER_SAMPLE_RATE);
let mut wav_writer: Option<WavWriter> = match audio_path.as_ref() {
Some(path) => match WavWriter::create(path, sample_rate, 1) {
Ok(w) => Some(w),
Err(e) => {
let _ = status_channel.send(LiveStatusMessage::Warning {
session_id,
message: format!(
"Failed to open audio recording file ({}); transcription will continue without saving audio.",
e
),
});
None
}
},
None => None,
};
// Resolve the reported audio_path only if the writer actually
// opened; otherwise the returned Summary must not claim a path.
let reported_audio_path: Option<String> = if wav_writer.is_some() {
audio_path
.as_ref()
.map(|p| p.to_string_lossy().to_string())
} else { } else {
None None
}; };
let mut buffer_start_sample: u64 = 0; let mut buffer_start_sample: u64 = 0;
let mut dropped_audio_ms: u64 = 0; let mut dropped_audio_ms: u64 = 0;
let mut chunk_id: u32 = 0; let mut chunk_id: u32 = 0;
@@ -412,7 +461,13 @@ fn run_live_session(
}; };
let resampled = resampler.push_samples(&mono).map_err(|e| e.to_string())?; let resampled = resampler.push_samples(&mono).map_err(|e| e.to_string())?;
append_resampled_audio(&mut capture_buffer, &mut kept_audio, &resampled); append_resampled_audio(
&mut capture_buffer,
&mut wav_writer,
&resampled,
session_id,
&status_channel,
);
} }
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
@@ -442,9 +497,25 @@ fn run_live_session(
if stopping && !resampler_flushed { if stopping && !resampler_flushed {
if let Some(resampler) = &mut resampler { if let Some(resampler) = &mut resampler {
let tail = resampler.flush().map_err(|e| e.to_string())?; let tail = resampler.flush().map_err(|e| e.to_string())?;
append_resampled_audio(&mut capture_buffer, &mut kept_audio, &tail); append_resampled_audio(
&mut capture_buffer,
&mut wav_writer,
&tail,
session_id,
&status_channel,
);
} }
resampler_flushed = true; resampler_flushed = true;
// Final flush for the WAV header so the last chunk's header
// update is on disk before we drop into the inference drain.
if let Some(writer) = wav_writer.as_mut() {
if let Err(e) = writer.flush() {
let _ = status_channel.send(LiveStatusMessage::Warning {
session_id,
message: format!("WAV flush failed near session end: {e}"),
});
}
}
} }
if inflight.is_none() { if inflight.is_none() {
@@ -481,25 +552,53 @@ fn run_live_session(
thread::sleep(Duration::from_millis(10)); thread::sleep(Duration::from_millis(10));
} }
// Finalise the progressive WAV writer so the terminal header
// reflects every written sample. A failure here leaves the last
// flushed header state on disk — still a playable WAV — so we
// report a warning but don't fail the whole session.
if let Some(writer) = wav_writer.take() {
if let Err(e) = writer.finalize() {
let _ = status_channel.send(LiveStatusMessage::Warning {
session_id,
message: format!("WAV finalise failed: {e}; partial recording is still playable."),
});
}
}
Ok(LiveSessionSummary { Ok(LiveSessionSummary {
session_id, session_id,
dropped_audio_ms, dropped_audio_ms,
audio_samples: kept_audio, audio_path: reported_audio_path,
}) })
} }
fn append_resampled_audio( fn append_resampled_audio(
capture_buffer: &mut Vec<f32>, capture_buffer: &mut Vec<f32>,
kept_audio: &mut Option<Vec<f32>>, wav_writer: &mut Option<WavWriter>,
resampled: &[f32], resampled: &[f32],
session_id: u64,
status_channel: &Channel<LiveStatusMessage>,
) { ) {
if resampled.is_empty() { if resampled.is_empty() {
return; return;
} }
capture_buffer.extend_from_slice(resampled); capture_buffer.extend_from_slice(resampled);
if let Some(kept_audio) = kept_audio { if let Some(writer) = wav_writer.as_mut() {
kept_audio.extend_from_slice(resampled); if let Err(e) = writer.append(resampled) {
// WAV write failure is non-fatal for live transcription —
// drop the writer so subsequent samples don't keep trying
// a broken file handle, and warn the user. The samples
// already written up to this point remain playable thanks
// to periodic header flushes.
let _ = status_channel.send(LiveStatusMessage::Warning {
session_id,
message: format!(
"Audio recording halted: {e}. Transcription continues; partial WAV is playable."
),
});
*wav_writer = None;
}
} }
} }