#![allow(clippy::too_many_arguments)] use std::collections::{HashMap, VecDeque}; use std::path::PathBuf; use std::sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, Arc, Mutex, }; use std::thread; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use tauri::ipc::Channel; use tokio::sync::Mutex as AsyncMutex; use crate::commands::audio::resolve_recording_path; use crate::commands::build_initial_prompt; use crate::commands::models::{default_model_id_for_engine, ensure_model_loaded}; use crate::commands::power::PowerAssertion; use crate::commands::security::ensure_main_window; use crate::AppState; use lumotia_ai_formatting::{post_process_segments, FormatMode, PostProcessOptions}; use lumotia_audio::{ AudioChunk, CaptureRuntimeError, MicrophoneCapture, StreamingResampler, WavWriter, }; use lumotia_core::constants::WHISPER_SAMPLE_RATE; use lumotia_core::types::{AudioSamples, Segment, TranscriptionOptions}; use lumotia_transcription::LocalEngine; const CHUNK_SAMPLES: usize = 32_000; // 2s at 16kHz const OVERLAP_SAMPLES: usize = 4_000; // 0.25s at 16kHz const FINAL_CHUNK_MIN_SAMPLES: usize = 4_000; // 0.25s const MAX_PENDING_SAMPLES: usize = CHUNK_SAMPLES; /// Headroom multiplier for `drain_inference`'s timeout, derived from the /// in-flight chunk's `duration_secs`. F3's original derivation used the /// constant CHUNK_SAMPLES and capped at ~6s, which broke for the /// realistic case of a slow CPU + large-v3 model running 3-5x realtime /// on a chunk that happens to be SHORTER than CHUNK_SAMPLES — the /// timeout would fire before the legitimate inference finished, treating /// healthy work as a wedge. 5x realtime is the documented upper bound /// for the slowest supported configuration (whisper.cpp large-v3, CPU /// only, low-end laptop). Anything beyond that is a genuine wedge worth /// aborting. const REALTIME_SAFETY_MULTIPLIER: u32 = 5; /// Floor on the drain timeout. Even a sub-second tail chunk needs more /// than its duration's worth of wall-clock to clear the decoder (model /// load amortisation, OS scheduling jitter, the abort callback's own /// poll cadence). 2s is conservative enough to never fire spuriously on /// healthy backends and short enough to keep the lifecycle responsive. const DRAIN_TIMEOUT_FLOOR: Duration = Duration::from_secs(2); const SPEECH_FRAME_SAMPLES: usize = 800; // 50ms const MIN_SPEECH_FRAMES: usize = 1; // any plausible speech-like frame const SILENCE_RMS_THRESHOLD: f32 = 0.001; const SPEECH_WINDOW_RMS_THRESHOLD: f32 = 0.0014; const SPEECH_WINDOW_PEAK_THRESHOLD: f32 = 0.004; const STRONG_SPEECH_RMS_THRESHOLD: f32 = 0.003; const STRONG_SPEECH_PEAK_THRESHOLD: f32 = 0.012; const FLATLINE_PEAK_THRESHOLD: f32 = 0.0005; const DUPLICATE_TRANSCRIPT_WINDOW_SECS: f64 = 6.0; const DUPLICATE_TRANSCRIPT_MERGE_LIMIT: usize = 3; const DUPLICATE_HISTORY_RETENTION_SECS: f64 = 8.0; const DUPLICATE_CHECK_LEADING_SECS: f64 = 1.5; const TOKEN_COVERAGE_THRESHOLD: f64 = 0.6; const TOKEN_SEQUENCE_THRESHOLD: f64 = 0.6; const MIN_TOKENS_FOR_OVERLAP: usize = 3; const MEANINGFUL_TOKEN_COVERAGE_THRESHOLD: f64 = 0.55; const MEANINGFUL_TOKEN_SEQUENCE_THRESHOLD: f64 = 0.55; const MIN_MEANINGFUL_TOKENS_FOR_OVERLAP: usize = 4; const LOW_SIGNAL_TOKENS: &[&str] = &[ "a", "an", "and", "are", "as", "at", "be", "been", "being", "but", "by", "d", "did", "do", "does", "for", "from", "had", "has", "have", "he", "her", "here", "his", "how", "i", "if", "in", "is", "it", "ll", "m", "me", "my", "of", "on", "or", "our", "out", "re", "s", "she", "so", "t", "that", "the", "their", "them", "there", "these", "they", "this", "those", "to", "ve", "was", "we", "well", "were", "what", "when", "where", "which", "who", "why", "with", "without", "you", "your", ]; #[derive(Default)] pub struct LiveTranscriptionState { next_session_id: AtomicU64, lifecycle: AsyncMutex<()>, running: Mutex>, } struct RunningLiveSession { id: u64, stop_flag: Arc, handle: tokio::task::JoinHandle>, status_channel: Channel, } #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub struct StartLiveTranscriptionConfig { pub engine: String, pub model_id: Option, pub language: Option, pub initial_prompt: Option, pub save_audio: bool, pub output_folder: Option, pub remove_fillers: bool, pub british_english: bool, pub anti_hallucination: bool, pub format_mode: String, /// Optional explicit microphone device name (from `list_audio_devices`). /// None or empty string = let `MicrophoneCapture::start` auto-select. pub microphone_device: Option, /// Optional profile id. None falls back to `DEFAULT_PROFILE_ID`. Drives /// the post-processing dictionary via `profile_terms` and, when the /// request's `initial_prompt` is empty, supplies the Whisper prompt. pub profile_id: Option, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct StartLiveTranscriptionResponse { pub session_id: u64, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct StopLiveTranscriptionResponse { pub session_id: u64, pub audio_path: Option, pub dropped_audio_ms: u64, } #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct LiveResultMessage { pub session_id: u64, pub chunk_id: u32, pub chunk_start_secs: f64, pub duration: f64, pub language: String, pub inference_ms: u64, pub segments: Vec, /// Concatenated text BEFORE post-processing (no filler removal, no /// British conversion, no LLM cleanup). Used by the transcription /// preview overlay so the user can see raw Whisper output as it /// streams in. pub raw_text: String, } #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase", tag = "type")] pub enum LiveStatusMessage { Warning { session_id: u64, message: String, }, Overload { session_id: u64, dropped_audio_ms: u64, message: String, }, Error { session_id: u64, message: String, }, Finished { session_id: u64, audio_path: Option, dropped_audio_ms: u64, }, } struct LiveSessionSummary { session_id: u64, dropped_audio_ms: u64, /// 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, } /// Session worker state is thread-confined to the single blocking task spawned /// by `start_live_transcription_session`. Cross-thread coordination happens via /// the stop flag and mpsc channels only; RB-01 will tighten the outer /// `live_state.running` lock discipline now that this worker lifecycle is /// explicit and locally structured. struct ActiveCapture { /// Keeping the capture handle alive keeps the underlying cpal stream alive. 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, 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, 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, status_channel: &Channel, ) { let Some(err_rx) = &self.mic_error_rx else { return; }; while let Ok(err) = err_rx.try_recv() { let _ = status_channel.send(LiveStatusMessage::Warning { session_id, message: format!( "Microphone '{}' reported an error: {}", err.device_name, err.message ), }); } } } #[derive(Default)] struct LiveLoopState { resampler: Option, capture_buffer: Vec, wav_writer: Option, buffer_start_sample: u64, dropped_audio_ms: u64, chunk_id: u32, inflight: Option, 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 { fn new(wav_writer: Option) -> Self { Self { wav_writer, ..Self::default() } } } struct LiveSessionRuntime { session_id: u64, engine: Arc, config: StartLiveTranscriptionConfig, audio_path: Option, dictionary_terms: Vec, result_channel: Channel, status_channel: Channel, stop_flag: Arc, capture: ActiveCapture, state: LiveLoopState, } impl LiveSessionRuntime { fn new( session_id: u64, engine: Arc, config: StartLiveTranscriptionConfig, audio_path: Option, dictionary_terms: Vec, result_channel: Channel, status_channel: Channel, stop_flag: Arc, ) -> Result { let capture = ActiveCapture::start(&config)?; let wav_writer = open_wav_writer(&engine, audio_path.as_ref(), session_id, &status_channel); Ok(Self { session_id, engine, config, audio_path, dictionary_terms, result_channel, status_channel, stop_flag, capture, state: LiveLoopState::new(wav_writer), }) } fn run(mut self) -> Result { loop { self.poll_inference()?; self.capture .drain_runtime_errors(self.session_id, &self.status_channel); 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() { continue; } if self.should_exit_loop() { break; } } self.drain_inference()?; self.finish() } fn poll_inference(&mut self) -> Result<(), String> { let _ = poll_inference( &mut self.state.inflight, &mut self.state.result_listener_lost, self.session_id, &self.config, &mut self.state.recent_segments, &self.dictionary_terms, &self.result_channel, &self.status_channel, &self.stop_flag, )?; Ok(()) } 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), Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { let message = "Microphone capture disconnected unexpectedly".to_string(); let _ = self.status_channel.send(LiveStatusMessage::Error { session_id: self.session_id, message: message.clone(), }); Err(message) } } } 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, None => { self.state.resampler = Some(StreamingResampler::new(chunk.sample_rate).map_err(|e| e.to_string())?); self.state.resampler.as_mut().expect("resampler just set") } }; let resampled = resampler.push_samples(&mono).map_err(|e| e.to_string())?; append_resampled_audio( &mut self.state.capture_buffer, &mut self.state.wav_writer, &resampled, self.session_id, &self.status_channel, ); 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; } let overflow = self.state.capture_buffer.len() - MAX_PENDING_SAMPLES; self.state.capture_buffer.drain(..overflow); self.state.buffer_start_sample = self .state .buffer_start_sample .saturating_add(overflow as u64); self.state.dropped_audio_ms = self .state .dropped_audio_ms .saturating_add((overflow as u64 * 1000) / WHISPER_SAMPLE_RATE as u64); let _ = self.status_channel.send(LiveStatusMessage::Overload { session_id: self.session_id, dropped_audio_ms: self.state.dropped_audio_ms, message: "Lumotia dropped older audio to keep live dictation responsive".into(), }); } fn flush_tail_if_stopping(&mut self) -> Result<(), String> { if !self.stopping() || self.state.resampler_flushed { return Ok(()); } if let Some(resampler) = &mut self.state.resampler { let tail = resampler.flush().map_err(|e| e.to_string())?; append_resampled_audio( &mut self.state.capture_buffer, &mut self.state.wav_writer, &tail, self.session_id, &self.status_channel, ); } self.flush_wav_header(); self.state.resampler_flushed = true; Ok(()) } fn flush_wav_header(&mut self) { let Some(writer) = self.state.wav_writer.as_mut() else { return; }; if let Err(e) = writer.flush() { let _ = self.status_channel.send(LiveStatusMessage::Warning { session_id: self.session_id, message: format!("WAV flush failed near session end: {e}"), }); } } fn dispatch_inference_if_ready(&mut self) -> bool { if self.state.inflight.is_some() { return false; } let stopping = self.stopping(); if let Some(task) = maybe_dispatch_chunk( &self.engine, &self.config, &mut self.state.capture_buffer, &mut self.state.buffer_start_sample, &mut self.state.chunk_id, stopping, &self.status_channel, self.session_id, ) { self.state.inflight = Some(task); return true; } false } fn stopping(&self) -> bool { self.stop_flag.load(Ordering::Relaxed) } fn should_exit_loop(&self) -> bool { self.stopping() && self.state.resampler_flushed && self.state.inflight.is_none() } fn drain_inference(&mut self) -> Result<(), String> { // Bound the wait so a wedged whisper (ggml deadlock, GPU stall, // any pathology that turns transcribe_sync into a black hole) // can't brick the live-session lifecycle and, via the outer // AsyncMutex, every subsequent start/stop. The timeout pairs // with the per-task `abort_flag` plumbed into whisper-rs's // abort callback — on expiry we set the flag and drop our // receiver, so the worker either honours the abort or exits // silently when it eventually tries to send its result. // // Budget is derived from the IN-FLIGHT task's own duration_secs // rather than the constant CHUNK_SAMPLES so a short tail chunk // running through a slow CPU/model combination (large-v3 at // 3-5x realtime is the documented upper bound) isn't aborted // mid-flight while it's still doing legitimate work. The // multiplier + floor gives us safety on both ends. let drain_timeout = drain_timeout_for_inflight(self.state.inflight.as_ref()); let drain_timeout_ms = drain_timeout.as_millis() as u64; let drain_started = Instant::now(); while self.state.inflight.is_some() { self.poll_inference()?; if self.state.inflight.is_none() { break; } if drain_started.elapsed() >= drain_timeout { // SAFETY: a real test of this path requires a wedged // whisper-rs, which is hard to fixture. We rely on the // abort_callback wiring plus cargo build + manual smoke // for verification. The timeout fires regardless, so // the lifecycle deadlock is broken either way. if let Some(task) = self.state.inflight.as_ref() { task.abort_flag.store(true, Ordering::Relaxed); // Targeting convention: implicit module-path target // (`lumotia_lib::commands::live`) so the operator's // `RUST_LOG=…,lumotia_lib::commands::live=debug` filter // covers this emit. A previous `target: "lumotia_live"` // literal slipped past EnvFilter (which matches on // `::`-segments, not substrings) and silenced this // warning during the operator's documented triage. tracing::warn!( session_id = self.session_id, chunk_id = task.chunk_id, timeout_ms = drain_timeout_ms, chunk_duration_secs = task.duration_secs, inflight_age_ms = task.dispatched_at.elapsed().as_millis() as u64, "drained inflight inference after timeout" ); } let _ = self.state.inflight.take(); let message = format!( "Inference timed out after {drain_timeout_ms}ms; live session stopping cleanly" ); let _ = self.status_channel.send(LiveStatusMessage::Error { session_id: self.session_id, message: message.clone(), }); return Err(message); } thread::sleep(Duration::from_millis(10)); } Ok(()) } fn finish(mut self) -> Result { let audio_path = finalize_wav_writer( self.state.wav_writer.take(), self.audio_path.as_ref(), self.session_id, &self.status_channel, ); Ok(LiveSessionSummary { session_id: self.session_id, dropped_audio_ms: self.state.dropped_audio_ms, audio_path, }) } } struct InferenceTask { chunk_id: u32, chunk_start_sample: u64, trim_before_secs: f64, duration_secs: f64, rx: std::sync::mpsc::Receiver>, /// Per-task abort flag. Set by `drain_inference` on timeout or by /// any early worker exit; polled by whisper-rs through /// `set_abort_callback_safe`. The Arc is shared with the spawned /// inference thread so the cancellation route survives a dropped /// `InferenceTask` (e.g. when the loop returns through `?`). abort_flag: Arc, /// Instant the inference thread was spawned. Used by /// `drain_inference` to compute a bounded wait based on the chunk /// duration. None for tasks dispatched before wall-clock tracking /// was required (kept Option to keep the test fixtures concise). dispatched_at: Instant, } impl Drop for InferenceTask { fn drop(&mut self) { // Any path that drops the task without explicitly clearing it // (early `?` propagation in the loop, panic unwind) signals the // worker thread to exit. Without this the orphaned thread keeps // running whisper-rs against the engine's Mutex; N orphans on // a reconnect/retry loop contend on the backend and stall every // future inference. self.abort_flag.store(true, Ordering::Relaxed); } } /// Compute the drain-inference timeout for the in-flight task. Derived /// from `task.duration_secs * REALTIME_SAFETY_MULTIPLIER` with a floor /// at `DRAIN_TIMEOUT_FLOOR` so even a sub-second tail chunk gets enough /// wall-clock to clear the decoder before we treat it as a wedge. /// /// `None` means there's nothing in flight — return the floor purely so /// the drain loop has a well-defined fallback; in practice the outer /// `while self.state.inflight.is_some()` short-circuits before the /// timeout is consulted. fn drain_timeout_for_inflight(task: Option<&InferenceTask>) -> Duration { let Some(task) = task else { return DRAIN_TIMEOUT_FLOOR; }; // Treat non-finite or non-positive durations as "use the floor" — // we never want a NaN or negative to silently produce a tiny or // overflowing budget. let duration_secs = task.duration_secs; if !duration_secs.is_finite() || duration_secs <= 0.0 { return DRAIN_TIMEOUT_FLOOR; } let scaled_secs = duration_secs * REALTIME_SAFETY_MULTIPLIER as f64; let scaled = Duration::from_secs_f64(scaled_secs); scaled.max(DRAIN_TIMEOUT_FLOOR) } #[derive(Debug, Clone)] struct RecentTranscriptSegment { start_secs: f64, end_secs: f64, text: String, } #[derive(Debug, Clone, Copy, Default)] struct SpeechGateState { peak_rms: f32, peak_amplitude: f32, window_count: usize, speech_window_count: usize, consecutive_speech_windows: usize, max_consecutive_speech_windows: usize, } #[derive(Debug, Clone, Copy, PartialEq)] struct SpeechGateDecision { skip: bool, reason: &'static str, peak_rms: f32, peak_amplitude: f32, window_count: usize, speech_window_count: usize, max_consecutive_speech_windows: usize, } #[tauri::command] pub async fn start_live_transcription_session( window: tauri::WebviewWindow, app: tauri::AppHandle, state: tauri::State<'_, AppState>, live_state: tauri::State<'_, LiveTranscriptionState>, mut config: StartLiveTranscriptionConfig, result_channel: Channel, status_channel: Channel, ) -> Result { ensure_main_window(&window)?; // Phase 1: acquire the lifecycle lock long enough to reserve the // single live-session slot. Held across `ensure_model_loaded` // because we want start/stop concurrency to remain serialised; the // dangerous pattern (lock held across a JoinHandle.await) was on // the stop path, not here. Released explicitly before the // RunningLiveSession is installed in `live_state.running` so the // symmetric stop path doesn't observe a half-initialised state. let lifecycle = live_state.lifecycle.lock().await; { let running = live_state.running.lock().unwrap(); if running.is_some() { return Err("A live transcription session is already running".into()); } } let resolved_profile_id = config .profile_id .clone() .unwrap_or_else(|| lumotia_storage::DEFAULT_PROFILE_ID.to_string()); let profile = lumotia_storage::database::get_profile(&state.db, &resolved_profile_id) .await .map_err(|e| e.to_string())? .ok_or_else(|| format!("Profile {resolved_profile_id} not found"))?; let profile_terms: Vec = lumotia_storage::database::list_profile_terms(&state.db, &resolved_profile_id) .await .map_err(|e| e.to_string())? .into_iter() .map(|t| t.term) .collect(); // Collapse the effective initial_prompt on the struct so downstream // `TranscriptionOptions` construction (see `maybe_dispatch_chunk`) picks // up profile fallback + vocabulary injection without further plumbing. let request_prompt = config.initial_prompt.clone().unwrap_or_default(); config.initial_prompt = build_initial_prompt(&request_prompt, &profile.initial_prompt, &profile_terms); let model_id = config .model_id .clone() .unwrap_or_else(|| default_model_id_for_engine(&config.engine).to_string()); tracing::info!( engine = %config.engine, model = %model_id, language = ?config.language, save_audio = config.save_audio, "starting live session" ); // None: live-transcription model loads don't enforce sequential-GPU // mode. The Settings-level load flow owns that guard. ensure_model_loaded(&state, &config.engine, &model_id, None).await?; let session_id = live_state .next_session_id .fetch_add(1, Ordering::Relaxed) .saturating_add(1); let stop_flag = Arc::new(AtomicBool::new(false)); let engine = pick_engine(&state, &config.engine)?; // 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_status = status_channel.clone(); let worker_results = result_channel.clone(); let dictionary_terms = profile_terms.clone(); let worker_audio_path = audio_path.clone(); let handle = tokio::task::spawn_blocking(move || { run_live_session( session_id, engine, config, worker_audio_path, dictionary_terms, worker_results, worker_status, worker_stop, ) }); *live_state.running.lock().unwrap() = Some(RunningLiveSession { id: session_id, stop_flag, handle, status_channel, }); // Phase 2: drop the lifecycle guard explicitly before returning so // a concurrent stop call can grab the lock without contention now // that the new RunningLiveSession is installed. drop(lifecycle); Ok(StartLiveTranscriptionResponse { session_id }) } #[tauri::command] pub async fn stop_live_transcription_session( window: tauri::WebviewWindow, app: tauri::AppHandle, live_state: tauri::State<'_, LiveTranscriptionState>, session_id: u64, ) -> Result { ensure_main_window(&window)?; // Take the running session and signal stop while holding the // lifecycle guard — but release the guard BEFORE awaiting the // worker's JoinHandle. The worker is a non-cancellable // spawn_blocking task; awaiting it under the lifecycle AsyncMutex // means a single wedged session blocks every future start/stop // until the OS thread eventually exits. Releasing here keeps the // mutual-exclusion invariant on the `running` slot (we already // took it out) without leaking the worker's wedge into the // lifecycle gate. let (handle, stop_flag, status_channel) = { let _lifecycle = live_state.lifecycle.lock().await; let running = live_state.running.lock().unwrap().take(); let Some(running) = running else { return Err("No live transcription session is running".into()); }; if running.id != session_id { *live_state.running.lock().unwrap() = Some(running); return Err(format!("Session {session_id} is not active")); } running.stop_flag.store(true, Ordering::Relaxed); (running.handle, running.stop_flag, running.status_channel) }; // `_lifecycle` dropped — other start/stop calls can now claim it // even if this handle.await is slow. let _ = stop_flag; let summary = handle .await .map_err(|e| format!("Live session task failed: {e}"))??; // Progressive WAV writer (brief item #19) wrote samples to disk // throughout the session; the path is already finalised. Nothing // further to persist. let _ = app; let audio_path = summary.audio_path; let response = StopLiveTranscriptionResponse { session_id: summary.session_id, audio_path: audio_path.clone(), dropped_audio_ms: summary.dropped_audio_ms, }; let _ = status_channel.send(LiveStatusMessage::Finished { session_id: summary.session_id, audio_path, dropped_audio_ms: summary.dropped_audio_ms, }); Ok(response) } fn pick_engine(state: &AppState, engine: &str) -> Result, String> { match engine { "whisper" => Ok(state.whisper_engine.clone()), "parakeet" => Ok(state.parakeet_engine.clone()), other => Err(format!("Unknown engine: {other}")), } } // instrument: attaches a `run_live_session` span carrying `session_id` // to every emit inside the `spawn_blocking` worker. Without this the // worker's tracing events landed in the runtime's root span and the // operator could not correlate per-chunk warnings back to the start // call that owned them. #[tracing::instrument( skip_all, fields(session_id = %session_id, engine = %config.engine, language = ?config.language) )] fn run_live_session( session_id: u64, engine: Arc, config: StartLiveTranscriptionConfig, audio_path: Option, dictionary_terms: Vec, result_channel: Channel, status_channel: Channel, stop_flag: Arc, ) -> Result { // macOS: disable App Nap while recording. On every other OS this // is a no-op. Keeping the guard in scope ties the assertion's // lifetime to the session — when the function returns, the Drop // impl lifts it. Item #9 in docs/whisper-ecosystem/brief.md. let _power_guard = PowerAssertion::begin("lumotia live dictation session"); LiveSessionRuntime::new( session_id, engine, config, audio_path, dictionary_terms, result_channel, status_channel, stop_flag, )? .run() } fn open_wav_writer( engine: &Arc, audio_path: Option<&PathBuf>, session_id: u64, status_channel: &Channel, ) -> Option { let sample_rate = engine .capabilities() .map(|c| c.sample_rate) .unwrap_or(WHISPER_SAMPLE_RATE); let path = audio_path?; match WavWriter::create(path, sample_rate, 1) { Ok(writer) => Some(writer), 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 } } } fn finalize_wav_writer( wav_writer: Option, audio_path: Option<&PathBuf>, session_id: u64, status_channel: &Channel, ) -> Option { match wav_writer { Some(writer) => match writer.finalize() { Ok(()) => audio_path.map(|path| path.to_string_lossy().to_string()), Err(e) => { let _ = status_channel.send(LiveStatusMessage::Warning { session_id, message: format!( "WAV finalise failed: {e}. A partial recording may still be on disk from earlier flushes." ), }); None } }, None => None, } } fn append_resampled_audio( capture_buffer: &mut Vec, wav_writer: &mut Option, resampled: &[f32], session_id: u64, status_channel: &Channel, ) { if resampled.is_empty() { return; } capture_buffer.extend_from_slice(resampled); if let Some(writer) = wav_writer.as_mut() { 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; } } } fn maybe_dispatch_chunk( engine: &Arc, config: &StartLiveTranscriptionConfig, capture_buffer: &mut Vec, buffer_start_sample: &mut u64, chunk_id: &mut u32, stopping: bool, status_channel: &Channel, session_id: u64, ) -> Option { let target_len = if capture_buffer.len() >= CHUNK_SAMPLES { CHUNK_SAMPLES } else if stopping && capture_buffer.len() >= FINAL_CHUNK_MIN_SAMPLES { capture_buffer.len() } else { return None; }; let trim_before_secs = if *chunk_id > 0 && !stopping && target_len > OVERLAP_SAMPLES { OVERLAP_SAMPLES as f64 / WHISPER_SAMPLE_RATE as f64 } else { 0.0 }; let speech_window = if trim_before_secs > 0.0 { &capture_buffer[OVERLAP_SAMPLES..target_len] } else { &capture_buffer[..target_len] }; let speech_gate = evaluate_speech_gate(speech_window); if speech_gate.skip { let skipped_ms = (target_len as u64 * 1000) / WHISPER_SAMPLE_RATE as u64; let gate_reason = match speech_gate.reason { "silence" => "near-silence", "insufficient_speech" => "insufficient speech energy", other => other, }; tracing::debug!( session_id, skipped_ms, gate_reason, peak_rms = speech_gate.peak_rms, peak_amplitude = speech_gate.peak_amplitude, speech_window_count = speech_gate.speech_window_count, window_count = speech_gate.window_count, max_consecutive_speech_windows = speech_gate.max_consecutive_speech_windows, "skipped chunk on speech gate" ); let _ = status_channel.send(LiveStatusMessage::Warning { session_id, message: match speech_gate.reason { "silence" => format!( "Skipped {skipped_ms}ms of near-silent audio. If this keeps happening, try a louder mic level or move closer to the microphone." ), _ => format!( "Skipped {skipped_ms}ms of low-confidence audio. If this keeps happening, try a louder mic level or reduce background noise." ), }, }); capture_buffer.drain(..target_len); *buffer_start_sample = buffer_start_sample.saturating_add(target_len as u64); return None; } *chunk_id = chunk_id.saturating_add(1); let current_chunk_id = *chunk_id; let chunk_start_sample = *buffer_start_sample; let duration_secs = target_len as f64 / WHISPER_SAMPLE_RATE as f64; let chunk_samples = capture_buffer[..target_len].to_vec(); tracing::debug!( session_id, chunk_id = current_chunk_id, duration_secs, samples = chunk_samples.len(), "dispatching chunk" ); let advance_by = if stopping || target_len < CHUNK_SAMPLES { target_len } else { target_len.saturating_sub(OVERLAP_SAMPLES) }; capture_buffer.drain(..advance_by); *buffer_start_sample = buffer_start_sample.saturating_add(advance_by as u64); let options = TranscriptionOptions { language: config.language.clone(), initial_prompt: config.initial_prompt.clone(), }; let engine = engine.clone(); let (tx, rx) = std::sync::mpsc::channel(); let abort_flag = Arc::new(AtomicBool::new(false)); let worker_abort = abort_flag.clone(); let dispatched_at = Instant::now(); // The handle is intentionally dropped — the cancellation route is // now `abort_flag` plumbed through whisper-rs's abort callback, not // a JoinHandle. On any early exit from the loop the per-task flag // is set and whisper-rs returns cleanly on its next poll. // // instrument: capture the parent `run_live_session` span (so // `session_id` propagates) and open an `inference` child span // carrying `chunk_id` on the new OS thread. `#[instrument]` can't // be applied to a closure, so we attach the span manually after // crossing the thread boundary. Without this, every tracing event // emitted by `LocalEngine::transcribe_sync_with_abort` lands in // an unparented span and the operator cannot correlate a runaway // chunk back to its session. let parent_span = tracing::Span::current(); thread::spawn(move || { let _parent = parent_span.enter(); let inference_span = tracing::info_span!( "inference", chunk_id = current_chunk_id, duration_secs, ); let _enter = inference_span.enter(); let audio = AudioSamples::mono_16khz(chunk_samples); let started = Instant::now(); let result = engine .transcribe_sync_with_abort(&audio, &options, worker_abort) .map(|mut timed| { timed.inference_ms = started.elapsed().as_millis() as u64; timed }) .map_err(|e| e.to_string()); let _ = tx.send(result); }); Some(InferenceTask { chunk_id: current_chunk_id, chunk_start_sample, trim_before_secs, duration_secs, rx, abort_flag, dispatched_at, }) } #[allow(clippy::too_many_arguments)] fn poll_inference( inflight: &mut Option, result_listener_lost: &mut bool, session_id: u64, config: &StartLiveTranscriptionConfig, recent_segments: &mut Vec, dictionary_terms: &[String], result_channel: &Channel, status_channel: &Channel, stop_flag: &Arc, ) -> Result, String> { let Some(task) = inflight else { return Ok(None); }; match task.rx.try_recv() { Ok(Ok(timed)) => { let mut segments: Vec = timed.transcript.segments().to_vec(); trim_overlap_segments(&mut segments, task.trim_before_secs); // Capture raw text BEFORE any post-processing so the preview // overlay can show what Whisper actually returned. let raw_text = segments .iter() .map(|segment| segment.text.trim()) .filter(|segment| !segment.is_empty()) .collect::>() .join(" "); post_process_segments( &mut segments, &PostProcessOptions { remove_fillers: config.remove_fillers, british_english: config.british_english, anti_hallucination: config.anti_hallucination, format_mode: FormatMode::parse(&config.format_mode), dictionary_terms: dictionary_terms.to_vec(), }, None, ); let chunk_start_secs = task.chunk_start_sample as f64 / WHISPER_SAMPLE_RATE as f64; let skipped_duplicates = filter_duplicate_boundary_segments( &mut segments, chunk_start_secs, recent_segments, ); let segment_count = segments.len(); let delivered_segments = segments.clone(); let result_message = LiveResultMessage { session_id, chunk_id: task.chunk_id, chunk_start_secs, duration: task.duration_secs, language: timed.transcript.language().to_string(), inference_ms: timed.inference_ms, segments, raw_text, }; let delivered = emit_live_result( result_channel, status_channel, result_listener_lost, stop_flag, &result_message, ); remember_recent_segments(recent_segments, &delivered_segments, chunk_start_secs); tracing::info!( session_id, chunk_id = task.chunk_id, segments = segment_count, inference_ms = timed.inference_ms, delivered, skipped_duplicates, "processed live chunk" ); *inflight = None; Ok(Some(true)) } Ok(Err(err)) => { tracing::error!(session_id, error = %err, "inference error"); *inflight = None; let _ = status_channel.send(LiveStatusMessage::Error { session_id, message: err.clone(), }); Err(err) } Err(std::sync::mpsc::TryRecvError::Empty) => Ok(Some(false)), Err(std::sync::mpsc::TryRecvError::Disconnected) => { *inflight = None; let message = "Inference worker disconnected unexpectedly".to_string(); tracing::error!(session_id, "{message}"); let _ = status_channel.send(LiveStatusMessage::Error { session_id, message: message.clone(), }); Err(message) } } } fn emit_live_result( result_channel: &Channel, status_channel: &Channel, result_listener_lost: &mut bool, stop_flag: &Arc, result_message: &LiveResultMessage, ) -> bool { if *result_listener_lost { return false; } match result_channel.send(result_message.clone()) { Ok(()) => true, Err(err) => { *result_listener_lost = true; tracing::warn!( session_id = result_message.session_id, chunk_id = result_message.chunk_id, error = %err, "result listener unavailable; continuing without live updates" ); // If the warning send also fails, the entire frontend channel // pair is dead — almost certainly the user closed the app // window without calling stop_live_transcription_session. // Self-assert stop_flag so the inference worker drains and // exits cleanly instead of polling every 10 ms forever, which // otherwise would burn CPU + GPU memory and keep the WAV // writer file handle open until the process dies. let warn_send = status_channel.send(LiveStatusMessage::Warning { session_id: result_message.session_id, message: "Live preview disconnected; transcription will continue in the background until you stop the session.".into(), }); if warn_send.is_err() { tracing::error!( session_id = result_message.session_id, "status channel also unavailable; self-asserting stop_flag so the worker exits" ); stop_flag.store(true, Ordering::Relaxed); } false } } } fn trim_overlap_segments(segments: &mut Vec, trim_before_secs: f64) { if trim_before_secs <= 0.0 { return; } segments.retain(|segment| segment.end > trim_before_secs); for segment in segments.iter_mut() { if segment.start < trim_before_secs { segment.start = trim_before_secs; } } } fn filter_duplicate_boundary_segments( segments: &mut Vec, chunk_start_secs: f64, recent_segments: &[RecentTranscriptSegment], ) -> usize { if recent_segments.is_empty() { return 0; } let mut skipped = 0usize; segments.retain(|segment| { if segment.start > DUPLICATE_CHECK_LEADING_SECS { return true; } let absolute_start = chunk_start_secs + segment.start; let candidates = build_nearby_transcript_candidates(recent_segments, absolute_start); if candidates.is_empty() { return true; } let duplicate = candidates.iter().any(|candidate| { transcripts_overlap(&segment.text, candidate) || transcripts_loosely_overlap(&segment.text, candidate) }); if duplicate { skipped += 1; return false; } true }); skipped } fn remember_recent_segments( recent_segments: &mut Vec, segments: &[Segment], chunk_start_secs: f64, ) { if segments.is_empty() { return; } for segment in segments { let text = segment.text.trim(); if text.is_empty() { continue; } recent_segments.push(RecentTranscriptSegment { start_secs: chunk_start_secs + segment.start, end_secs: chunk_start_secs + segment.end, text: text.to_string(), }); } let cutoff = recent_segments .last() .map(|segment| segment.end_secs - DUPLICATE_HISTORY_RETENTION_SECS) .unwrap_or(0.0); recent_segments.retain(|segment| segment.end_secs >= cutoff); } fn build_nearby_transcript_candidates( recent_segments: &[RecentTranscriptSegment], timestamp_secs: f64, ) -> Vec { let mut nearby: Vec<&RecentTranscriptSegment> = recent_segments .iter() .filter(|segment| { !segment.text.trim().is_empty() && (segment.end_secs - timestamp_secs).abs() <= DUPLICATE_TRANSCRIPT_WINDOW_SECS }) .collect(); nearby.sort_by(|left, right| { left.start_secs .partial_cmp(&right.start_secs) .unwrap_or(std::cmp::Ordering::Equal) }); let mut texts: Vec = Vec::new(); for start in 0..nearby.len() { let mut merged = String::new(); let upper = nearby.len().min(start + DUPLICATE_TRANSCRIPT_MERGE_LIMIT); for segment in &nearby[start..upper] { if !merged.is_empty() { merged.push(' '); } merged.push_str(segment.text.trim()); if !texts.iter().any(|existing| existing == &merged) { texts.push(merged.clone()); } } } texts } fn normalize_transcript_text(text: &str) -> String { let mut normalized = String::with_capacity(text.len()); for ch in text.chars() { if ch.is_alphanumeric() { for lower in ch.to_lowercase() { normalized.push(lower); } } else { normalized.push(' '); } } normalized.split_whitespace().collect::>().join(" ") } fn count_common_tokens<'a>(a: &[&'a str], b: &[&'a str]) -> usize { let mut counts: HashMap<&'a str, usize> = HashMap::new(); for token in a { *counts.entry(*token).or_insert(0) += 1; } let mut common = 0usize; for token in b { if let Some(remaining) = counts.get_mut(*token) { if *remaining > 0 { *remaining -= 1; common += 1; } } } common } fn longest_common_token_subsequence(a: &[&str], b: &[&str]) -> usize { let mut prev = vec![0usize; b.len() + 1]; let mut curr = vec![0usize; b.len() + 1]; for token_a in a { for (j, token_b) in b.iter().enumerate() { curr[j + 1] = if token_a == token_b { prev[j] + 1 } else { prev[j + 1].max(curr[j]) }; } prev.clone_from(&curr); curr.fill(0); } prev[b.len()] } fn is_low_signal_token(token: &str) -> bool { LOW_SIGNAL_TOKENS.contains(&token) } fn meaningful_tokens(text: &str) -> Vec<&str> { text.split_whitespace() .filter(|token| !token.is_empty() && token.len() > 1 && !is_low_signal_token(token)) .collect() } fn transcripts_overlap(a: &str, b: &str) -> bool { let normalized_a = normalize_transcript_text(a); let normalized_b = normalize_transcript_text(b); if normalized_a.is_empty() || normalized_b.is_empty() { return false; } if normalized_a == normalized_b || normalized_a.contains(&normalized_b) || normalized_b.contains(&normalized_a) { return true; } let tokens_a: Vec<&str> = normalized_a.split_whitespace().collect(); let tokens_b: Vec<&str> = normalized_b.split_whitespace().collect(); let shorter = tokens_a.len().min(tokens_b.len()); if shorter < MIN_TOKENS_FOR_OVERLAP { return false; } let common = count_common_tokens(&tokens_a, &tokens_b); if common as f64 / shorter as f64 >= TOKEN_COVERAGE_THRESHOLD { return true; } let sequence = longest_common_token_subsequence(&tokens_a, &tokens_b); sequence as f64 / shorter as f64 >= TOKEN_SEQUENCE_THRESHOLD } fn transcripts_loosely_overlap(a: &str, b: &str) -> bool { let normalized_a = normalize_transcript_text(a); let normalized_b = normalize_transcript_text(b); if normalized_a.is_empty() || normalized_b.is_empty() { return false; } if normalized_a == normalized_b || normalized_a.contains(&normalized_b) || normalized_b.contains(&normalized_a) { return true; } let tokens_a = meaningful_tokens(&normalized_a); let tokens_b = meaningful_tokens(&normalized_b); let shorter = tokens_a.len().min(tokens_b.len()); if shorter < MIN_MEANINGFUL_TOKENS_FOR_OVERLAP { return false; } let common = count_common_tokens(&tokens_a, &tokens_b); if common as f64 / shorter as f64 >= MEANINGFUL_TOKEN_COVERAGE_THRESHOLD { return true; } let sequence = longest_common_token_subsequence(&tokens_a, &tokens_b); sequence as f64 / shorter as f64 >= MEANINGFUL_TOKEN_SEQUENCE_THRESHOLD } fn record_speech_window(state: &mut SpeechGateState, rms: f32, peak: f32) { state.window_count += 1; state.peak_rms = state.peak_rms.max(rms); state.peak_amplitude = state.peak_amplitude.max(peak); let is_speech_window = rms >= SPEECH_WINDOW_RMS_THRESHOLD && peak >= SPEECH_WINDOW_PEAK_THRESHOLD; if !is_speech_window { state.consecutive_speech_windows = 0; return; } state.speech_window_count += 1; state.consecutive_speech_windows += 1; state.max_consecutive_speech_windows = state .max_consecutive_speech_windows .max(state.consecutive_speech_windows); } fn speech_gate_decision(state: SpeechGateState, chunk_peak: f32) -> SpeechGateDecision { if state.window_count == 0 { return SpeechGateDecision { skip: false, reason: "unavailable", peak_rms: state.peak_rms, peak_amplitude: state.peak_amplitude, window_count: state.window_count, speech_window_count: state.speech_window_count, max_consecutive_speech_windows: state.max_consecutive_speech_windows, }; } let reason = if chunk_peak < FLATLINE_PEAK_THRESHOLD || state.peak_rms < SILENCE_RMS_THRESHOLD { Some("silence") } else if state.speech_window_count < MIN_SPEECH_FRAMES && state.peak_rms < STRONG_SPEECH_RMS_THRESHOLD && state.peak_amplitude < STRONG_SPEECH_PEAK_THRESHOLD { Some("insufficient_speech") } else { None }; SpeechGateDecision { skip: reason.is_some(), reason: reason.unwrap_or("speech_detected"), peak_rms: state.peak_rms, peak_amplitude: state.peak_amplitude, window_count: state.window_count, speech_window_count: state.speech_window_count, max_consecutive_speech_windows: state.max_consecutive_speech_windows, } } fn evaluate_speech_gate(samples: &[f32]) -> SpeechGateDecision { if samples.is_empty() { return SpeechGateDecision { skip: true, reason: "silence", peak_rms: 0.0, peak_amplitude: 0.0, window_count: 0, speech_window_count: 0, max_consecutive_speech_windows: 0, }; } let chunk_peak = samples .iter() .map(|sample| sample.abs()) .fold(0.0_f32, f32::max); let mut state = SpeechGateState::default(); for frame in samples.chunks(SPEECH_FRAME_SAMPLES) { let len = frame.len().max(1) as f32; let rms = (frame.iter().map(|sample| sample * sample).sum::() / len).sqrt(); let peak = frame .iter() .map(|sample| sample.abs()) .fold(0.0_f32, f32::max); record_speech_window(&mut state, rms, peak); } speech_gate_decision(state, chunk_peak) } fn downmix_chunk(samples: Vec, channels: usize) -> Vec { if channels <= 1 { return samples; } samples .chunks(channels) .map(|frame| frame.iter().sum::() / channels as f32) .collect() } #[cfg(test)] mod tests { use super::*; use tauri::ipc::InvokeResponseBody; fn noop_status_channel() -> Channel { Channel::new(|_| Ok(())) } fn collecting_status_channel(payloads: Arc>>) -> Channel { Channel::new(move |body| { if let InvokeResponseBody::Json(json) = body { payloads.lock().unwrap().push(json); } Ok(()) }) } fn dummy_running_session( id: u64, release_join: Option>, ) -> RunningLiveSession { let stop_flag = Arc::new(AtomicBool::new(false)); let handle = tokio::spawn(async move { if let Some(notify) = release_join { notify.notified().await; } Ok(LiveSessionSummary { session_id: id, dropped_audio_ms: 0, audio_path: None, }) }); RunningLiveSession { id, stop_flag, handle, status_channel: noop_status_channel(), } } async fn test_begin_session_start( live_state: Arc, session_id: u64, release_setup: Option>, ) -> Result { let _lifecycle = live_state.lifecycle.lock().await; { let running = live_state.running.lock().unwrap(); if running.is_some() { return Err("A live transcription session is already running".into()); } } if let Some(notify) = release_setup { notify.notified().await; } *live_state.running.lock().unwrap() = Some(dummy_running_session(session_id, None)); Ok(session_id) } async fn test_stop_session( live_state: Arc, session_id: u64, ) -> Result { let _lifecycle = live_state.lifecycle.lock().await; let running = live_state.running.lock().unwrap().take(); let Some(running) = running else { return Err("No live transcription session is running".into()); }; if running.id != session_id { *live_state.running.lock().unwrap() = Some(running); return Err(format!("Session {session_id} is not active")); } running.stop_flag.store(true, Ordering::Relaxed); running .handle .await .map_err(|e| format!("Live session task failed: {e}"))? } fn segment(start: f64, end: f64, text: &str) -> Segment { Segment { start, end, text: text.to_string(), } } #[test] fn transcripts_overlap_detects_boundary_repeat() { assert!(transcripts_overlap( "I need to go to the shops tomorrow", "to go to the shops tomorrow" )); } #[test] fn loose_overlap_ignores_low_signal_only_match() { assert!(!transcripts_loosely_overlap( "I think we should do that soon", "we should maybe do it soon" )); } #[test] fn duplicate_boundary_filter_skips_repeated_opening_segment() { let recent_segments = vec![RecentTranscriptSegment { start_secs: 10.0, end_secs: 12.0, text: "I need to go to the shops tomorrow".to_string(), }]; let mut segments = vec![ segment(0.2, 1.0, "Need to go to the shops tomorrow"), segment(1.8, 2.4, "While I am there I need some cheese"), ]; let skipped = filter_duplicate_boundary_segments(&mut segments, 11.8, &recent_segments); assert_eq!(skipped, 1); assert_eq!(segments.len(), 1); assert_eq!(segments[0].text, "While I am there I need some cheese"); } #[test] fn remember_recent_segments_prunes_old_history() { let mut recent_segments = vec![RecentTranscriptSegment { start_secs: 0.0, end_secs: 1.0, text: "old text".to_string(), }]; remember_recent_segments( &mut recent_segments, &[segment(0.0, 0.8, "new text")], DUPLICATE_HISTORY_RETENTION_SECS + 1.0, ); assert_eq!(recent_segments.len(), 1); assert_eq!(recent_segments[0].text, "new text"); } #[test] fn speech_gate_treats_near_silence_as_skippable() { let samples = vec![0.0004_f32, 0.0002, 0.0003, 0.0001] .into_iter() .cycle() .take(SPEECH_FRAME_SAMPLES * 3) .collect::>(); let decision = evaluate_speech_gate(&samples); assert!(decision.skip); assert_eq!(decision.reason, "silence"); } #[test] fn speech_gate_rejects_isolated_noise_without_speech_windows() { let mut samples = Vec::new(); for i in 0..(SPEECH_FRAME_SAMPLES * 3) { let sample = if i % SPEECH_FRAME_SAMPLES == 0 { 0.010 } else { 0.0011 }; samples.push(sample); } let decision = evaluate_speech_gate(&samples); assert!(decision.skip); assert_eq!(decision.reason, "insufficient_speech"); assert_eq!(decision.speech_window_count, 0); } #[test] fn speech_gate_allows_sustained_speech_like_audio() { let samples = vec![0.014_f32; SPEECH_FRAME_SAMPLES * 3]; let decision = evaluate_speech_gate(&samples); assert!(!decision.skip); assert_eq!(decision.reason, "speech_detected"); assert_eq!(decision.speech_window_count, 3); assert_eq!(decision.max_consecutive_speech_windows, 3); } #[test] fn result_listener_loss_is_warned_once_and_not_treated_as_inference_failure() { let statuses = Arc::new(Mutex::new(Vec::new())); let status_channel = collecting_status_channel(statuses.clone()); let result_channel = Channel::new(|_| Err(tauri::Error::FailedToReceiveMessage)); let config = StartLiveTranscriptionConfig { engine: "whisper".into(), model_id: None, language: Some("en".into()), initial_prompt: None, save_audio: false, output_folder: None, remove_fillers: false, british_english: false, anti_hallucination: false, format_mode: "Raw".into(), microphone_device: None, profile_id: None, }; let mut recent_segments = Vec::new(); let mut result_listener_lost = false; let stop_flag = Arc::new(AtomicBool::new(false)); let (tx1, rx1) = std::sync::mpsc::channel(); tx1.send(Ok(lumotia_transcription::TimedTranscript { transcript: lumotia_core::types::Transcript::new( vec![segment(0.0, 0.8, "first chunk")], "en".into(), 0.8, ), inference_ms: 12, })) .unwrap(); let mut inflight = Some(InferenceTask { chunk_id: 1, chunk_start_sample: 0, trim_before_secs: 0.0, duration_secs: 0.8, rx: rx1, abort_flag: Arc::new(AtomicBool::new(false)), dispatched_at: Instant::now(), }); let first = poll_inference( &mut inflight, &mut result_listener_lost, 77, &config, &mut recent_segments, &[], &result_channel, &status_channel, &stop_flag, ) .unwrap(); assert_eq!(first, Some(true)); assert!(result_listener_lost); assert!(inflight.is_none()); assert_eq!(recent_segments.len(), 1); let warning_count_after_first = statuses.lock().unwrap().len(); assert_eq!(warning_count_after_first, 1); assert!( statuses.lock().unwrap()[0].contains("Live preview disconnected"), "expected a warning about background continuation after listener loss" ); let (tx2, rx2) = std::sync::mpsc::channel(); tx2.send(Ok(lumotia_transcription::TimedTranscript { transcript: lumotia_core::types::Transcript::new( vec![segment(0.0, 0.9, "second chunk")], "en".into(), 0.9, ), inference_ms: 14, })) .unwrap(); inflight = Some(InferenceTask { chunk_id: 2, chunk_start_sample: 16_000, trim_before_secs: 0.0, duration_secs: 0.9, rx: rx2, abort_flag: Arc::new(AtomicBool::new(false)), dispatched_at: Instant::now(), }); let second = poll_inference( &mut inflight, &mut result_listener_lost, 77, &config, &mut recent_segments, &[], &result_channel, &status_channel, &stop_flag, ) .unwrap(); assert_eq!(second, Some(true)); assert!(inflight.is_none()); assert_eq!(recent_segments.len(), 2); // Status channel still alive in this scenario, so stop_flag must // NOT have been auto-asserted — the worker keeps running so the // user can still call stop_live_transcription_session and receive // the Finished status. assert!( !stop_flag.load(Ordering::Relaxed), "stop_flag should stay false when the status channel is alive" ); assert_eq!( statuses.lock().unwrap().len(), warning_count_after_first, "listener-loss warning should only be emitted once" ); } #[test] fn dead_result_and_status_channels_self_assert_stop_flag() { // Both channels error on every send: the frontend has gone away // entirely (e.g., the user closed the main window without a // graceful stop). The worker must self-stop via stop_flag so it // doesn't burn CPU + GPU for an indefinitely-long session that // nobody is observing. let result_channel: Channel = Channel::new(|_| Err(tauri::Error::FailedToReceiveMessage)); let status_channel: Channel = Channel::new(|_| Err(tauri::Error::FailedToReceiveMessage)); let stop_flag = Arc::new(AtomicBool::new(false)); let mut result_listener_lost = false; let message = LiveResultMessage { session_id: 99, chunk_id: 1, chunk_start_secs: 0.0, duration: 0.5, language: "en".into(), inference_ms: 10, segments: vec![], raw_text: String::new(), }; let delivered = emit_live_result( &result_channel, &status_channel, &mut result_listener_lost, &stop_flag, &message, ); assert!( !delivered, "send must report not delivered when result_channel errors" ); assert!(result_listener_lost, "result_listener_lost must be set"); assert!( stop_flag.load(Ordering::Relaxed), "stop_flag must self-assert when both channels are dead so the worker exits" ); } #[tokio::test] async fn concurrent_starts_allow_only_one_session_to_claim_the_slot() { let live_state = Arc::new(LiveTranscriptionState::default()); let release_setup = Arc::new(tokio::sync::Notify::new()); let first = tokio::spawn(test_begin_session_start( live_state.clone(), 1, Some(release_setup.clone()), )); tokio::time::sleep(Duration::from_millis(20)).await; let second = tokio::spawn(test_begin_session_start(live_state.clone(), 2, None)); tokio::time::sleep(Duration::from_millis(20)).await; assert!( !second.is_finished(), "second start should wait on the lifecycle lock" ); release_setup.notify_one(); assert_eq!(first.await.unwrap().unwrap(), 1); let err = second.await.unwrap().unwrap_err(); assert_eq!(err, "A live transcription session is already running"); } #[tokio::test] async fn start_waits_for_stop_to_finish_joining_before_reusing_slot() { let live_state = Arc::new(LiveTranscriptionState::default()); let release_join = Arc::new(tokio::sync::Notify::new()); *live_state.running.lock().unwrap() = Some(dummy_running_session(7, Some(release_join.clone()))); let stop = tokio::spawn(test_stop_session(live_state.clone(), 7)); tokio::time::sleep(Duration::from_millis(20)).await; let start = tokio::spawn(test_begin_session_start(live_state.clone(), 8, None)); tokio::time::sleep(Duration::from_millis(20)).await; assert!( !start.is_finished(), "new start should block until stop finishes joining the old worker" ); release_join.notify_one(); let summary = stop.await.unwrap().unwrap(); assert_eq!(summary.session_id, 7); assert_eq!(start.await.unwrap().unwrap(), 8); } #[test] fn drain_timeout_scales_with_inflight_chunk_duration_secs() { // Time-bomb-1 regression: the drain timeout used to be a // constant derived from CHUNK_SAMPLES (capped at ~6s). On slow // CPU + large-v3 model (3-5x realtime), a 4-second chunk could // legitimately need ~20s to finish; the old cap aborted // healthy work mid-flight. The new derivation pulls duration // from the in-flight task itself. let task = InferenceTask { chunk_id: 1, chunk_start_sample: 0, trim_before_secs: 0.0, duration_secs: 4.0, rx: std::sync::mpsc::channel().1, abort_flag: Arc::new(AtomicBool::new(false)), dispatched_at: Instant::now(), }; let timeout = drain_timeout_for_inflight(Some(&task)); // 4.0s * 5x safety = 20s. Old constant-derived budget would // have been ~6s, aborting at chunk_duration * 1.5 — well short // of the 20s a 5x-realtime backend genuinely needs. assert_eq!( timeout, Duration::from_secs(20), "5x realtime headroom must be the actual budget" ); } #[test] fn drain_timeout_honours_floor_for_short_chunks() { // A sub-second tail chunk still needs more than its raw // duration to clear the decoder (model load amortisation, OS // scheduling, abort poll cadence). The floor guards against // the timeout firing on legitimate short-chunk inference. let task = InferenceTask { chunk_id: 1, chunk_start_sample: 0, trim_before_secs: 0.0, duration_secs: 0.3, // 0.3s * 5x = 1.5s, below the 2s floor rx: std::sync::mpsc::channel().1, abort_flag: Arc::new(AtomicBool::new(false)), dispatched_at: Instant::now(), }; let timeout = drain_timeout_for_inflight(Some(&task)); assert_eq!( timeout, DRAIN_TIMEOUT_FLOOR, "tiny chunks must still get the floor's worth of budget" ); } #[test] fn drain_timeout_uses_floor_when_no_inflight_task() { // Defensive: no-inflight is in practice short-circuited by the // outer `while self.state.inflight.is_some()` loop, but the // helper must still produce a well-defined non-zero value. assert_eq!(drain_timeout_for_inflight(None), DRAIN_TIMEOUT_FLOOR); } #[test] fn drain_timeout_rejects_non_finite_duration() { // Defensive against NaN / negative durations slipping in from // a malformed task. We must not silently produce a tiny or // overflowing budget — fall back to the floor. for bad_duration in [f64::NAN, f64::INFINITY, -1.0, 0.0] { let task = InferenceTask { chunk_id: 1, chunk_start_sample: 0, trim_before_secs: 0.0, duration_secs: bad_duration, rx: std::sync::mpsc::channel().1, abort_flag: Arc::new(AtomicBool::new(false)), dispatched_at: Instant::now(), }; assert_eq!( drain_timeout_for_inflight(Some(&task)), DRAIN_TIMEOUT_FLOOR, "non-finite duration {bad_duration:?} must fall back to the floor" ); } } #[test] fn dropping_inference_task_sets_abort_flag() { // Race-A regression: dropping an InferenceTask (via `?` // propagation, panic unwind, or early loop exit) MUST signal // the spawned worker thread to exit via the abort flag. Without // this, every early exit leaves an orphan thread running // whisper-rs against the engine's Mutex; reconnect/retry loops // pile up N orphans contending on the backend lock. let abort_flag = Arc::new(AtomicBool::new(false)); let (_tx, rx) = std::sync::mpsc::channel(); let task = InferenceTask { chunk_id: 99, chunk_start_sample: 0, trim_before_secs: 0.0, duration_secs: 2.0, rx, abort_flag: abort_flag.clone(), dispatched_at: Instant::now(), }; assert!( !abort_flag.load(Ordering::Relaxed), "abort flag must start cleared" ); drop(task); assert!( abort_flag.load(Ordering::Relaxed), "Drop for InferenceTask must signal cancellation to the worker" ); } /// Regression: every `tracing::*!` emit in this file must use the /// implicit module-path target so it is covered by the operator's /// documented triage filter /// `RUST_LOG=...,lumotia_lib::commands::live=debug`. /// /// A previous custom literal target slipped past EnvFilter (which /// matches on `::`-segments, not substrings) and silenced the /// drain-timeout warning at incident time. If you find yourself /// reaching for a custom literal target here, instead extend the /// `DEFAULT_*_FILTER` constants in `src-tauri/src/lib.rs`. #[test] fn no_lumotia_live_literal_target_in_live_rs() { let source = include_str!("live.rs"); // The forbidden literal, assembled at runtime so this test's own // source does not contain a stray match. Comment + doc-comment // lines are skipped so the rationale above does not self-trip. let forbidden = format!("\"{}\"", "lumotia_live"); let mut offenders = Vec::new(); for (idx, raw) in source.lines().enumerate() { let trimmed = raw.trim_start(); if trimmed.starts_with("//") { continue; } if raw.contains("target:") && raw.contains(&forbidden) { offenders.push(format!("L{}: {}", idx + 1, raw.trim())); } } assert!( offenders.is_empty(), "found custom literal target(s) that dodge the \ `lumotia_lib::commands::live` EnvFilter directive. Drop the \ `target:` parameter so the emit uses the module-path target. \ Offenders:\n{}", offenders.join("\n") ); } }