Land release blocker fixes and workspace cleanup

src-tauri/src/commands/live.rs

@@ -11,6 +11,7 @@ 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;
@@ -18,7 +19,9 @@ use crate::commands::models::{default_model_id_for_engine, ensure_model_loaded};
 use crate::commands::power::PowerAssertion;
 use crate::AppState;
 use kon_ai_formatting::{post_process_segments, FormatMode, PostProcessOptions};
-use kon_audio::{MicrophoneCapture, StreamingResampler, WavWriter};
+use kon_audio::{
+    AudioChunk, CaptureRuntimeError, MicrophoneCapture, StreamingResampler, WavWriter,
+};
 use kon_core::constants::WHISPER_SAMPLE_RATE;
 use kon_core::types::{AudioSamples, Segment, TranscriptionOptions};
 use kon_transcription::LocalEngine;
@@ -57,6 +60,7 @@ const LOW_SIGNAL_TOKENS: &[&str] = &[
 #[derive(Default)]
 pub struct LiveTranscriptionState {
     next_session_id: AtomicU64,
+    lifecycle: AsyncMutex<()>,
     running: Mutex<Option<RunningLiveSession>>,
 }
 
@@ -153,6 +157,292 @@ struct LiveSessionSummary {
     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,
+        )?;
+        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: "Kon 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,
@@ -198,6 +488,7 @@ pub async fn start_live_transcription_session(
     result_channel: Channel<LiveResultMessage>,
     status_channel: Channel<LiveStatusMessage>,
 ) -> Result<StartLiveTranscriptionResponse, String> {
+    let _lifecycle = live_state.lifecycle.lock().await;
     {
         let running = live_state.running.lock().unwrap();
         if running.is_some() {
@@ -227,11 +518,8 @@ pub async fn start_live_transcription_session(
     // `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,
-    );
+    config.initial_prompt =
+        build_initial_prompt(&request_prompt, &profile.initial_prompt, &profile_terms);
 
     let model_id = config
         .model_id
@@ -258,7 +546,10 @@ pub async fn start_live_transcription_session(
     // 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())?)
+        Some(resolve_recording_path(
+            &app,
+            config.output_folder.as_deref(),
+        )?)
     } else {
         None
     };
@@ -299,6 +590,7 @@ pub async fn stop_live_transcription_session(
     live_state: tauri::State<'_, LiveTranscriptionState>,
     session_id: u64,
 ) -> Result<StopLiveTranscriptionResponse, 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());
@@ -360,203 +652,54 @@ fn run_live_session(
     // 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("kon live dictation session");
+    LiveSessionRuntime::new(
+        session_id,
+        engine,
+        config,
+        audio_path,
+        dictionary_terms,
+        result_channel,
+        status_channel,
+        stop_flag,
+    )?
+    .run()
+}
 
-    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())?;
-    // Drain runtime stream errors into the status channel so the user
-    // gets a toast when the device disconnects mid-recording instead of
-    // silently producing empty transcripts. The `_capture` binding keeps
-    // the cpal stream alive for the duration of the session.
-    let mic_error_rx = capture.take_error_rx();
-    let _capture = capture;
-
-    let mut resampler: Option<StreamingResampler> = None;
-    let mut capture_buffer: Vec<f32> = Vec::new();
-
-    // Progressive WAV writer (brief item #19). Sample rate comes from
-    // the loaded backend's capabilities (#13 wiring) so a future
-    // non-16kHz backend records at its native rate without further
-    // plumbing. The writer flushes its header every ~500 ms, so the
-    // file on disk is a playable WAV even if the process is killed.
+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 mut wav_writer: Option<WavWriter> = match audio_path.as_ref() {
-        Some(path) => match WavWriter::create(path, sample_rate, 1) {
-            Ok(w) => Some(w),
-            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
-            }
-        },
-        None => None,
-    };
-    // `reported_audio_path` is decided at end-of-session based on
-    // whether the writer finalised successfully, not at open time.
-    // This way a writer that dies mid-session (append error clearing
-    // wav_writer, or finalise returning Err) does not leak a stale
-    // path back to the frontend that might point to a file whose
-    // header is out of sync with its data chunk.
-
-    let mut buffer_start_sample: u64 = 0;
-    let mut dropped_audio_ms: u64 = 0;
-    let mut chunk_id: u32 = 0;
-    let mut inflight: Option<InferenceTask> = None;
-    let mut resampler_flushed = false;
-    let mut recent_segments: Vec<RecentTranscriptSegment> = Vec::new();
-
-    loop {
-        if let Some(_done) = poll_inference(
-            &mut inflight,
-            session_id,
-            &config,
-            &mut recent_segments,
-            &dictionary_terms,
-            &result_channel,
-            &status_channel,
-        )? {}
-
-        // Surface any cpal runtime errors as warnings. Non-fatal: a hard
-        // disconnect will also drop the audio sender and be caught by
-        // the `Disconnected` arm below. This lets the user see a toast
-        // even when cpal recovers without tearing the stream down.
-        if let Some(err_rx) = &mic_error_rx {
-            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
-                    ),
-                });
-            }
-        }
-
-        match rx.recv_timeout(Duration::from_millis(25)) {
-            Ok(chunk) => {
-                let mono = downmix_chunk(chunk.samples, chunk.channels as usize);
-                let resampler = match &mut resampler {
-                    Some(resampler) => resampler,
-                    None => {
-                        resampler = Some(
-                            StreamingResampler::new(chunk.sample_rate)
-                                .map_err(|e| e.to_string())?,
-                        );
-                        resampler.as_mut().expect("resampler just set")
-                    }
-                };
-
-                let resampled = resampler.push_samples(&mono).map_err(|e| e.to_string())?;
-                append_resampled_audio(
-                    &mut capture_buffer,
-                    &mut wav_writer,
-                    &resampled,
-                    session_id,
-                    &status_channel,
-                );
-            }
-            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
-            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
-                let message = "Microphone capture disconnected unexpectedly".to_string();
-                let _ = status_channel.send(LiveStatusMessage::Error {
-                    session_id,
-                    message: message.clone(),
-                });
-                return Err(message);
-            }
-        }
-
-        if inflight.is_some() && capture_buffer.len() > MAX_PENDING_SAMPLES {
-            let overflow = capture_buffer.len() - MAX_PENDING_SAMPLES;
-            capture_buffer.drain(..overflow);
-            buffer_start_sample = buffer_start_sample.saturating_add(overflow as u64);
-            dropped_audio_ms = dropped_audio_ms
-                .saturating_add((overflow as u64 * 1000) / WHISPER_SAMPLE_RATE as u64);
-            let _ = status_channel.send(LiveStatusMessage::Overload {
+    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,
-                dropped_audio_ms,
-                message: "Kon dropped older audio to keep live dictation responsive".into(),
+                message: format!(
+                    "Failed to open audio recording file ({}); transcription will continue without saving audio.",
+                    e
+                ),
             });
-        }
-
-        let stopping = stop_flag.load(Ordering::Relaxed);
-        if stopping && !resampler_flushed {
-            if let Some(resampler) = &mut resampler {
-                let tail = resampler.flush().map_err(|e| e.to_string())?;
-                append_resampled_audio(
-                    &mut capture_buffer,
-                    &mut wav_writer,
-                    &tail,
-                    session_id,
-                    &status_channel,
-                );
-            }
-            resampler_flushed = true;
-            // Final flush for the WAV header so the last chunk's header
-            // update is on disk before we drop into the inference drain.
-            if let Some(writer) = wav_writer.as_mut() {
-                if let Err(e) = writer.flush() {
-                    let _ = status_channel.send(LiveStatusMessage::Warning {
-                        session_id,
-                        message: format!("WAV flush failed near session end: {e}"),
-                    });
-                }
-            }
-        }
-
-        if inflight.is_none() {
-            if let Some(task) = maybe_dispatch_chunk(
-                &engine,
-                &config,
-                &mut capture_buffer,
-                &mut buffer_start_sample,
-                &mut chunk_id,
-                stopping,
-                &status_channel,
-                session_id,
-            ) {
-                inflight = Some(task);
-                continue;
-            }
-
-            if stopping && resampler_flushed {
-                break;
-            }
+            None
         }
     }
+}
 
-    while inflight.is_some() {
-        poll_inference(
-            &mut inflight,
-            session_id,
-            &config,
-            &mut recent_segments,
-            &dictionary_terms,
-            &result_channel,
-            &status_channel,
-        )?;
-        thread::sleep(Duration::from_millis(10));
-    }
-
-    // Finalise the progressive WAV writer and decide whether to
-    // report a path to the frontend. Only a clean finalise produces a
-    // reported path: a writer that died mid-session (wav_writer was
-    // already None) or a finalise that itself errored both yield
-    // `None`, so `StopLiveTranscriptionResponse.audio_path` reflects
-    // "recording is known-good" rather than "recording was attempted".
-    let audio_path = match wav_writer.take() {
+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.as_ref().map(|p| p.to_string_lossy().to_string()),
+            Ok(()) => audio_path.map(|path| path.to_string_lossy().to_string()),
             Err(e) => {
                 let _ = status_channel.send(LiveStatusMessage::Warning {
                     session_id,
@@ -568,13 +711,7 @@ fn run_live_session(
             }
         },
         None => None,
-    };
-
-    Ok(LiveSessionSummary {
-        session_id,
-        dropped_audio_ms,
-        audio_path,
-    })
+    }
 }
 
 fn append_resampled_audio(
@@ -719,6 +856,7 @@ fn maybe_dispatch_chunk(
 
 fn poll_inference(
     inflight: &mut Option<InferenceTask>,
+    result_listener_lost: &mut bool,
     session_id: u64,
     config: &StartLiveTranscriptionConfig,
     recent_segments: &mut Vec<RecentTranscriptSegment>,
@@ -761,22 +899,30 @@ fn poll_inference(
             );
             let segment_count = segments.len();
             let delivered_segments = segments.clone();
-
-            result_channel
-                .send(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,
-                })
-                .map_err(|e| e.to_string())?;
+            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,
+                &result_message,
+            );
             remember_recent_segments(recent_segments, &delivered_segments, chunk_start_secs);
             eprintln!(
-                "[live] session {session_id}: delivered chunk {} with {} segments in {}ms{}",
+                "[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,
@@ -813,6 +959,33 @@ fn poll_inference(
     }
 }
 
+fn emit_live_result(
+    result_channel: &Channel<LiveResultMessage>,
+    status_channel: &Channel<LiveStatusMessage>,
+    result_listener_lost: &mut bool,
+    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
+            );
+            let _ = 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(),
+            });
+            false
+        }
+    }
+}
+
 fn trim_overlap_segments(segments: &mut Vec<Segment>, trim_before_secs: f64) {
     if trim_before_secs <= 0.0 {
         return;
@@ -1150,6 +1323,82 @@ fn downmix_chunk(samples: Vec<f32>, channels: usize) -> Vec<f32> {
 #[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 {
@@ -1256,4 +1505,157 @@ mod tests {
         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 (tx1, rx1) = std::sync::mpsc::channel();
+        tx1.send(Ok(kon_transcription::TimedTranscript {
+            transcript: kon_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,
+        )
+        .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(kon_transcription::TimedTranscript {
+            transcript: kon_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,
+        )
+        .unwrap();
+
+        assert_eq!(second, Some(true));
+        assert!(inflight.is_none());
+        assert_eq!(recent_segments.len(), 2);
+        assert_eq!(
+            statuses.lock().unwrap().len(),
+            warning_count_after_first,
+            "listener-loss warning should only be emitted once"
+        );
+    }
+
+    #[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);
+    }
 }