470 lines
18 KiB
Rust
470 lines
18 KiB
Rust
use std::path::PathBuf;
|
||
use std::sync::{Arc, Mutex};
|
||
|
||
use tauri::{Emitter, Manager};
|
||
use tokio::sync::{mpsc as tokio_mpsc, Mutex as AsyncMutex};
|
||
use tokio::task::JoinHandle;
|
||
|
||
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())
|
||
}
|
||
|
||
/// A running native-capture accumulator worker, held so the command
|
||
/// layer can both signal it to stop and `await` its termination. RB-06
|
||
/// replaced a fire-and-forget `tokio::spawn` that let the previous
|
||
/// worker keep flushing and appending samples after `stop_native_capture`
|
||
/// returned — a rapid start → stop → start could contaminate the new
|
||
/// session's samples vector with tail writes from the old one.
|
||
struct CaptureWorker {
|
||
stop_tx: tokio_mpsc::Sender<()>,
|
||
join: JoinHandle<()>,
|
||
}
|
||
|
||
/// Send the stop signal and await full worker termination. Consumes
|
||
/// `CaptureWorker` because the contained handles are single-use.
|
||
/// Errors from `join.await` (task panicked or was cancelled) are
|
||
/// logged and swallowed — the caller only needs the synchronisation
|
||
/// barrier, not the worker's return value.
|
||
async fn stop_worker(worker: CaptureWorker) {
|
||
let _ = worker.stop_tx.send(()).await;
|
||
drop(worker.stop_tx);
|
||
if let Err(e) = worker.join.await {
|
||
eprintln!("[native-capture] worker task did not terminate cleanly: {e}");
|
||
}
|
||
}
|
||
|
||
/// Shared state for native microphone capture.
|
||
pub struct NativeCaptureState {
|
||
/// The running accumulator worker, if any. `tokio::sync::Mutex`
|
||
/// because the fastest-moving consumer (`stop_worker`) awaits while
|
||
/// holding the lock — a `std::sync::Mutex` would have to be released
|
||
/// and reacquired around each await point.
|
||
worker: AsyncMutex<Option<CaptureWorker>>,
|
||
/// All captured samples (16kHz mono) for save_audio.
|
||
all_samples: Arc<Mutex<Vec<f32>>>,
|
||
}
|
||
|
||
impl NativeCaptureState {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
worker: AsyncMutex::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 in-flight worker and AWAIT its termination before opening
|
||
// a new capture. Without the join we would race a draining worker
|
||
// against the `all_samples.clear()` below, leaving old-session
|
||
// samples in the new-session vector (RB-06).
|
||
if let Some(existing) = state.worker.lock().await.take() {
|
||
stop_worker(existing).await;
|
||
}
|
||
|
||
// `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);
|
||
|
||
let all_samples_clone = all_samples.clone();
|
||
|
||
// Spawn a task that reads cpal chunks, downsamples to 16kHz mono,
|
||
// and emits events to the frontend. The JoinHandle is retained in
|
||
// `state.worker` so `stop_native_capture` can await full termination.
|
||
let join = 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();
|
||
}
|
||
});
|
||
|
||
*state.worker.lock().await = Some(CaptureWorker { stop_tx, join });
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Stop native microphone capture. Returns all captured samples (16kHz mono).
|
||
///
|
||
/// Awaits full worker termination before reading `all_samples`, so the
|
||
/// returned vector contains every sample the worker flushed — and
|
||
/// nothing from a worker that technically outlived the call (RB-06).
|
||
#[tauri::command]
|
||
pub async fn stop_native_capture(
|
||
state: tauri::State<'_, NativeCaptureState>,
|
||
) -> Result<Vec<f32>, String> {
|
||
if let Some(worker) = state.worker.lock().await.take() {
|
||
stop_worker(worker).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, stop_worker, CaptureWorker};
|
||
use std::sync::atomic::{AtomicU32, Ordering};
|
||
use std::sync::Arc;
|
||
use tokio::sync::mpsc;
|
||
|
||
#[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()));
|
||
}
|
||
|
||
// RB-06 regression: after `stop_worker(worker).await` completes, the
|
||
// underlying task must have exited — no lingering writes to shared
|
||
// state can leak past the stop point. The real native-capture
|
||
// worker drains a capture queue and appends to `all_samples`; this
|
||
// test swaps that for a synthetic worker that bumps an atomic
|
||
// counter in a loop and applies a distinct "flush" marker at exit.
|
||
// The assertions mirror the real-world invariant a caller needs:
|
||
// (a) after stop_worker returns, the worker has run its flush;
|
||
// (b) subsequent sleeps see the counter frozen — no writes occur
|
||
// after the join barrier.
|
||
// Pre-fix behaviour (fire-and-forget `tokio::spawn`) failed both:
|
||
// a start→stop→start cycle could observe tail writes from the
|
||
// previous worker in the new session's vector.
|
||
#[tokio::test]
|
||
async fn stop_worker_awaits_full_termination_no_writes_after_join() {
|
||
let counter = Arc::new(AtomicU32::new(0));
|
||
let counter_task = counter.clone();
|
||
let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
|
||
|
||
let join = tokio::spawn(async move {
|
||
loop {
|
||
if stop_rx.try_recv().is_ok() {
|
||
break;
|
||
}
|
||
counter_task.fetch_add(1, Ordering::SeqCst);
|
||
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
|
||
}
|
||
// Flush marker — mirrors the final pcm_buffer drain in the
|
||
// real worker. Setting a value with a distinctive high bit
|
||
// so the test can prove the flush ran.
|
||
counter_task.fetch_or(0x8000_0000, Ordering::SeqCst);
|
||
});
|
||
|
||
// Let the worker accumulate a few bumps before we signal stop.
|
||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||
|
||
stop_worker(CaptureWorker { stop_tx, join }).await;
|
||
|
||
let after_stop = counter.load(Ordering::SeqCst);
|
||
assert!(
|
||
after_stop & 0x8000_0000 != 0,
|
||
"flush marker must be set post-stop (got {after_stop:#x})"
|
||
);
|
||
|
||
// Post-join, no further writes are possible because the task
|
||
// has ended. Sleep briefly and re-read to confirm.
|
||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||
let later = counter.load(Ordering::SeqCst);
|
||
assert_eq!(
|
||
later, after_stop,
|
||
"no writes must happen after stop_worker returns"
|
||
);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn stop_worker_is_idempotent_on_a_worker_that_has_already_exited() {
|
||
// A worker that stops itself (channel disconnected, capture
|
||
// dead, etc.) must still be join-able cleanly by stop_worker —
|
||
// the helper should swallow any expected "task already done"
|
||
// condition without panicking.
|
||
let (stop_tx, _stop_rx) = mpsc::channel::<()>(1);
|
||
let join = tokio::spawn(async { /* exit immediately */ });
|
||
|
||
// Give the task a tick to finish.
|
||
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
||
|
||
// This must not hang or panic.
|
||
stop_worker(CaptureWorker { stop_tx, join }).await;
|
||
}
|
||
}
|
||
|
||
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
|
||
}
|