Jake
fd24b81a5f
MINOR from the batch review of 6e9ed99: SystemTime::now() alone
cannot guarantee uniqueness under tight loops — two calls in the
same clock tick can return identical secs + nanos on some OS
timing resolutions. The filename reduction from "every second"
to "every nanosecond" addresses the flagged bug but leaves a
theoretical gap.
Adds a process-lifetime AtomicU64 counter, zero-padded to 4 digits,
as the third filename component. New shape:
kon-<secs>-<nanos_in_sec>-<counter>.wav
e.g. kon-1776828000-123456789-0000.wav
Across process restarts the counter resets to 0, but the wall-clock
secs/nanos have advanced — no cross-launch collisions possible.
Within a single process, the counter guarantees uniqueness regardless
of clock behaviour.
Test strengthened from ">=32 of 64 unique" (probabilistic) to
"1024 of 1024 unique" (absolute).
360 lines
13 KiB
Rust
360 lines
13 KiB
Rust
use std::path::PathBuf;
|
||
use std::sync::{Arc, Mutex};
|
||
|
||
use tauri::{Emitter, Manager};
|
||
use tokio::sync::mpsc as tokio_mpsc;
|
||
|
||
use kon_audio::{DeviceInfo, MicrophoneCapture};
|
||
use kon_core::constants::WHISPER_SAMPLE_RATE;
|
||
use kon_core::types::AudioSamples;
|
||
|
||
/// Enumerate every input device available to cpal, with metadata for the
|
||
/// Settings device-picker UI. Includes a flag for likely PulseAudio /
|
||
/// PipeWire monitor sources so the UI can warn the user.
|
||
#[tauri::command]
|
||
pub async fn list_audio_devices() -> Result<Vec<DeviceInfo>, String> {
|
||
tokio::task::spawn_blocking(MicrophoneCapture::list_devices)
|
||
.await
|
||
.map_err(|e| format!("join error: {e}"))?
|
||
.map_err(|e| e.to_string())
|
||
}
|
||
|
||
/// Shared state for native microphone capture.
|
||
pub struct NativeCaptureState {
|
||
/// Stop signal sender — dropping this stops the accumulator task.
|
||
stop_tx: Mutex<Option<tokio_mpsc::Sender<()>>>,
|
||
/// All captured samples (16kHz mono) for save_audio.
|
||
all_samples: Arc<Mutex<Vec<f32>>>,
|
||
}
|
||
|
||
impl NativeCaptureState {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
stop_tx: Mutex::new(None),
|
||
all_samples: Arc::new(Mutex::new(Vec::new())),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Start native microphone capture via cpal.
|
||
/// Streams 16kHz mono PCM chunks to the frontend via `native-pcm` events.
|
||
///
|
||
/// `device_name`: explicit device name (from `list_audio_devices`) or None / ""
|
||
/// to auto-select. The frontend passes `settings.microphoneDevice` here so the
|
||
/// user's pick from Settings → Audio → Microphone takes effect.
|
||
#[tauri::command]
|
||
pub async fn start_native_capture(
|
||
app: tauri::AppHandle,
|
||
state: tauri::State<'_, NativeCaptureState>,
|
||
device_name: Option<String>,
|
||
) -> Result<(), String> {
|
||
eprintln!(
|
||
"[native-capture] start_native_capture called (device='{}')",
|
||
device_name.as_deref().unwrap_or("<auto>")
|
||
);
|
||
|
||
// 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);
|
||
}
|
||
|
||
// `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()
|
||
})?;
|
||
eprintln!(
|
||
"[native-capture] cpal capture started successfully on '{}'",
|
||
capture.device_name
|
||
);
|
||
|
||
// Wrap capture in Arc<Mutex> so it can be moved into the blocking task
|
||
let capture = Arc::new(Mutex::new(Some(capture)));
|
||
let capture_clone = capture.clone();
|
||
|
||
let all_samples = state.all_samples.clone();
|
||
all_samples.lock().unwrap().clear();
|
||
|
||
let (stop_tx, mut stop_rx) = tokio_mpsc::channel::<()>(1);
|
||
*state.stop_tx.lock().unwrap() = Some(stop_tx);
|
||
|
||
let all_samples_clone = all_samples.clone();
|
||
|
||
// Spawn a task that reads cpal chunks, downsamples to 16kHz mono,
|
||
// and emits events to the frontend
|
||
tokio::spawn(async move {
|
||
let mut pcm_buffer: Vec<f32> = Vec::new();
|
||
let chunk_size = 8000_usize; // ~0.5s at 16kHz
|
||
|
||
loop {
|
||
// Check for stop signal (non-blocking)
|
||
if stop_rx.try_recv().is_ok() {
|
||
break;
|
||
}
|
||
|
||
// 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;
|
||
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
|
||
};
|
||
|
||
// 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;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Emit chunks to frontend when we have enough
|
||
while pcm_buffer.len() >= chunk_size {
|
||
let chunk: Vec<f32> = pcm_buffer.drain(..chunk_size).collect();
|
||
|
||
// Store for save_audio
|
||
if let Ok(mut all) = all_samples_clone.lock() {
|
||
all.extend_from_slice(&chunk);
|
||
}
|
||
|
||
let _ = app.emit(
|
||
"native-pcm",
|
||
serde_json::json!({
|
||
"samples": chunk,
|
||
}),
|
||
);
|
||
}
|
||
|
||
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;
|
||
}
|
||
}
|
||
|
||
// Emit any remaining samples
|
||
if !pcm_buffer.is_empty() {
|
||
if let Ok(mut all) = all_samples_clone.lock() {
|
||
all.extend_from_slice(&pcm_buffer);
|
||
}
|
||
let _ = app.emit(
|
||
"native-pcm",
|
||
serde_json::json!({
|
||
"samples": pcm_buffer,
|
||
}),
|
||
);
|
||
}
|
||
|
||
// Drop the capture to stop the cpal stream
|
||
if let Ok(mut cap) = capture_clone.lock() {
|
||
cap.take();
|
||
}
|
||
});
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Stop native microphone capture. Returns all captured samples (16kHz mono).
|
||
#[tauri::command]
|
||
pub async fn stop_native_capture(
|
||
state: tauri::State<'_, NativeCaptureState>,
|
||
) -> Result<Vec<f32>, String> {
|
||
// Extract the stop sender without holding the guard across an await
|
||
let stop_tx = state.stop_tx.lock().unwrap().take();
|
||
if let Some(tx) = stop_tx {
|
||
let _ = tx.send(()).await;
|
||
}
|
||
|
||
// Brief delay to let the accumulator flush
|
||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||
|
||
let samples = {
|
||
let mut all = state.all_samples.lock().unwrap();
|
||
std::mem::take(&mut *all)
|
||
};
|
||
|
||
Ok(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,
|
||
output_folder: Option<&str>,
|
||
) -> Result<PathBuf, String> {
|
||
let recordings_dir = match output_folder.map(str::trim).filter(|s| !s.is_empty()) {
|
||
Some(folder) => PathBuf::from(folder),
|
||
None => app
|
||
.path()
|
||
.app_local_data_dir()
|
||
.map_err(|e: tauri::Error| e.to_string())?
|
||
.join("recordings"),
|
||
};
|
||
|
||
std::fs::create_dir_all(&recordings_dir)
|
||
.map_err(|e| format!("Failed to create recordings dir: {e}"))?;
|
||
|
||
Ok(recordings_dir.join(recording_filename()))
|
||
}
|
||
|
||
/// Deterministic recording filename generator. Combines three fields
|
||
/// for absolute uniqueness across rapid calls:
|
||
///
|
||
/// - wall-clock seconds since the epoch — human-readable and
|
||
/// sortable;
|
||
/// - the sub-second nanosecond component — defeats same-second
|
||
/// collisions;
|
||
/// - a process-lifetime atomic counter — defeats even same-nanosecond
|
||
/// collisions, which `SystemTime::now()` alone cannot guarantee
|
||
/// (two calls in the same clock tick can return identical nanos).
|
||
///
|
||
/// Format: `kon-<secs>-<nanos_in_sec>-<counter>.wav`, e.g.
|
||
/// `kon-1776828000-123456789-0000.wav`.
|
||
fn recording_filename() -> String {
|
||
let duration = std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.unwrap_or_default();
|
||
let secs = duration.as_secs();
|
||
let nanos = duration.subsec_nanos();
|
||
let counter = RECORDING_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||
format!("kon-{secs}-{nanos:09}-{counter:04}.wav")
|
||
}
|
||
|
||
/// Process-lifetime monotonic counter for `recording_filename`. Starts
|
||
/// at 0 on each Kon launch; wall-clock secs/nanos still advance across
|
||
/// restarts, so cross-launch collisions are already impossible — the
|
||
/// counter is the last-mile guarantee against within-launch same-tick
|
||
/// collisions.
|
||
static RECORDING_COUNTER: std::sync::atomic::AtomicU64 =
|
||
std::sync::atomic::AtomicU64::new(0);
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::recording_filename;
|
||
|
||
#[test]
|
||
fn recording_filenames_are_unique_across_rapid_calls() {
|
||
// Regression for the 2026-04-22 review AND the review-of-
|
||
// review MINOR: SystemTime::now() alone cannot guarantee
|
||
// uniqueness under tight loops on every OS clock resolution,
|
||
// so the filename now includes a process-lifetime atomic
|
||
// counter. With the counter, uniqueness is absolute across
|
||
// any number of in-process calls.
|
||
let mut names = std::collections::HashSet::new();
|
||
for _ in 0..1024 {
|
||
names.insert(recording_filename());
|
||
}
|
||
assert_eq!(
|
||
names.len(),
|
||
1024,
|
||
"every filename must be unique (counter-backed guarantee)"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn recording_filename_has_expected_shape() {
|
||
let name = recording_filename();
|
||
assert!(name.starts_with("kon-"));
|
||
assert!(name.ends_with(".wav"));
|
||
// Shape: kon-<digits>-<9 digits>-<>=4 digits>.wav
|
||
let rest = name
|
||
.strip_prefix("kon-")
|
||
.and_then(|s| s.strip_suffix(".wav"))
|
||
.expect("shape prefix/suffix");
|
||
let parts: Vec<&str> = rest.split('-').collect();
|
||
assert_eq!(
|
||
parts.len(),
|
||
3,
|
||
"expected three '-' separated parts, got {parts:?}"
|
||
);
|
||
assert!(parts[0].chars().all(|c| c.is_ascii_digit()), "secs is digits");
|
||
assert_eq!(parts[1].len(), 9, "nanos component is zero-padded to 9 digits");
|
||
assert!(parts[1].chars().all(|c| c.is_ascii_digit()));
|
||
assert!(parts[2].len() >= 4, "counter component is zero-padded to >=4 digits");
|
||
assert!(parts[2].chars().all(|c| c.is_ascii_digit()));
|
||
}
|
||
}
|
||
|
||
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();
|
||
|
||
tokio::task::spawn_blocking(move || {
|
||
let audio = AudioSamples::mono_16khz(samples);
|
||
kon_audio::write_wav(&path_clone, &audio).map_err(|e| e.to_string())
|
||
})
|
||
.await
|
||
.map_err(|e| e.to_string())??;
|
||
|
||
Ok(path.to_string_lossy().to_string())
|
||
}
|
||
|
||
/// Save PCM f32 samples as a WAV file. Returns the file path.
|
||
#[tauri::command]
|
||
pub async fn save_audio(
|
||
app: tauri::AppHandle,
|
||
samples: Vec<f32>,
|
||
output_folder: Option<String>,
|
||
) -> Result<String, String> {
|
||
persist_audio_samples(&app, samples, output_folder).await
|
||
}
|