Lumotia/src-tauri/src/commands/live.rs

#![allow(clippy::too_many_arguments)]

use std::collections::HashMap;
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 magnotia_ai_formatting::{post_process_segments, FormatMode, PostProcessOptions};
use magnotia_audio::{
    AudioChunk, CaptureRuntimeError, MicrophoneCapture, StreamingResampler, WavWriter,
};
use magnotia_core::constants::WHISPER_SAMPLE_RATE;
use magnotia_core::types::{AudioSamples, Segment, TranscriptionOptions};
use magnotia_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;
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>>,
}

impl ActiveCapture {
    fn start(config: &StartLiveTranscriptionConfig) -> Result<Self, String> {
        let (mut capture, 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: capture,
            rx,
            mic_error_rx,
        })
    }

    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>,
}

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)?;
            }
            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> {
        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> {
        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(())
    }

    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: "Magnotia 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> {
        while self.state.inflight.is_some() {
            self.poll_inference()?;
            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<magnotia_transcription::TimedTranscript, String>>,
}

#[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)?;
    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(|| magnotia_storage::DEFAULT_PROFILE_ID.to_string());

    let profile = magnotia_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> =
        magnotia_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());
    eprintln!(
        "[live] starting session: engine={}, model={}, language={:?}, save_audio={}",
        config.engine, model_id, config.language, config.save_audio
    );
    // 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,
    });

    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)?;
    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);

    let summary = running
        .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 _ = running.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}")),
    }
}

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("magnotia 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,
        };
        eprintln!(
            "[live] session {session_id}: skipped {skipped_ms}ms chunk as {gate_reason} \
(peak_rms={:.6}, peak={:.6}, speech_windows={}/{}, max_consecutive={})",
            speech_gate.peak_rms,
            speech_gate.peak_amplitude,
            speech_gate.speech_window_count,
            speech_gate.window_count,
            speech_gate.max_consecutive_speech_windows,
        );
        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();
    eprintln!(
        "[live] session {session_id}: dispatching chunk {} ({duration_secs:.2}s, {} samples)",
        current_chunk_id,
        chunk_samples.len()
    );
    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();

    thread::spawn(move || {
        let audio = AudioSamples::mono_16khz(chunk_samples);
        let started = Instant::now();
        let result = engine
            .transcribe_sync(&audio, &options)
            .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,
    })
}

#[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);
            eprintln!(
                "[live] session {session_id}: {} chunk {} with {} segments in {}ms{}",
                if delivered {
                    "delivered"
                } else {
                    "processed without listener for"
                },
                task.chunk_id,
                segment_count,
                timed.inference_ms,
                if skipped_duplicates > 0 {
                    format!(" (skipped {skipped_duplicates} duplicate boundary segment(s))")
                } else {
                    String::new()
                }
            );

            *inflight = None;
            Ok(Some(true))
        }
        Ok(Err(err)) => {
            eprintln!("[live] session {session_id}: inference error: {err}");
            *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();
            eprintln!("[live] session {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;
            eprintln!(
                "[live] session {}: result listener unavailable on chunk {}: {}; continuing without live updates",
                result_message.session_id, result_message.chunk_id, err
            );
            // 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() {
                eprintln!(
                    "[live] session {}: status channel also unavailable; \
                     self-asserting stop_flag so the worker exits",
                    result_message.session_id
                );
                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(magnotia_transcription::TimedTranscript {
            transcript: magnotia_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,
        });

        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(magnotia_transcription::TimedTranscript {
            transcript: magnotia_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,
        });

        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);
    }
}