diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 7f7a52114..1a2101a2d 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -30,8 +30,9 @@ pub struct AppState { /// Workspace-provided relay URL override. Set by `apply_workspace` on app /// init and takes priority over env vars and compile-time defaults. pub relay_url_override: Mutex>, - /// Set during setup when managed agents are eligible for launch restore; - /// consumed after workspace identity/relay install to avoid the fallback. + /// Set during backend setup when managed agents are eligible for launch + /// restore. `apply_workspace` consumes it after installing the workspace + /// relay and identity, so agents never start against the fallback relay. pub managed_agent_restore_pending: AtomicBool, /// Whether desktop may repair managed-agent kind:0 profiles from its local /// records. Disabled by the agent-managed profiles experiment so an agent's @@ -39,8 +40,8 @@ pub struct AppState { pub managed_agent_profile_reconcile_enabled: AtomicBool, /// Shared shutdown signal checked by launch-time agent restoration. pub shutdown_started: AtomicBool, - /// Serializes managed-runtime transitions that change the protected PID set: - /// spawn/register, adoption, stop, shutdown, and sweep snapshots. + /// Serializes every managed-runtime transition that changes the protected + /// PID set: spawn/register, adoption, stop, shutdown, and sweep snapshots. /// Never perform network I/O while holding this lock. pub managed_agent_runtime_transition: Mutex<()>, pub managed_agents_store_lock: Mutex<()>, @@ -50,6 +51,11 @@ pub struct AppState { pub tts_settings: Mutex, pub tts_settings_load_error: Mutex>, pub tts_settings_transition: tokio::sync::Mutex<()>, + /// Tauri app handle — stored after setup so huddle commands can emit + /// `huddle-state-changed` events without needing the handle threaded + /// through every call site. + /// + /// Set once during `setup()` in `lib.rs`; never cleared. pub app_handle: Mutex>, /// Selected audio output device name. `None` = system default. /// Used by `connect_audio_relay` and TTS pipeline when opening sinks. diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 478771712..8a90d386b 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -16,7 +16,7 @@ use crate::events; use super::models; use super::relay_api::{self, fetch_channel_members, parse_channel_uuid}; -use super::state::{HuddlePhase, VoiceInputMode}; +use super::state::{HuddlePhase, HuddleState, VoiceInputMode}; use super::stt; use super::tts; @@ -216,7 +216,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result Result Result Result { + let mut huddle = state.huddle()?; + huddle.tts_starting.store(false, Ordering::Release); + if !huddle.tts_enabled + || !matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active) + || huddle.tts_pipeline.is_some() + { + return Ok(false); + } + let voice = state + .tts_settings + .lock() + .map_err(|error| format!("text-to-speech settings lock poisoned: {error}")) + .map(|settings| { + super::tts_settings::pocket_voice_name(&settings.voice_preferences).to_string() + })?; + publish(&voice, &mut huddle); Ok(true) } @@ -364,3 +388,98 @@ pub(crate) fn spawn_transcription_task( } }); } + +#[cfg(test)] +mod tts_start_race_tests { + use std::sync::{atomic::Ordering, Arc, Barrier, Mutex}; + + use crate::app_state::build_app_state; + + use super::{finalize_tts_pipeline_start, HuddlePhase}; + + #[test] + fn construction_reconciles_a_voice_selected_while_starting() { + let state = Arc::new(build_app_state()); + { + let mut huddle = state.huddle().expect("huddle state"); + huddle.phase = HuddlePhase::Active; + huddle.tts_enabled = true; + huddle.tts_starting.store(true, Ordering::Release); + } + + let constructed = Arc::new(Barrier::new(2)); + let publish = Arc::new(Barrier::new(2)); + let selected_voice = Arc::new(Mutex::new(None)); + let worker_state = Arc::clone(&state); + let worker_constructed = Arc::clone(&constructed); + let worker_publish = Arc::clone(&publish); + let worker_voice = Arc::clone(&selected_voice); + let worker = std::thread::spawn(move || { + worker_constructed.wait(); + worker_publish.wait(); + finalize_tts_pipeline_start(&worker_state, |voice, _| { + *worker_voice.lock().expect("selected voice") = Some(voice.to_string()); + }) + }); + + constructed.wait(); + assert!(state + .huddle() + .expect("huddle state") + .tts_starting + .load(Ordering::Acquire)); + state + .tts_settings + .lock() + .expect("text-to-speech settings") + .voice_preferences = vec!["pocket:marius".to_string()]; + publish.wait(); + + assert!(worker.join().expect("starter thread").expect("finalize")); + assert_eq!( + *selected_voice.lock().expect("selected voice"), + Some("marius".to_string()) + ); + } + + #[test] + fn construction_is_discarded_when_disabled_while_starting() { + let state = Arc::new(build_app_state()); + { + let mut huddle = state.huddle().expect("huddle state"); + huddle.phase = HuddlePhase::Active; + huddle.tts_enabled = true; + huddle.tts_starting.store(true, Ordering::Release); + } + + let constructed = Arc::new(Barrier::new(2)); + let publish = Arc::new(Barrier::new(2)); + let did_publish = Arc::new(Mutex::new(false)); + let worker_state = Arc::clone(&state); + let worker_constructed = Arc::clone(&constructed); + let worker_publish = Arc::clone(&publish); + let worker_did_publish = Arc::clone(&did_publish); + let worker = std::thread::spawn(move || { + worker_constructed.wait(); + worker_publish.wait(); + finalize_tts_pipeline_start(&worker_state, |_, _| { + *worker_did_publish.lock().expect("publish flag") = true; + }) + }); + + constructed.wait(); + { + let mut huddle = state.huddle().expect("huddle state"); + huddle.tts_enabled = false; + } + publish.wait(); + + assert!(!worker.join().expect("starter thread").expect("finalize")); + assert!(!*did_publish.lock().expect("publish flag")); + assert!(!state + .huddle() + .expect("huddle state") + .tts_starting + .load(Ordering::Acquire)); + } +} diff --git a/desktop/src-tauri/src/huddle/tts.rs b/desktop/src-tauri/src/huddle/tts.rs index b1e4cdcf0..5f6c6e373 100644 --- a/desktop/src-tauri/src/huddle/tts.rs +++ b/desktop/src-tauri/src/huddle/tts.rs @@ -35,10 +35,11 @@ //! can gate microphone input while the agent is speaking. use std::{ + collections::VecDeque, num::NonZero, path::PathBuf, sync::{ - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{self, SyncSender}, Arc, Mutex, MutexGuard, PoisonError, }, @@ -46,9 +47,50 @@ use std::{ time::Duration, }; -use super::pocket::{load_text_to_speech, load_voice_style, SAMPLE_RATE, VOICE_FILE_EXT}; +use super::pocket::{ + load_text_to_speech, load_voice_style, VoiceStyle, SAMPLE_RATE, VOICE_FILE_EXT, +}; use super::preprocessing::{preprocess_for_tts, split_sentences}; +#[derive(Debug)] +struct PendingVoiceChange { + generation: u64, + acknowledged: tokio::sync::oneshot::Sender<()>, +} + +type VoiceChangeAck = Arc>>; +type WorkerVoiceState = (Arc>, Arc, VoiceChangeAck); +type WorkerCancelSignals = (Arc, Arc); +type CancelTextState<'a> = ( + &'a mpsc::Receiver, + &'a mut VecDeque, + &'a mut Option, +); +type CancelSignals<'a> = (&'a AtomicBool, &'a AtomicBool); + +#[derive(Debug)] +struct QueuedText { + generation: u64, + text: String, +} + +#[derive(Clone, Debug)] +pub(crate) struct TtsTextSender { + text_tx: SyncSender, + generation: u64, +} + +impl TtsTextSender { + pub(crate) fn send(&self, text: String) -> Result<(), String> { + self.text_tx + .send(QueuedText { + generation: self.generation, + text, + }) + .map_err(|error| error.to_string()) + } +} + // ── Constants ───────────────────────────────────────────────────────────────── /// Maximum number of queued text items. @@ -117,7 +159,7 @@ const INTER_SENTENCE_SILENCE: f32 = 0.1; #[derive(Debug)] pub struct TtsPipeline { /// Send preprocessed text into the pipeline. - text_tx: SyncSender, + text_tx: SyncSender, /// `true` while the agent is speaking. Shared with the STT pipeline for gating. #[allow(dead_code)] pub tts_active: Arc, @@ -127,9 +169,16 @@ pub struct TtsPipeline { /// Kept alive here so the Arc isn't dropped — the worker holds a clone. #[allow(dead_code)] cancel: Arc, + /// Internal cancellation used only for voice changes. Kept separate so a + /// concurrent human barge-in always clears every queued message. + voice_cancel: Arc, /// Selected manifest voice. The worker reloads only the lightweight style /// when this changes; the warmed Pocket engine and audio player stay alive. voice: Arc>, + /// Tags messages so a voice change drops only pre-change queue entries. + voice_generation: Arc, + /// Completed after the worker drains pre-change text and installs the new style. + voice_change_ack: VoiceChangeAck, /// Worker thread handle — taken on drop to join cleanly. thread: Option>, } @@ -146,15 +195,21 @@ impl TtsPipeline { voice: &str, output_device: Option, ) -> Result { - let (text_tx, text_rx) = mpsc::sync_channel::(TEXT_QUEUE_DEPTH); + let (text_tx, text_rx) = mpsc::sync_channel::(TEXT_QUEUE_DEPTH); let shutdown = Arc::new(AtomicBool::new(false)); // cancel is passed in from HuddleState.tts_cancel — shared with STT for barge-in. let shutdown_worker = Arc::clone(&shutdown); let cancel_worker = Arc::clone(&cancel); + let voice_cancel = Arc::new(AtomicBool::new(false)); + let worker_voice_cancel = Arc::clone(&voice_cancel); let tts_active_worker = Arc::clone(&tts_active); let voice = Arc::new(Mutex::new(voice.to_string())); let voice_worker = Arc::clone(&voice); + let voice_generation = Arc::new(AtomicU64::new(1)); + let worker_voice_generation = Arc::clone(&voice_generation); + let voice_change_ack = Arc::new(Mutex::new(None)); + let worker_voice_change_ack = Arc::clone(&voice_change_ack); let model_dir_worker = model_dir.clone(); let handle = thread::Builder::new() @@ -162,11 +217,15 @@ impl TtsPipeline { .spawn(move || { tts_worker( model_dir_worker, - voice_worker, + ( + voice_worker, + worker_voice_generation, + worker_voice_change_ack, + ), text_rx, tts_active_worker, shutdown_worker, - cancel_worker, + (cancel_worker, worker_voice_cancel), output_device, ) }) @@ -177,7 +236,10 @@ impl TtsPipeline { tts_active, shutdown, cancel, + voice_cancel, voice, + voice_generation, + voice_change_ack, thread: Some(handle), }) } @@ -187,17 +249,25 @@ impl TtsPipeline { /// Non-blocking. Returns `Err` if the queue is full (bounded at /// `TEXT_QUEUE_DEPTH`) — caller may log and discard. pub fn speak(&self, text: String) -> Result<(), String> { - self.text_tx.try_send(text).map_err(|e| { - eprintln!("buzz-desktop: TTS queue saturated, dropping message: {e}"); - format!("TTS queue full, dropping: {e}") - }) + self.text_tx + .try_send(QueuedText { + generation: self.voice_generation.load(Ordering::Acquire), + text, + }) + .map_err(|e| { + eprintln!("buzz-desktop: TTS queue saturated, dropping message: {e}"); + format!("TTS queue full, dropping: {e}") + }) } /// Clone the bounded queue sender so callers can apply backpressure without /// holding the huddle mutex. Disabling TTS drops the receiver and unblocks /// any waiting sender while the shared cancellation flag stops playback. - pub(crate) fn text_sender(&self) -> SyncSender { - self.text_tx.clone() + pub(crate) fn text_sender(&self) -> TtsTextSender { + TtsTextSender { + text_tx: self.text_tx.clone(), + generation: self.voice_generation.load(Ordering::Acquire), + } } /// Select a bundled Pocket voice for subsequent speech. @@ -205,9 +275,23 @@ impl TtsPipeline { /// Current playback and queued text are cancelled immediately so content /// cannot continue in the old voice. The worker keeps its warmed inference /// engine and reloads only the reference style before the next utterance. - pub fn select_voice(&self, voice: &str) { + pub fn select_voice(&self, voice: &str) -> Option> { + begin_voice_change( + &self.voice, + &self.voice_generation, + &self.voice_cancel, + &self.voice_change_ack, + voice, + ) + } + + /// Reconcile the voice of a pipeline that has not been published yet. + /// + /// No caller can enqueue text before publication, so raising the shared + /// cancellation flag here would create a race that could discard the first + /// message queued immediately after installation. + pub(crate) fn select_voice_before_publish(&self, voice: &str) { *self.voice.lock().unwrap_or_else(|error| error.into_inner()) = voice.to_string(); - self.cancel.store(true, Ordering::Release); } /// Signal the worker thread to stop. @@ -237,13 +321,15 @@ impl Drop for TtsPipeline { fn tts_worker( model_dir: PathBuf, - selected_voice: Arc>, - text_rx: mpsc::Receiver, + voice_state: WorkerVoiceState, + text_rx: mpsc::Receiver, tts_active: Arc, shutdown: Arc, - cancel: Arc, + cancel_signals: WorkerCancelSignals, output_device: Option, ) { + let (selected_voice, voice_generation, voice_change_ack) = voice_state; + let (cancel, voice_cancel) = cancel_signals; // ── 1. Initialise TTS engine ────────────────────────────────────────────── let model_dir_str = model_dir.to_string_lossy().to_string(); @@ -254,7 +340,12 @@ fn tts_worker( "buzz-desktop: TTS engine init failed (model_dir={}): {e}. TTS disabled.", model_dir.display() ); - drain_until_shutdown(text_rx, &shutdown); + drain_tts_until_shutdown( + text_rx, + &shutdown, + (&cancel, &voice_cancel), + &voice_change_ack, + ); return; } }; @@ -271,7 +362,12 @@ fn tts_worker( eprintln!( "buzz-desktop: TTS voice style load failed ({voice_name}): {e}. TTS disabled." ); - drain_until_shutdown(text_rx, &shutdown); + drain_tts_until_shutdown( + text_rx, + &shutdown, + (&cancel, &voice_cancel), + &voice_change_ack, + ); return; } }; @@ -304,7 +400,12 @@ fn tts_worker( Ok(h) => h, Err(e) => { eprintln!("buzz-desktop: TTS audio output failed: {e}. TTS disabled."); - drain_until_shutdown(text_rx, &shutdown); + drain_tts_until_shutdown( + text_rx, + &shutdown, + (&cancel, &voice_cancel), + &voice_change_ack, + ); return; } }; @@ -374,6 +475,7 @@ fn tts_worker( let monitor = { let player = Arc::clone(&player); let cancel = Arc::clone(&cancel); + let voice_cancel = Arc::clone(&voice_cancel); let tts_active = Arc::clone(&tts_active); let stop = Arc::clone(&monitor_stop); let player_ops = Arc::clone(&player_ops); @@ -381,12 +483,12 @@ fn tts_worker( .name("tts-barge-in-monitor".into()) .spawn(move || { while !stop.load(Ordering::Acquire) { - if cancel.load(Ordering::Acquire) { + if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { let _ops = lock_player_ops(&player_ops); // Re-check under the lock: the worker may have // consumed this cancel (and appended fresh audio) // between the load above and the lock acquisition. - if cancel.load(Ordering::Acquire) { + if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { // clear() pauses the persistent player; play() // un-pauses (see handle_cancel_or_shutdown). // Idempotent — safe to repeat every tick until @@ -420,13 +522,16 @@ fn tts_worker( // idle branch below uses it to decide when to drop `tts_active` and to // arm a fresh lead-in cushion for the next utterance. let mut first_append = true; + let mut deferred_text = VecDeque::new(); loop { + let mut no_current_text = None; if handle_cancel_or_shutdown( - &cancel, + (&cancel, &voice_cancel), &shutdown, &tts_active, - &text_rx, + (&text_rx, &mut deferred_text, &mut no_current_text), + &voice_change_ack, Some((&player, &player_ops)), ) { if shutdown.load(Ordering::Acquire) { @@ -441,65 +546,39 @@ fn tts_worker( // Voice changes cancel the old utterance/queue and are observed here, // before receiving subsequent text. A bad bundled asset falls back to // Mary without discarding the already-warmed Pocket engine. - let requested_voice = selected_voice - .lock() - .unwrap_or_else(|error| error.into_inner()) - .clone(); - if requested_voice != voice_name { - let requested_path = model_dir.join(format!("{requested_voice}.{VOICE_FILE_EXT}")); - match load_voice_style(&requested_path) { - Ok(requested_style) => { - style = requested_style; - voice_name = requested_voice; - } - Err(error) => { - use super::pocket::DEFAULT_VOICE; - eprintln!( - "buzz-desktop: Pocket voice {requested_voice} is unavailable ({error}); falling back to Mary" - ); - let fallback_path = model_dir.join(format!("{DEFAULT_VOICE}.{VOICE_FILE_EXT}")); - match load_voice_style(&fallback_path) { - Ok(fallback_style) => { - style = fallback_style; - voice_name = DEFAULT_VOICE.to_string(); - *selected_voice - .lock() - .unwrap_or_else(|lock_error| lock_error.into_inner()) = - DEFAULT_VOICE.to_string(); - } - Err(fallback_error) => { - eprintln!( - "buzz-desktop: Mary voice fallback is unavailable: {fallback_error}" - ); - continue; - } - } - } - } + let voice_ready = + reconcile_selected_voice(&model_dir, &selected_voice, &mut voice_name, &mut style); + acknowledge_voice_change(&voice_change_ack, &voice_cancel); + if !voice_ready { + continue; } - let raw_text = match text_rx.recv_timeout(RECV_TIMEOUT) { - Ok(t) => t, - Err(mpsc::RecvTimeoutError::Timeout) => { - // Nothing queued. If playback has also finished, the agent - // has gone quiet — release the mic gate and reset the - // lead-in so the next utterance gets a fresh cushion. - if player.empty() && !first_append { - tts_active.store(false, Ordering::Release); - first_append = true; + let mut queued_text = Some(match deferred_text.pop_front() { + Some(text) => text, + None => match text_rx.recv_timeout(RECV_TIMEOUT) { + Ok(text) => text, + Err(mpsc::RecvTimeoutError::Timeout) => { + // Nothing queued. If playback has also finished, the agent + // has gone quiet — release the mic gate and reset the + // lead-in so the next utterance gets a fresh cushion. + if player.empty() && !first_append { + tts_active.store(false, Ordering::Release); + first_append = true; + } + continue; } - continue; - } - Err(mpsc::RecvTimeoutError::Disconnected) => break, - }; + Err(mpsc::RecvTimeoutError::Disconnected) => break, + }, + }); // Check cancel again after unblocking — a cancel may have arrived // while we were waiting. if handle_cancel_or_shutdown( - &cancel, + (&cancel, &voice_cancel), &shutdown, &tts_active, - &text_rx, + (&text_rx, &mut deferred_text, &mut queued_text), + &voice_change_ack, Some((&player, &player_ops)), ) { if shutdown.load(Ordering::Acquire) { @@ -508,6 +587,21 @@ fn tts_worker( first_append = true; continue; } + let Some(queued_text) = queued_text else { + continue; + }; + if queued_text.generation < voice_generation.load(Ordering::Acquire) { + continue; + } + let raw_text = queued_text.text; + + // The selected voice can change while this worker is blocked in + // recv_timeout. Reconcile again after receipt so the first message + // queued after an unpublished pipeline is installed cannot use the + // voice captured when construction began. + if !reconcile_selected_voice(&model_dir, &selected_voice, &mut voice_name, &mut style) { + continue; + } // If playback already drained while we were waiting for this item, // the agent is silent — release the mic gate BEFORE preprocessing/ @@ -541,11 +635,13 @@ fn tts_worker( let chunks = group_sentences_into_chunks(&sentences, MAX_CHUNK_CHARS); 'playback_chunks: for chunk in &chunks { + let mut no_current_text = None; if handle_cancel_or_shutdown( - &cancel, + (&cancel, &voice_cancel), &shutdown, &tts_active, - &text_rx, + (&text_rx, &mut deferred_text, &mut no_current_text), + &voice_change_ack, Some((&player, &player_ops)), ) { first_append = true; @@ -566,11 +662,13 @@ fn tts_worker( }; let model_chunk_count = model_chunks.len(); for (model_chunk_index, model_chunk) in model_chunks.iter().enumerate() { + let mut no_current_text = None; if handle_cancel_or_shutdown( - &cancel, + (&cancel, &voice_cancel), &shutdown, &tts_active, - &text_rx, + (&text_rx, &mut deferred_text, &mut no_current_text), + &voice_change_ack, Some((&player, &player_ops)), ) { first_append = true; @@ -608,11 +706,11 @@ fn tts_worker( // does the full consume (drain queue, reset lead-in) on // the next iteration. let _ops = lock_player_ops(&player_ops); - if cancel.load(Ordering::Acquire) { + if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { // Nothing appended; the loop-top consume re-arms // `first_append` (the flag is still set — the worker // is its only consumer). - break; + break 'playback_chunks; } player.append(SamplesBuffer::new(channels, rate, buf)); // NOTE: tts_active is set AFTER player.append(), not @@ -643,11 +741,133 @@ fn tts_worker( let _ = handle.join(); } + finish_voice_change_ack(&voice_change_ack); tts_active.store(false, Ordering::Release); } // ── Helpers ─────────────────────────────────────────────────────────────────── +fn begin_voice_change( + selected_voice: &Mutex, + voice_generation: &AtomicU64, + voice_cancel: &AtomicBool, + voice_change_ack: &VoiceChangeAck, + voice: &str, +) -> Option> { + let mut pending_ack = voice_change_ack + .lock() + .unwrap_or_else(|error| error.into_inner()); + let mut selected = selected_voice + .lock() + .unwrap_or_else(|error| error.into_inner()); + if selected.as_str() == voice { + return None; + } + + let (sender, receiver) = tokio::sync::oneshot::channel(); + voice_cancel.store(true, Ordering::Release); + let generation = voice_generation.fetch_add(1, Ordering::AcqRel) + 1; + if let Some(superseded) = pending_ack.replace(PendingVoiceChange { + generation, + acknowledged: sender, + }) { + let _ = superseded.acknowledged.send(()); + } + *selected = voice.to_string(); + Some(receiver) +} + +fn acknowledge_voice_change(voice_change_ack: &VoiceChangeAck, voice_cancel: &AtomicBool) { + let mut pending_ack = voice_change_ack + .lock() + .unwrap_or_else(|error| error.into_inner()); + if voice_cancel.load(Ordering::Acquire) { + return; + } + if let Some(pending) = pending_ack.take() { + let _ = pending.acknowledged.send(()); + } +} + +fn finish_voice_change_ack(voice_change_ack: &VoiceChangeAck) { + if let Some(pending) = voice_change_ack + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take() + { + let _ = pending.acknowledged.send(()); + } +} + +fn drain_tts_until_shutdown( + text_rx: mpsc::Receiver, + shutdown: &AtomicBool, + cancel_signals: CancelSignals<'_>, + voice_change_ack: &VoiceChangeAck, +) { + let (cancel, voice_cancel) = cancel_signals; + loop { + if cancel.swap(false, Ordering::AcqRel) | voice_cancel.swap(false, Ordering::AcqRel) { + while text_rx.try_recv().is_ok() {} + } + acknowledge_voice_change(voice_change_ack, voice_cancel); + if shutdown.load(Ordering::Acquire) { + break; + } + match text_rx.recv_timeout(RECV_TIMEOUT) { + Ok(_) | Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => break, + } + } + finish_voice_change_ack(voice_change_ack); +} + +fn reconcile_selected_voice( + model_dir: &std::path::Path, + selected_voice: &Mutex, + voice_name: &mut String, + style: &mut VoiceStyle, +) -> bool { + let requested_voice = selected_voice + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone(); + if requested_voice == *voice_name { + return true; + } + + let requested_path = model_dir.join(format!("{requested_voice}.{VOICE_FILE_EXT}")); + match load_voice_style(&requested_path) { + Ok(requested_style) => { + *style = requested_style; + *voice_name = requested_voice; + true + } + Err(error) => { + use super::pocket::DEFAULT_VOICE; + eprintln!( + "buzz-desktop: Pocket voice {requested_voice} is unavailable ({error}); falling back to Mary" + ); + let fallback_path = model_dir.join(format!("{DEFAULT_VOICE}.{VOICE_FILE_EXT}")); + match load_voice_style(&fallback_path) { + Ok(fallback_style) => { + *style = fallback_style; + *voice_name = DEFAULT_VOICE.to_string(); + *selected_voice + .lock() + .unwrap_or_else(|lock_error| lock_error.into_inner()) = + DEFAULT_VOICE.to_string(); + true + } + Err(fallback_error) => { + eprintln!("buzz-desktop: Mary voice fallback is unavailable: {fallback_error}"); + false + } + } + } + } +} + /// Check for cancel or shutdown. Returns `true` if the caller should break/continue. /// On cancel: drains the text queue and clears the cancel flag. /// @@ -656,12 +876,15 @@ fn tts_worker( /// it is serialized with the monitor's stale-branch re-check (see the monitor /// block in `tts_worker`). fn handle_cancel_or_shutdown( - cancel: &AtomicBool, + cancel_signals: CancelSignals<'_>, shutdown: &AtomicBool, tts_active: &AtomicBool, - text_rx: &mpsc::Receiver, + text_state: CancelTextState<'_>, + voice_change_ack: &VoiceChangeAck, player: Option<(&rodio::Player, &Mutex<()>)>, ) -> bool { + let (cancel, voice_cancel) = cancel_signals; + let (text_rx, deferred_text, current_text) = text_state; if shutdown.load(Ordering::Acquire) { if let Some((p, ops)) = player { let _ops = lock_player_ops(ops); @@ -670,7 +893,24 @@ fn handle_cancel_or_shutdown( tts_active.store(false, Ordering::Release); return true; } - if cancel.load(Ordering::Acquire) { + if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { + // Serialize with begin_voice_change so the generation boundary and + // cancel consumption are observed as one transition. + let pending_voice_change = voice_change_ack + .lock() + .unwrap_or_else(|error| error.into_inner()); + // Consume at the serialization point. A later barge-in remains true + // for the next pass instead of being overwritten after queue cleanup. + let barge_in = cancel.swap(false, Ordering::AcqRel); + voice_cancel.store(false, Ordering::Release); + let preserve_generation = (!barge_in) + .then(|| { + pending_voice_change + .as_ref() + .map(|pending| pending.generation) + }) + .flatten(); + retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation); if let Some((p, ops)) = player { let _ops = lock_player_ops(ops); // `Player::clear()` removes queued sources AND pauses the player @@ -683,11 +923,6 @@ fn handle_cancel_or_shutdown( // Consume the flag under the lock: once released with // `cancel == false`, the monitor's stale branch no-ops instead // of clearing the fresh post-cancel utterance. - while text_rx.try_recv().is_ok() {} - cancel.store(false, Ordering::Release); - } else { - while text_rx.try_recv().is_ok() {} - cancel.store(false, Ordering::Release); } tts_active.store(false, Ordering::Release); return true; @@ -695,6 +930,31 @@ fn handle_cancel_or_shutdown( false } +fn retain_cancelled_text( + deferred_text: &mut VecDeque, + current_text: &mut Option, + text_rx: &mpsc::Receiver, + preserve_generation: Option, +) { + if let Some(generation) = preserve_generation { + deferred_text.retain(|text| text.generation >= generation); + if let Some(text) = current_text.take() { + if text.generation >= generation { + deferred_text.push_front(text); + } + } + while let Ok(text) = text_rx.try_recv() { + if text.generation >= generation { + deferred_text.push_back(text); + } + } + } else { + deferred_text.clear(); + current_text.take(); + while text_rx.try_recv().is_ok() {} + } +} + /// Acquire the `player_ops` lock, recovering from poison. /// /// The data under the mutex is `()` — it only serializes Player mutations — @@ -834,9 +1094,6 @@ fn group_sentences_into_chunks(sentences: &[String], max_chars: usize) -> Vec, + tokio::sync::oneshot::Receiver<()>, +); + const VOICE_AVAILABILITY_BUNDLED: &str = "bundled"; const VOICE_AVAILABILITY_INSTALLED: &str = "installed"; @@ -338,11 +348,28 @@ fn commit_effective_off(state: &AppState) -> Result<(), String> { Ok(()) } +fn enable_tts_runtime(huddle: &mut HuddleState, voice: &str) -> Option { + huddle.tts_enabled = true; + // OFF removes the pipeline. Clear a prior cancellation only when enabling + // a fresh pipeline; an idempotent ON write must not erase a voice + // transition that the existing worker still needs to drain. + if huddle.tts_pipeline.is_none() { + huddle + .tts_cancel + .store(false, std::sync::atomic::Ordering::Release); + } + huddle.tts_pipeline.as_ref().and_then(|pipeline| { + pipeline + .select_voice(voice) + .map(|acknowledged| (Arc::clone(pipeline), acknowledged)) + }) +} + async fn apply_tts_settings( settings: TtsSettings, app: &AppHandle, state: &AppState, -) -> Result { +) -> Result, String> { if settings.version != CURRENT_VERSION { return Err(format!( "Unsupported text-to-speech settings version: {}", @@ -366,18 +393,18 @@ async fn apply_tts_settings( .map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))? = settings.clone(); + let mut voice_change_wait = None; if settings.agent_text_to_speech { - let active = { + let (active, voice_change_ack) = { let mut huddle = state.huddle()?; - huddle.tts_enabled = true; - huddle - .tts_cancel - .store(false, std::sync::atomic::Ordering::Release); - if let Some(pipeline) = huddle.tts_pipeline.as_ref() { - pipeline.select_voice(&pocket_voice_name(&settings.voice_preferences)); - } - matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active) + let voice_change_ack = + enable_tts_runtime(&mut huddle, &pocket_voice_name(&settings.voice_preferences)); + ( + matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active), + voice_change_ack, + ) }; + voice_change_wait = voice_change_ack; if active { if let Err(error) = super::pipeline::maybe_start_tts_pipeline(state).await { eprintln!("buzz-desktop: could not hot-start text to speech: {error}"); @@ -385,7 +412,50 @@ async fn apply_tts_settings( } state.emit_huddle_state_changed(); } - Ok(settings) + Ok(voice_change_wait) +} + +fn current_settings(state: &AppState) -> Result { + state + .tts_settings + .lock() + .map_err(|error| format!("text-to-speech settings lock poisoned: {error}")) + .map(|settings| settings.clone()) +} + +async fn finish_voice_change(voice_change: Option) -> Result<(), String> { + let Some((pipeline, acknowledged)) = voice_change else { + return Ok(()); + }; + wait_for_voice_change_ack(acknowledged, VOICE_CHANGE_ACK_TIMEOUT, || { + pipeline.is_finished() + }) + .await +} + +async fn wait_for_voice_change_ack( + mut acknowledged: tokio::sync::oneshot::Receiver<()>, + timeout: Duration, + mut worker_is_finished: impl FnMut() -> bool, +) -> Result<(), String> { + let deadline = tokio::time::sleep(timeout); + tokio::pin!(deadline); + loop { + tokio::select! { + _ = &mut acknowledged => return Ok(()), + _ = &mut deadline => { + return Err( + "Pocket TTS is still finishing the previous voice. Turn Agent text to speech off and try again." + .to_string(), + ); + } + _ = tokio::time::sleep(Duration::from_millis(25)) => { + if worker_is_finished() { + return Ok(()); + } + } + } + } } /// Compatibility command for the huddle speaker button. It updates the same @@ -396,14 +466,17 @@ pub async fn set_tts_enabled( app: AppHandle, state: State<'_, AppState>, ) -> Result { - let _transition = state.tts_settings_transition.lock().await; + let transition = state.tts_settings_transition.lock().await; let mut settings = state .tts_settings .lock() .map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))? .clone(); settings.agent_text_to_speech = enabled; - apply_tts_settings(settings, &app, &state).await + let voice_change = apply_tts_settings(settings, &app, &state).await?; + drop(transition); + finish_voice_change(voice_change).await?; + current_settings(&state) } fn settings_with_pocket_voice( @@ -436,14 +509,17 @@ pub async fn set_pocket_voice( app: AppHandle, state: State<'_, AppState>, ) -> Result { - let _transition = state.tts_settings_transition.lock().await; + let transition = state.tts_settings_transition.lock().await; let settings = state .tts_settings .lock() .map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))? .clone(); let settings = settings_with_pocket_voice(settings, &voice_key)?; - apply_tts_settings(settings, &app, &state).await + let voice_change = apply_tts_settings(settings, &app, &state).await?; + drop(transition); + finish_voice_change(voice_change).await?; + current_settings(&state) } #[tauri::command] @@ -499,6 +575,49 @@ pub async fn preview_pocket_voice( mod tests { use super::*; + #[tokio::test] + async fn stalled_voice_change_returns_an_actionable_error() { + let (_keep_pending, acknowledged) = tokio::sync::oneshot::channel(); + + let error = wait_for_voice_change_ack(acknowledged, Duration::from_millis(1), || false) + .await + .expect_err("stalled worker should time out"); + + assert!(error.contains("Turn Agent text to speech off")); + } + + #[test] + fn idempotent_enable_preserves_an_existing_pipeline_cancel() { + let state = crate::app_state::build_app_state(); + let model_dir = tempfile::tempdir().expect("temp model dir"); + let cancel = state.huddle().expect("huddle state").tts_cancel.clone(); + let pipeline = Arc::new( + super::super::tts::TtsPipeline::new_with_voice( + model_dir.path().to_path_buf(), + Arc::new(std::sync::atomic::AtomicBool::new(false)), + Arc::clone(&cancel), + "reference_sample", + None, + ) + .expect("pipeline"), + ); + pipeline.shutdown(); + for _ in 0..100 { + if pipeline.is_finished() { + break; + } + std::thread::sleep(Duration::from_millis(1)); + } + assert!(pipeline.is_finished(), "test pipeline should stop"); + + cancel.store(true, std::sync::atomic::Ordering::Release); + let mut huddle = state.huddle().expect("huddle state"); + huddle.tts_pipeline = Some(pipeline); + + assert!(enable_tts_runtime(&mut huddle, "reference_sample").is_none()); + assert!(cancel.load(std::sync::atomic::Ordering::Acquire)); + } + #[test] fn defaults_are_backwards_compatible_and_use_mary() { assert_eq!( diff --git a/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs b/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs index a2a8410bb..04bccd460 100644 --- a/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs +++ b/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs @@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; #[test] -fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() { +fn selecting_a_voice_raises_only_the_internal_cancel_and_retains_the_engine_handle() { let model_dir = tempfile::tempdir().expect("temp model dir"); let active = Arc::new(AtomicBool::new(false)); let cancel = Arc::new(AtomicBool::new(false)); @@ -16,9 +16,10 @@ fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() { ) .expect("pipeline handle"); - pipeline.select_voice("marius"); + let _acknowledged = pipeline.select_voice("marius"); - assert!(cancel.load(Ordering::Acquire)); + assert!(!cancel.load(Ordering::Acquire)); + assert!(pipeline.voice_cancel.load(Ordering::Acquire)); assert_eq!( pipeline .voice @@ -28,3 +29,310 @@ fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() { "marius" ); } + +#[test] +fn reconciling_an_unpublished_pipeline_does_not_cancel_its_first_message() { + let model_dir = tempfile::tempdir().expect("temp model dir"); + let active = Arc::new(AtomicBool::new(false)); + let cancel = Arc::new(AtomicBool::new(false)); + let pipeline = TtsPipeline::new_with_voice( + model_dir.path().to_path_buf(), + active, + Arc::clone(&cancel), + "reference_sample", + None, + ) + .expect("pipeline handle"); + + pipeline.select_voice_before_publish("marius"); + + assert!(!cancel.load(Ordering::Acquire)); + assert_eq!( + pipeline + .voice + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_str(), + "marius" + ); +} + +#[test] +fn received_text_reconciles_a_voice_changed_while_the_worker_was_waiting() { + let model_dir = tempfile::tempdir().expect("temp model dir"); + let bundled_voice = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("resources/pocket-voices/marius.wav"); + std::fs::copy( + &bundled_voice, + model_dir.path().join("reference_sample.wav"), + ) + .expect("Mary test voice"); + std::fs::copy(&bundled_voice, model_dir.path().join("marius.wav")).expect("Marius test voice"); + + let selected_voice = Arc::new(std::sync::Mutex::new("reference_sample".to_string())); + let mut style = + load_voice_style(&model_dir.path().join("reference_sample.wav")).expect("initial style"); + let waiting = Arc::new(std::sync::Barrier::new(2)); + let (text_tx, text_rx) = std::sync::mpsc::channel(); + let worker_voice = Arc::clone(&selected_voice); + let worker_waiting = Arc::clone(&waiting); + let worker_model_dir = model_dir.path().to_path_buf(); + let worker = std::thread::spawn(move || { + let mut voice_name = "reference_sample".to_string(); + worker_waiting.wait(); + let text = text_rx.recv().expect("first queued text"); + assert!(reconcile_selected_voice( + &worker_model_dir, + &worker_voice, + &mut voice_name, + &mut style, + )); + (text, voice_name) + }); + + waiting.wait(); + *selected_voice.lock().expect("selected voice") = "marius".to_string(); + text_tx + .send("first message".to_string()) + .expect("queue first message"); + + assert_eq!( + worker.join().expect("worker"), + ("first message".to_string(), "marius".to_string()) + ); +} + +#[test] +fn an_in_hand_post_change_message_survives_cancellation() { + let selected_voice = Arc::new(std::sync::Mutex::new("reference_sample".to_string())); + let voice_generation = AtomicU64::new(1); + let barge_in = AtomicBool::new(false); + let voice_cancel = Arc::new(AtomicBool::new(false)); + let voice_change_ack = Arc::new(std::sync::Mutex::new(None)); + let (text_tx, text_rx) = std::sync::mpsc::sync_channel(1); + let mut acknowledged = begin_voice_change( + &selected_voice, + &voice_generation, + &voice_cancel, + &voice_change_ack, + "marius", + ) + .expect("voice changed"); + assert!(voice_cancel.load(Ordering::Acquire)); + assert!(matches!( + acknowledged.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + )); + acknowledge_voice_change(&voice_change_ack, &voice_cancel); + assert!(matches!( + acknowledged.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + )); + text_tx + .send(QueuedText { + generation: voice_generation.load(Ordering::Acquire), + text: "new message".to_string(), + }) + .expect("new message"); + let mut current_text = Some(text_rx.recv().expect("in-hand new message")); + + let shutdown = AtomicBool::new(false); + let active = AtomicBool::new(true); + let mut deferred_text = VecDeque::from([ + QueuedText { + generation: 1, + text: "old message".to_string(), + }, + QueuedText { + generation: voice_generation.load(Ordering::Acquire), + text: "later new message".to_string(), + }, + ]); + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + acknowledge_voice_change(&voice_change_ack, &voice_cancel); + acknowledged.blocking_recv().expect("voice change ack"); + + assert_eq!( + deferred_text + .pop_front() + .expect("preserved post-change message") + .text, + "new message" + ); + assert_eq!( + deferred_text + .pop_front() + .expect("later post-change message") + .text, + "later new message" + ); + assert!(text_rx.try_recv().is_err()); +} + +#[test] +fn superseding_voice_change_removes_earlier_deferred_messages() { + let selected_voice = std::sync::Mutex::new("reference_sample".to_string()); + let voice_generation = AtomicU64::new(1); + let barge_in = AtomicBool::new(false); + let voice_cancel = AtomicBool::new(false); + let voice_change_ack = Arc::new(std::sync::Mutex::new(None)); + let (_text_tx, text_rx) = std::sync::mpsc::channel(); + let shutdown = AtomicBool::new(false); + let active = AtomicBool::new(true); + let mut deferred_text = VecDeque::new(); + let mut current_text = None; + + let first = begin_voice_change( + &selected_voice, + &voice_generation, + &voice_cancel, + &voice_change_ack, + "marius", + ) + .expect("first voice change"); + deferred_text.push_back(QueuedText { + generation: voice_generation.load(Ordering::Acquire), + text: "message for Marius".to_string(), + }); + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + acknowledge_voice_change(&voice_change_ack, &voice_cancel); + first.blocking_recv().expect("first acknowledgement"); + + let _second = begin_voice_change( + &selected_voice, + &voice_generation, + &voice_cancel, + &voice_change_ack, + "reference_sample", + ) + .expect("second voice change"); + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + + assert!(deferred_text.is_empty()); +} + +#[test] +fn barge_in_clears_deferred_voice_change_messages() { + let barge_in = AtomicBool::new(true); + let voice_cancel = AtomicBool::new(false); + let shutdown = AtomicBool::new(false); + let active = AtomicBool::new(true); + let voice_change_ack = Arc::new(std::sync::Mutex::new(None)); + let (_text_tx, text_rx) = std::sync::mpsc::channel(); + let mut deferred_text = VecDeque::from([QueuedText { + generation: 2, + text: "deferred message".to_string(), + }]); + let mut current_text = None; + + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + + assert!(deferred_text.is_empty()); +} + +#[test] +fn barge_in_during_a_voice_change_clears_post_change_messages() { + let selected_voice = std::sync::Mutex::new("reference_sample".to_string()); + let voice_generation = AtomicU64::new(1); + let barge_in = AtomicBool::new(false); + let voice_cancel = AtomicBool::new(false); + let voice_change_ack = Arc::new(std::sync::Mutex::new(None)); + let (_text_tx, text_rx) = std::sync::mpsc::channel(); + let shutdown = AtomicBool::new(false); + let active = AtomicBool::new(true); + let mut deferred_text = VecDeque::new(); + let mut current_text = None; + + let _acknowledged = begin_voice_change( + &selected_voice, + &voice_generation, + &voice_cancel, + &voice_change_ack, + "marius", + ) + .expect("voice change"); + deferred_text.push_back(QueuedText { + generation: voice_generation.load(Ordering::Acquire), + text: "post-change message".to_string(), + }); + barge_in.store(true, Ordering::Release); + + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + assert!(deferred_text.is_empty()); +} + +#[test] +fn a_sender_captured_before_voice_change_is_stale_even_if_it_sends_after_drain() { + let selected_voice = std::sync::Mutex::new("reference_sample".to_string()); + let voice_generation = Arc::new(AtomicU64::new(1)); + let barge_in = AtomicBool::new(false); + let voice_cancel = AtomicBool::new(false); + let voice_change_ack = Arc::new(std::sync::Mutex::new(None)); + let (text_tx, text_rx) = std::sync::mpsc::sync_channel(1); + let old_sender = TtsTextSender { + text_tx, + generation: voice_generation.load(Ordering::Acquire), + }; + let shutdown = AtomicBool::new(false); + let active = AtomicBool::new(true); + let mut deferred_text = VecDeque::new(); + let mut current_text = None; + + let _acknowledged = begin_voice_change( + &selected_voice, + &voice_generation, + &voice_cancel, + &voice_change_ack, + "marius", + ) + .expect("voice change"); + assert!(handle_cancel_or_shutdown( + (&barge_in, &voice_cancel), + &shutdown, + &active, + (&text_rx, &mut deferred_text, &mut current_text), + &voice_change_ack, + None, + )); + old_sender + .send("late old message".to_string()) + .expect("late send"); + let late = text_rx.recv().expect("late queued text"); + + assert!(late.generation < voice_generation.load(Ordering::Acquire)); +}