Jake
6c212a0d2c
Phase B.1 survey finding (commit 5725836 cancellable Whisper inference +
bounded drain + lock-over-await).
The upper comment on `start_live_transcription_session` claimed:
Released explicitly before the RunningLiveSession is installed in
`live_state.running` so the symmetric stop path doesn't observe a
half-initialised state.
The actual code (lines 801-811) does the opposite: it installs the
RunningLiveSession FIRST, then drops the lifecycle guard. That ordering
is the SAFER one — a concurrent stop_live acquiring lifecycle observes
a fully-installed `running` slot or none, never a half-initialised
state. The bug was in the comment, not the code.
Future-reader trap: an atomiser-grade review of the locking discipline
would have trusted the comment over the code and "fixed" the code to
match — reintroducing the half-initialised window between drop and
install. Rewrote the Phase 1 comment to describe the actual behaviour
(hold-through-install + Phase 2 drop) and explain why holding is
intentional. Phase 2 comment already accurate; left untouched.
Other B.1 findings:
* 8 existing unit tests cover the atomiser-targetable surface (Race-A
drop sets abort flag, drain_timeout helpers defend NaN/inf/negative/
zero/no-inflight cases, channel-loss observability, tracing target
convention).
* Race-B drain-timeout end-to-end test remains a known gap. The SAFETY
comment in drain_inference acknowledges it ("requires a wedged
whisper-rs, which is hard to fixture"). Closing it would require
refactoring LiveSessionRuntime for testability — invasive, no
identified residual bug in the audited 60-line drain_inference body.
* stop_live comments + code consistent. No fix needed.
Verification:
- cargo test -p lumotia --lib commands::live: 17/17 (no change to test
count — comment-only edit)
- cargo clippy -p lumotia --all-targets -- -D warnings: clean
- cargo fmt --check: clean
2168 lines
77 KiB
Rust
2168 lines
77 KiB
Rust
#![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<Option<RunningLiveSession>>,
|
|
}
|
|
|
|
struct RunningLiveSession {
|
|
id: u64,
|
|
stop_flag: Arc<AtomicBool>,
|
|
handle: tokio::task::JoinHandle<Result<LiveSessionSummary, String>>,
|
|
status_channel: Channel<LiveStatusMessage>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct StartLiveTranscriptionConfig {
|
|
pub engine: String,
|
|
pub model_id: Option<String>,
|
|
pub language: Option<String>,
|
|
pub initial_prompt: Option<String>,
|
|
pub save_audio: bool,
|
|
pub output_folder: Option<String>,
|
|
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<String>,
|
|
/// 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<String>,
|
|
}
|
|
|
|
#[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<String>,
|
|
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<Segment>,
|
|
/// 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<String>,
|
|
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<String>,
|
|
}
|
|
|
|
/// 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<AudioChunk>,
|
|
mic_error_rx: Option<std::sync::mpsc::Receiver<CaptureRuntimeError>>,
|
|
/// Pre-roll chunks collected during the 350ms device validation
|
|
/// window. Drained BEFORE reading from `rx` so the head of the
|
|
/// recording survives small-buffer hosts that would otherwise
|
|
/// overflow the 32-slot capture channel during validation.
|
|
replay_buffer: VecDeque<AudioChunk>,
|
|
}
|
|
|
|
impl ActiveCapture {
|
|
fn start(config: &StartLiveTranscriptionConfig) -> Result<Self, String> {
|
|
let (mut capture, 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<LiveStatusMessage>,
|
|
) {
|
|
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<StreamingResampler>,
|
|
capture_buffer: Vec<f32>,
|
|
wav_writer: Option<WavWriter>,
|
|
buffer_start_sample: u64,
|
|
dropped_audio_ms: u64,
|
|
chunk_id: u32,
|
|
inflight: Option<InferenceTask>,
|
|
resampler_flushed: bool,
|
|
result_listener_lost: bool,
|
|
recent_segments: Vec<RecentTranscriptSegment>,
|
|
/// Cumulative value of `MicrophoneCapture::dropped_chunks()` last
|
|
/// observed by `poll_capture_drops`. Subtracted from the live
|
|
/// reading to compute the per-tick delta added to
|
|
/// `dropped_audio_ms`. Counts callback-level losses that the
|
|
/// session's own overflow path can't see.
|
|
last_dropped_chunks: u64,
|
|
/// Most-recent chunk dimensions used to convert a `dropped_chunks`
|
|
/// delta into milliseconds. Populated on the first received chunk
|
|
/// and refreshed thereafter. Zeroed sentinel means "no chunk yet";
|
|
/// drops observed before any chunk arrives are deferred until we
|
|
/// know the device's native rate.
|
|
last_chunk_samples_per_chan: u32,
|
|
last_chunk_sample_rate: u32,
|
|
}
|
|
|
|
impl LiveLoopState {
|
|
fn new(wav_writer: Option<WavWriter>) -> Self {
|
|
Self {
|
|
wav_writer,
|
|
..Self::default()
|
|
}
|
|
}
|
|
}
|
|
|
|
struct LiveSessionRuntime {
|
|
session_id: u64,
|
|
engine: Arc<LocalEngine>,
|
|
config: StartLiveTranscriptionConfig,
|
|
audio_path: Option<PathBuf>,
|
|
dictionary_terms: Vec<String>,
|
|
result_channel: Channel<LiveResultMessage>,
|
|
status_channel: Channel<LiveStatusMessage>,
|
|
stop_flag: Arc<AtomicBool>,
|
|
capture: ActiveCapture,
|
|
state: LiveLoopState,
|
|
}
|
|
|
|
impl LiveSessionRuntime {
|
|
fn new(
|
|
session_id: u64,
|
|
engine: Arc<LocalEngine>,
|
|
config: StartLiveTranscriptionConfig,
|
|
audio_path: Option<PathBuf>,
|
|
dictionary_terms: Vec<String>,
|
|
result_channel: Channel<LiveResultMessage>,
|
|
status_channel: Channel<LiveStatusMessage>,
|
|
stop_flag: Arc<AtomicBool>,
|
|
) -> Result<Self, String> {
|
|
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<LiveSessionSummary, String> {
|
|
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<Option<AudioChunk>, String> {
|
|
// Drain the validation pre-roll first. This is the head-of-the
|
|
// -recording audio that was collected during the 350ms device
|
|
// validation window; replaying it from a consumer-side VecDeque
|
|
// bypasses the 32-slot channel cap that previously dropped half
|
|
// of it silently on small-buffer hosts.
|
|
if let Some(chunk) = self.capture.replay_buffer.pop_front() {
|
|
return Ok(Some(chunk));
|
|
}
|
|
match self.capture.rx.recv_timeout(Duration::from_millis(25)) {
|
|
Ok(chunk) => Ok(Some(chunk)),
|
|
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Ok(None),
|
|
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<LiveSessionSummary, String> {
|
|
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<Result<lumotia_transcription::TimedTranscript, String>>,
|
|
/// 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<AtomicBool>,
|
|
/// 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<LiveResultMessage>,
|
|
status_channel: Channel<LiveStatusMessage>,
|
|
) -> Result<StartLiveTranscriptionResponse, String> {
|
|
ensure_main_window(&window)?;
|
|
// Phase 1: acquire the lifecycle lock and hold it across the full
|
|
// startup sequence — model load, audio-path resolution, worker
|
|
// spawn, AND installation of the RunningLiveSession in
|
|
// `live_state.running`. The lock is released by the explicit drop
|
|
// in Phase 2 (see comment near the bottom of this function).
|
|
//
|
|
// Holding through the install is intentional: a concurrent
|
|
// `stop_live_transcription_session` acquiring lifecycle MUST
|
|
// observe either a fully-installed `running` slot or none at all,
|
|
// never a half-initialised state. The dangerous pattern of holding
|
|
// the lock across a `JoinHandle.await` is on the stop path, not
|
|
// here — start hands the handle off to RunningLiveSession and
|
|
// never awaits it from inside this function.
|
|
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<String> =
|
|
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<StopLiveTranscriptionResponse, String> {
|
|
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<Arc<LocalEngine>, 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<LocalEngine>,
|
|
config: StartLiveTranscriptionConfig,
|
|
audio_path: Option<PathBuf>,
|
|
dictionary_terms: Vec<String>,
|
|
result_channel: Channel<LiveResultMessage>,
|
|
status_channel: Channel<LiveStatusMessage>,
|
|
stop_flag: Arc<AtomicBool>,
|
|
) -> Result<LiveSessionSummary, String> {
|
|
// 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<LocalEngine>,
|
|
audio_path: Option<&PathBuf>,
|
|
session_id: u64,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
) -> Option<WavWriter> {
|
|
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<WavWriter>,
|
|
audio_path: Option<&PathBuf>,
|
|
session_id: u64,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
) -> Option<String> {
|
|
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<f32>,
|
|
wav_writer: &mut Option<WavWriter>,
|
|
resampled: &[f32],
|
|
session_id: u64,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
) {
|
|
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<LocalEngine>,
|
|
config: &StartLiveTranscriptionConfig,
|
|
capture_buffer: &mut Vec<f32>,
|
|
buffer_start_sample: &mut u64,
|
|
chunk_id: &mut u32,
|
|
stopping: bool,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
session_id: u64,
|
|
) -> Option<InferenceTask> {
|
|
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<InferenceTask>,
|
|
result_listener_lost: &mut bool,
|
|
session_id: u64,
|
|
config: &StartLiveTranscriptionConfig,
|
|
recent_segments: &mut Vec<RecentTranscriptSegment>,
|
|
dictionary_terms: &[String],
|
|
result_channel: &Channel<LiveResultMessage>,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
stop_flag: &Arc<AtomicBool>,
|
|
) -> Result<Option<bool>, String> {
|
|
let Some(task) = inflight else {
|
|
return Ok(None);
|
|
};
|
|
|
|
match task.rx.try_recv() {
|
|
Ok(Ok(timed)) => {
|
|
let mut segments: Vec<Segment> = 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::<Vec<_>>()
|
|
.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<LiveResultMessage>,
|
|
status_channel: &Channel<LiveStatusMessage>,
|
|
result_listener_lost: &mut bool,
|
|
stop_flag: &Arc<AtomicBool>,
|
|
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<Segment>, 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<Segment>,
|
|
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<RecentTranscriptSegment>,
|
|
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<String> {
|
|
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<String> = 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::<Vec<_>>().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::<f32>() / 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<f32>, channels: usize) -> Vec<f32> {
|
|
if channels <= 1 {
|
|
return samples;
|
|
}
|
|
|
|
samples
|
|
.chunks(channels)
|
|
.map(|frame| frame.iter().sum::<f32>() / channels as f32)
|
|
.collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use tauri::ipc::InvokeResponseBody;
|
|
|
|
fn noop_status_channel() -> Channel<LiveStatusMessage> {
|
|
Channel::new(|_| Ok(()))
|
|
}
|
|
|
|
fn collecting_status_channel(payloads: Arc<Mutex<Vec<String>>>) -> Channel<LiveStatusMessage> {
|
|
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<Arc<tokio::sync::Notify>>,
|
|
) -> 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<LiveTranscriptionState>,
|
|
session_id: u64,
|
|
release_setup: Option<Arc<tokio::sync::Notify>>,
|
|
) -> Result<u64, String> {
|
|
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<LiveTranscriptionState>,
|
|
session_id: u64,
|
|
) -> Result<LiveSessionSummary, String> {
|
|
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::<Vec<_>>();
|
|
|
|
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<LiveResultMessage> =
|
|
Channel::new(|_| Err(tauri::Error::FailedToReceiveMessage));
|
|
let status_channel: Channel<LiveStatusMessage> =
|
|
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")
|
|
);
|
|
}
|
|
}
|