From ce3cf3cd2591f132f286fbc0a42a9e6699d0b08d Mon Sep 17 00:00:00 2001 From: klopez4212 Date: Tue, 4 Aug 2026 21:07:50 +0100 Subject: [PATCH] Polish Huddle voice controls (#4694) ## Summary - add a visible Stop control for interrupting agent speech - make push-to-talk available by default while preserving manual mute controls - refine agent management, muted audio states, drawer layering, and return navigation - suppress duplicate notification sounds for Huddle messages ## Why Huddles could trap users behind long agent speech, hide useful agent controls, and leave temporary Huddle state visible after the call. The drawer also regressed when the terminal substrate began painting behind the rounded app surface. ## Validation - `just desktop-ci` - focused Huddle Playwright coverage for the drawer, speech interruption, agent picker, and leave navigation --------- Signed-off-by: kenny lopez Signed-off-by: Wes Co-authored-by: Wes Co-authored-by: Carl --- desktop/src-tauri/src/huddle/commands.rs | 132 +++++ desktop/src-tauri/src/huddle/mod.rs | 30 +- desktop/src-tauri/src/huddle/pipeline.rs | 14 +- desktop/src-tauri/src/huddle/state.rs | 17 +- desktop/src-tauri/src/huddle/stt.rs | 62 ++- desktop/src-tauri/src/huddle/tts.rs | 344 ++++++------- .../src/huddle/tts_pipeline_controls.rs | 100 ++++ .../src/huddle/tts_speaker_cancellation.rs | 177 +++++++ .../src/huddle/tts_voice_selection_tests.rs | 12 + .../src/huddle/tts_voice_transition.rs | 437 ++++++++++++++++ desktop/src-tauri/src/lib.rs | 10 +- desktop/src/app/AppHuddleShell.tsx | 9 +- desktop/src/app/AppShell.tsx | 4 +- .../app/useAppShellDesktopNotifications.ts | 15 +- desktop/src/app/useHuddlePresentation.ts | 25 +- desktop/src/features/huddle/HuddleContext.tsx | 75 ++- .../features/huddle/HuddleContext.types.ts | 2 + .../huddle/components/AddAgentDialog.tsx | 197 ++++--- .../huddle/components/AgentVoiceMenu.tsx | 126 +++-- .../features/huddle/components/HuddleBar.tsx | 54 +- .../components/HuddleProfileControl.tsx | 2 - .../huddle/components/HuddleRoomHeader.tsx | 64 +-- .../huddle/components/MicControls.tsx | 66 +-- .../huddle/components/ParticipantList.tsx | 88 +++- .../src/features/huddle/lib/audioWorklet.ts | 58 +-- .../features/huddle/lib/useHuddlePttState.ts | 18 +- desktop/src/features/notifications/hooks.ts | 2 + .../features/notifications/lib/sound.test.mjs | 18 + .../src/features/notifications/lib/sound.ts | 7 + .../use-feed-desktop-notifications.ts | 7 +- .../src/shared/styles/globals/components.css | 31 +- .../src/shared/ui/chooser-dialog-content.tsx | 12 +- desktop/src/testing/e2eBridge.ts | 48 +- .../tests/e2e/huddle-transcription.spec.ts | 479 +++++++++++++++++- 34 files changed, 2196 insertions(+), 546 deletions(-) create mode 100644 desktop/src-tauri/src/huddle/commands.rs create mode 100644 desktop/src-tauri/src/huddle/tts_pipeline_controls.rs create mode 100644 desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs create mode 100644 desktop/src/features/notifications/lib/sound.test.mjs diff --git a/desktop/src-tauri/src/huddle/commands.rs b/desktop/src-tauri/src/huddle/commands.rs new file mode 100644 index 0000000000..993d8e54eb --- /dev/null +++ b/desktop/src-tauri/src/huddle/commands.rs @@ -0,0 +1,132 @@ +//! Small Huddle controls that mutate an active session. + +use std::sync::{atomic::Ordering, Arc}; + +use tauri::State; +use uuid::Uuid; + +use crate::{app_state::AppState, events, relay::submit_event}; + +use super::{relay_api::validate_pubkey_hex, HuddlePhase}; + +/// Update the clickable microphone control independently from the PTT shortcut. +#[tauri::command] +pub fn set_huddle_manual_mic_unmuted( + enabled: bool, + state: State<'_, AppState>, +) -> Result<(), String> { + let huddle = state.huddle()?; + if !matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active) { + return Err("no active huddle".to_string()); + } + huddle.manual_mic_unmuted.store(enabled, Ordering::Release); + Ok(()) +} + +/// Immediately interrupt the agent utterance that is currently speaking. +#[tauri::command] +pub fn interrupt_huddle_speech( + agent_pubkey: String, + state: State<'_, AppState>, +) -> Result<(), String> { + validate_pubkey_hex(&agent_pubkey)?; + let tts_pipeline = { + let huddle = state.huddle()?; + if !matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active) { + return Err("no active huddle".to_string()); + } + huddle.tts_pipeline.as_ref().map(Arc::clone) + }; + if let Some(tts_pipeline) = tts_pipeline { + tts_pipeline.cancel_active_speaker(&agent_pubkey); + } + Ok(()) +} + +/// Remove an agent from the active huddle without removing its parent-channel +/// membership. Keeping the parent membership intact means it remains available +/// to rejoin this huddle from the agent picker. +#[tauri::command] +pub async fn remove_agent_from_huddle( + agent_pubkey: String, + state: State<'_, AppState>, +) -> Result<(), String> { + validate_pubkey_hex(&agent_pubkey)?; + + let (ephemeral_channel_id, huddle_generation) = { + let huddle = state.huddle()?; + if !matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active) { + return Err("no active huddle".to_string()); + } + + let is_huddle_agent = huddle + .agent_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) + .iter() + .any(|pubkey| pubkey.eq_ignore_ascii_case(&agent_pubkey)); + if !is_huddle_agent { + return Err("agent is not in this huddle".to_string()); + } + + ( + huddle + .ephemeral_channel_id + .clone() + .ok_or("no ephemeral channel")?, + huddle.huddle_generation, + ) + }; + + let ephemeral_channel_uuid = + Uuid::parse_str(&ephemeral_channel_id).map_err(|error| error.to_string())?; + submit_event( + events::build_remove_member(ephemeral_channel_uuid, &agent_pubkey)?, + &state, + ) + .await?; + + let (roster_changed, tts_pipeline) = { + let mut huddle = state.huddle()?; + if !huddle.is_current_huddle(&ephemeral_channel_id, huddle_generation) { + (false, None) + } else { + let mut agent_pubkeys = huddle + .agent_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()); + let initial_count = agent_pubkeys.len(); + agent_pubkeys.retain(|pubkey| !pubkey.eq_ignore_ascii_case(&agent_pubkey)); + let changed = agent_pubkeys.len() != initial_count; + drop(agent_pubkeys); + + if changed { + huddle + .participants + .retain(|pubkey| !pubkey.eq_ignore_ascii_case(&agent_pubkey)); + if let Some(settings_pubkey) = huddle + .agent_voice_settings + .keys() + .find(|pubkey| pubkey.eq_ignore_ascii_case(&agent_pubkey)) + .cloned() + { + huddle.agent_voice_settings.remove(&settings_pubkey); + } + } + let tts_pipeline = changed + .then_some(huddle.tts_pipeline.as_ref()) + .flatten() + .map(Arc::clone); + (changed, tts_pipeline) + } + }; + + if let Some(tts_pipeline) = tts_pipeline { + tts_pipeline.cancel_speaker(&agent_pubkey); + } + if roster_changed { + state.emit_huddle_state_changed(); + } + + Ok(()) +} diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index 99337400c9..fcf29d688b 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -27,6 +27,7 @@ mod agent_tts_routing; pub mod agent_voice; pub mod agents; pub mod audio_output; +mod commands; pub mod jitter; pub mod models; pub mod pipeline; @@ -67,6 +68,9 @@ pub(super) fn drain_until_shutdown( // ── Re-exports ──────────────────────────────────────────────────────────────── +pub use commands::{ + interrupt_huddle_speech, remove_agent_from_huddle, set_huddle_manual_mic_unmuted, +}; pub use state::{HuddleJoinInfo, HuddlePhase, HuddleState, VoiceInputMode}; pub use transcription::{set_huddle_transcription_enabled, start_stt_pipeline}; pub use tts_settings::set_tts_enabled; @@ -868,11 +872,27 @@ pub async fn speak_agent_message( let sender = { let hs = state.huddle()?; + let agent_is_present = hs + .agent_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) + .iter() + .any(|pubkey| pubkey.eq_ignore_ascii_case(&speaker_pubkey)); + if !agent_is_present { + eprintln!( + "buzz-desktop: tts stage=queue status=dropped reason=speaker_removed route_id={route_id}" + ); + return Ok(()); + } hs.tts_pipeline .as_ref() .map(|pipeline| pipeline.text_sender()) + .map(|sender| { + let speaker_generation = sender.speaker_generation(&speaker_pubkey); + (sender, speaker_generation) + }) }; - let Some(sender) = sender else { + let Some((sender, speaker_generation)) = sender else { eprintln!( "buzz-desktop: tts stage=invoke status=failed reason=unavailable route_id={route_id}" ); @@ -880,7 +900,13 @@ pub async fn speak_agent_message( }; enqueue_agent_tts_text(route_id, text, move |route_id, text| { sender - .send(route_id, speaker_pubkey, voice_reference, text) + .send( + route_id, + speaker_pubkey, + speaker_generation, + voice_reference, + text, + ) .map_err(|error| format!("TTS queue closed while waiting to enqueue: {error}")) }) .await diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 9572ac25bf..e523ee22bf 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -315,6 +315,7 @@ pub(crate) async fn maybe_start_stt_pipeline( expected_generation, stt_starting, ptt_active_for_stt, + manual_mic_unmuted_for_stt, old_stt, ) = { let mut hs = state.huddle()?; @@ -338,6 +339,11 @@ pub(crate) async fn maybe_start_stt_pipeline( } else { None }; + let manual_mic_unmuted = if hs.voice_input_mode == VoiceInputMode::PushToTalk { + Some(Arc::clone(&hs.manual_mic_unmuted)) + } else { + None + }; ( Arc::clone(&hs.tts_active), Arc::clone(&hs.agent_pubkeys), @@ -345,6 +351,7 @@ pub(crate) async fn maybe_start_stt_pipeline( hs.session_generation.load(Ordering::Acquire), stt_starting, ptt, + manual_mic_unmuted, old, ) }; @@ -352,7 +359,12 @@ pub(crate) async fn maybe_start_stt_pipeline( drop(old_stt); let constructed = tokio::task::spawn_blocking(move || { - stt::SttPipeline::new(model_dir, tts_active, ptt_active_for_stt) + stt::SttPipeline::new( + model_dir, + tts_active, + ptt_active_for_stt, + manual_mic_unmuted_for_stt, + ) }) .await; let (pipeline, text_rx) = match constructed { diff --git a/desktop/src-tauri/src/huddle/state.rs b/desktop/src-tauri/src/huddle/state.rs index 0fe3a46f5a..7acf5fe633 100644 --- a/desktop/src-tauri/src/huddle/state.rs +++ b/desktop/src-tauri/src/huddle/state.rs @@ -15,7 +15,7 @@ use super::{stt, tts}; /// Voice input mode: push-to-talk (PTT) or voice-activity detection (VAD). /// -/// PTT: mic is gated by a global shortcut (Ctrl+Space). Pressing the key sets +/// PTT (the default): mic is gated by a global shortcut (Ctrl+Space). Pressing the key sets /// `ptt_active` and immediately cancels any playing TTS. Releasing the key /// (after a 200 ms delay) stops mic capture and flushes the utterance. /// @@ -26,8 +26,8 @@ use super::{stt, tts}; #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "snake_case")] pub enum VoiceInputMode { - PushToTalk, #[default] + PushToTalk, VoiceActivity, } @@ -135,6 +135,10 @@ pub struct HuddleState { /// Shared with the STT pipeline for mic gating. #[serde(skip)] pub ptt_active: Arc, + /// True while the clickable microphone control is manually unmuted. + /// In PTT mode, either this flag or `ptt_active` opens the STT gate. + #[serde(skip)] + pub manual_mic_unmuted: Arc, } fn serialize_agent_pubkeys(v: &Arc>>, s: S) -> Result @@ -190,6 +194,7 @@ impl Clone for HuddleState { session_generation: Arc::clone(&self.session_generation), voice_input_mode: self.voice_input_mode.clone(), ptt_active: Arc::clone(&self.ptt_active), + manual_mic_unmuted: Arc::clone(&self.manual_mic_unmuted), } } } @@ -221,6 +226,7 @@ impl Default for HuddleState { session_generation: Arc::new(AtomicU64::new(0)), voice_input_mode: VoiceInputMode::default(), ptt_active: Arc::new(AtomicBool::new(false)), + manual_mic_unmuted: Arc::new(AtomicBool::new(true)), } } } @@ -332,6 +338,13 @@ mod tests { assert!(!state.maybe_auto_enable_transcription_for_agents()); } + #[test] + fn defaults_to_push_to_talk_with_an_open_microphone() { + let state = HuddleState::default(); + assert_eq!(state.voice_input_mode, super::VoiceInputMode::PushToTalk); + assert!(state.manual_mic_unmuted.load(Ordering::Acquire)); + } + #[test] fn explicit_user_disable_is_not_undone_by_agent_presence() { let mut state = HuddleState::default(); diff --git a/desktop/src-tauri/src/huddle/stt.rs b/desktop/src-tauri/src/huddle/stt.rs index 30a47f449a..70a8088640 100644 --- a/desktop/src-tauri/src/huddle/stt.rs +++ b/desktop/src-tauri/src/huddle/stt.rs @@ -71,9 +71,10 @@ impl SttPipeline { /// therefore never cancel TTS. Push-to-talk and remote participant speech /// remain explicit, reliable barge-in paths. /// - /// `ptt_active` (optional) is the push-to-talk flag. When `Some`, the STT - /// pipeline only accumulates speech while the flag is true (key held). - /// When `None`, the pipeline runs in continuous VAD mode. + /// `ptt_active` and `manual_mic_unmuted` are present when the PTT shortcut + /// is enabled. The pipeline accepts speech while either input path is open; + /// manual unmute uses normal VAD flushing while a shortcut hold is grouped + /// into one utterance. /// /// Returns `Err` only if the thread cannot be spawned (OS error). /// If model files are missing, the worker logs and exits cleanly — @@ -87,6 +88,7 @@ impl SttPipeline { model_dir: PathBuf, tts_active: Arc, ptt_active: Option>, + manual_mic_unmuted: Option>, ) -> Result<(Self, tokio_mpsc::Receiver), String> { let (audio_tx, audio_rx) = mpsc::sync_channel::>(AUDIO_QUEUE_DEPTH); let (text_tx, text_rx) = tokio_mpsc::channel::(64); @@ -94,6 +96,7 @@ impl SttPipeline { let shutdown_worker = Arc::clone(&shutdown); let ptt_active_worker = ptt_active.as_ref().map(Arc::clone); + let manual_mic_unmuted_worker = manual_mic_unmuted.as_ref().map(Arc::clone); let handle = thread::Builder::new() .name("stt-worker".into()) .spawn(move || { @@ -104,6 +107,7 @@ impl SttPipeline { shutdown_worker, tts_active, ptt_active_worker, + manual_mic_unmuted_worker, ) }) .map_err(|e| format!("failed to spawn stt-worker thread: {e}"))?; @@ -203,6 +207,7 @@ fn stt_worker( shutdown: Arc, tts_active: Arc, ptt_active: Option>, + manual_mic_unmuted: Option>, ) { // ── 1. Initialise rubato resampler (48 kHz → 16 kHz, mono) ─────────────── use rubato::{Fft, FixedSync, Resampler}; @@ -275,9 +280,12 @@ fn stt_worker( // ── 5. Main loop ────────────────────────────────────────────────────────── let mut tts_was_active = false; - let mut ptt_was_active = ptt_active + let mut transmit_was_active = ptt_active .as_ref() - .is_some_and(|p| p.load(Ordering::Acquire)); + .is_some_and(|ptt| ptt.load(Ordering::Acquire)) + || manual_mic_unmuted + .as_ref() + .is_some_and(|manual| manual.load(Ordering::Acquire)); loop { // Check shutdown flag before blocking. if shutdown.load(Ordering::Acquire) { @@ -292,20 +300,22 @@ fn stt_worker( } tts_was_active = tts_now; - // Track PTT transitions — flush accumulated speech when key is released. - // The worklet stops sending frames when PTT is inactive, so the normal - // silence-accumulation flush path never runs. We must flush here on the - // active→inactive edge to avoid buffering speech across PTT presses. + // Track the combined manual/PTT transmission edge. When both paths + // close, the worklet stops sending frames, so flush here rather than + // waiting for silence that will never arrive. if let Some(ref ptt) = ptt_active { - let ptt_now = ptt.load(Ordering::Acquire); - if ptt_was_active && !ptt_now && in_speech && !speech_buf.is_empty() { + let transmit_now = ptt.load(Ordering::Acquire) + || manual_mic_unmuted + .as_ref() + .is_some_and(|manual| manual.load(Ordering::Acquire)); + if transmit_was_active && !transmit_now && in_speech && !speech_buf.is_empty() { flush_to_stt(&speech_buf, voiced_frames, &recognizer, &text_tx); speech_buf.clear(); silence_frames = 0; in_speech = false; voiced_frames = 0; } - ptt_was_active = ptt_now; + transmit_was_active = transmit_now; } // Use recv_timeout so we can periodically check the shutdown flag. @@ -343,6 +353,7 @@ fn stt_worker( &tts_active, &mut tts_stopped_at, ptt_active.as_ref(), + manual_mic_unmuted.as_ref(), ); } } @@ -385,11 +396,9 @@ fn resample_chunk(resampler: &mut rubato::Fft, chunk_48k: &[f32]) -> Vec, tts_stopped_at: &mut Option, ptt_active: Option<&Arc>, + manual_mic_unmuted: Option<&Arc>, ) { leftover.extend_from_slice(samples); @@ -413,13 +423,11 @@ fn process_16k_samples( let prob = vad.predict_f32(&clamped); let is_speech = prob > VAD_THRESHOLD; - // PTT gating: when PTT key is not held, treat as silence. - // This causes natural flush when the key is released — silence_frames - // accumulates and the existing flush logic kicks in after - // SILENCE_FLUSH_FRAMES. The 200 ms release delay + ~300 ms silence - // flush gives a natural utterance tail. + let manually_open = manual_mic_unmuted.is_some_and(|manual| manual.load(Ordering::Acquire)); + // Shortcut-enabled mode accepts input from either the held shortcut or + // a manually open microphone. let is_speech = if let Some(ptt) = ptt_active { - is_speech && ptt.load(Ordering::Acquire) + is_speech && (ptt.load(Ordering::Acquire) || manually_open) } else { is_speech }; @@ -478,11 +486,9 @@ fn process_16k_samples( speech_buf.extend_from_slice(&frame); *silence_frames += 1; - // In PTT mode, don't flush on silence — accumulate the entire - // key-hold as one utterance. The PTT release edge in the main - // loop handles the flush. In VAD mode, flush after the silence - // threshold so each natural pause becomes a separate message. - if ptt_active.is_none() && *silence_frames >= SILENCE_FLUSH_FRAMES { + // A manually open microphone behaves like normal VAD. A + // shortcut-only transmission stays grouped until key release. + if (ptt_active.is_none() || manually_open) && *silence_frames >= SILENCE_FLUSH_FRAMES { // End of utterance — transcribe. flush_to_stt(speech_buf, *voiced_frames, recognizer, text_tx); speech_buf.clear(); diff --git a/desktop/src-tauri/src/huddle/tts.rs b/desktop/src-tauri/src/huddle/tts.rs index 1901bb3d2e..6a56f85444 100644 --- a/desktop/src-tauri/src/huddle/tts.rs +++ b/desktop/src-tauri/src/huddle/tts.rs @@ -64,6 +64,11 @@ use audio::*; #[path = "tts_activity.rs"] mod activity; use activity::*; +#[path = "tts_pipeline_controls.rs"] +mod pipeline_controls; +#[path = "tts_speaker_cancellation.rs"] +mod speaker_cancellation; +use speaker_cancellation::*; // ── Constants ───────────────────────────────────────────────────────────────── @@ -126,7 +131,15 @@ const MAX_CHUNK_CHARS: usize = 200; /// Injected as a silent buffer between each synthesized sentence chunk. const INTER_SENTENCE_SILENCE: f32 = 0.1; -type WorkerControlState = (Arc, Arc, WorkerCancelSignals); +type WorkerControlState = ( + Arc, + Arc, + WorkerCancelSignals, + SpeakerGenerations, + ActiveSpeaker, + SpeakerCancellation, + PlaybackProbe, +); // ── Public pipeline handle ──────────────────────────────────────────────────── @@ -154,6 +167,15 @@ pub struct TtsPipeline { voice: Arc>, /// Tags messages so a voice change drops only pre-change queue entries. voice_generation: Arc, + /// Per-agent generations let removal invalidate that agent's queued and + /// in-flight text without poisoning speech queued after the agent rejoins. + speaker_generations: SpeakerGenerations, + /// Speaker whose audio currently owns the shared player queue. + active_speaker: ActiveSpeaker, + /// Targeted cancellation used when an agent leaves the huddle. + speaker_cancel: SpeakerCancellation, + /// Shared player handle used to reject Stop clicks after playback drains. + playback_probe: PlaybackProbe, /// 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. @@ -187,6 +209,14 @@ impl TtsPipeline { let voice_worker = Arc::clone(&voice); let voice_generation = Arc::new(AtomicU64::new(1)); let worker_voice_generation = Arc::clone(&voice_generation); + let speaker_generations = Arc::new(Mutex::new(HashMap::new())); + let worker_speaker_generations = Arc::clone(&speaker_generations); + let active_speaker = Arc::new(Mutex::new(None)); + let worker_active_speaker = Arc::clone(&active_speaker); + let speaker_cancel = Arc::new(Mutex::new(None)); + let worker_speaker_cancel = Arc::clone(&speaker_cancel); + let playback_probe = PlaybackProbe::new(); + let worker_playback_probe = playback_probe.clone(); 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(); @@ -207,6 +237,10 @@ impl TtsPipeline { tts_active_worker, shutdown_worker, (cancel_worker, worker_voice_cancel), + worker_speaker_generations, + worker_active_speaker, + worker_speaker_cancel, + worker_playback_probe, ), output_device, activity_app, @@ -224,79 +258,14 @@ impl TtsPipeline { voice_cancel, voice, voice_generation, + speaker_generations, + active_speaker, + speaker_cancel, + playback_probe, voice_change_ack, thread: Some(handle), }) } - - /// Queue `text` for TTS synthesis and playback. - /// - /// 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(QueuedText { - generation: self.voice_generation.load(Ordering::Acquire), - route_id: 0, - speaker_pubkey: None, - voice_reference: None, - 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) -> TtsTextSender { - TtsTextSender { - text_tx: self.text_tx.clone(), - generation: self.voice_generation.load(Ordering::Acquire), - } - } - - /// Select a bundled Pocket voice for subsequent speech. - /// - /// 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) -> Option> { - let acknowledged = begin_voice_change( - &self.voice, - &self.voice_generation, - &self.voice_cancel, - &self.voice_change_ack, - voice, - ); - if acknowledged.is_some() { - eprintln!("buzz-desktop: tts stage=cancellation reason=voice_switch route_id=0"); - } - acknowledged - } - - /// 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(); - } - - /// Signal the worker thread to stop. - pub fn shutdown(&self) { - eprintln!("buzz-desktop: tts stage=cancellation reason=shutdown route_id=0"); - self.shutdown.store(true, Ordering::Release); - } - - /// Returns `true` if the worker thread has exited (init failure, crash, or normal exit). - /// Used by hot-start to detect dead pipelines and clear them for retry. - pub fn is_finished(&self) -> bool { - self.thread.as_ref().is_none_or(|h| h.is_finished()) - } } impl Drop for TtsPipeline { @@ -322,7 +291,15 @@ fn tts_worker( startup_tx: mpsc::SyncSender>, ) { let (selected_voice, voice_generation, voice_change_ack) = voice_state; - let (tts_active, shutdown, cancel_signals) = control_state; + let ( + tts_active, + shutdown, + cancel_signals, + speaker_generations, + active_speaker, + speaker_cancel, + playback_probe, + ) = control_state; let (cancel, voice_cancel) = cancel_signals; // ── 1. Initialise TTS engine ────────────────────────────────────────────── let model_dir_str = model_dir.to_string_lossy().to_string(); @@ -415,6 +392,7 @@ fn tts_worker( // Shared (Arc) with the barge-in monitor thread below, which needs to // silence it while this thread is blocked inside `synth_chunk`. let player = Arc::new(Player::connect_new(sink_handle.mixer())); + playback_probe.install(Arc::clone(&player)); // Prime the audio output stream with a short silent buffer. // On macOS, CoreAudio initializes the output device lazily on first use. @@ -443,103 +421,21 @@ fn tts_worker( } eprintln!("buzz-desktop: tts stage=startup status=ready"); - // ── 3b. Barge-in monitor thread ─────────────────────────────────────────── - // - // The worker loop only observes `cancel` between sentences — while it is - // blocked inside `synth_chunk` (hundreds of ms for a long sentence), - // nothing would silence the audio that is already playing. The monitor - // closes that gap: every MONITOR_TICK it checks the flag and, while set, - // silences the player and releases the mic gate. It does NOT consume the - // flag — the worker still owns that (drain queue, reset lead-in), so the - // monitor keeps re-clearing until the worker catches up, which also - // covers a sentence appended in the race window after the worker's own - // post-synthesis cancel check. - // - // `player_ops` closes the converse race (found in review): the monitor - // loads `cancel == true`, is preempted, the worker consumes the cancel - // and appends a fresh post-cancel utterance, then the monitor resumes - // from its stale branch and deletes audio that was meant to play. All - // worker player mutations (appends and cancel/shutdown clears) hold this - // lock, and the monitor re-checks `cancel` *while holding it* — so its - // clear either runs before fresh audio can be appended, or observes - // `cancel == false` and no-ops. The lock is uncontended except during an - // actual barge-in, so the hot path is unaffected. - let player_ops = Arc::new(Mutex::new(())); + let player_ops = Arc::clone(&playback_probe.player_ops); let activity_frames = Arc::new(Mutex::new(VecDeque::::new())); let monitor_stop = Arc::new(AtomicBool::new(false)); - 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); - let activity_frames = Arc::clone(&activity_frames); - thread::Builder::new() - .name("tts-barge-in-monitor".into()) - .spawn(move || { - let mut last_activity_pubkey: Option = None; - let mut next_activity_tick = Instant::now(); - while !stop.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) || 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 - // the worker consumes the flag. - player.clear(); - player.play(); - tts_active.store(false, Ordering::Release); - } - } - if let Some(ref app) = activity_app { - if tts_active.load(Ordering::Acquire) { - let now = Instant::now(); - if now >= next_activity_tick { - let frame = activity_frames - .lock() - .unwrap_or_else(|error| error.into_inner()) - .pop_front(); - if let Some(frame) = frame { - use tauri::Emitter; - let _ = app.emit( - "huddle-tts-speaker-level", - TtsSpeakerActivityPayload { - pubkey: Some(frame.pubkey.clone()), - level: frame.level, - }, - ); - last_activity_pubkey = Some(frame.pubkey); - } - next_activity_tick = now + SPEAKER_ACTIVITY_TICK; - } - } else { - let had_activity = last_activity_pubkey.take().is_some(); - activity_frames - .lock() - .unwrap_or_else(|error| error.into_inner()) - .clear(); - if had_activity { - use tauri::Emitter; - let _ = app.emit( - "huddle-tts-speaker-level", - TtsSpeakerActivityPayload { - pubkey: None, - level: 0.0, - }, - ); - } - next_activity_tick = Instant::now(); - } - } - thread::sleep(MONITOR_TICK); - } - }) - }; + let monitor = spawn_tts_monitor(TtsMonitorState { + player: Arc::clone(&player), + cancel: Arc::clone(&cancel), + voice_cancel: Arc::clone(&voice_cancel), + tts_active: Arc::clone(&tts_active), + stop: Arc::clone(&monitor_stop), + player_ops: Arc::clone(&player_ops), + activity_frames: Arc::clone(&activity_frames), + active_speaker: Arc::clone(&active_speaker), + speaker_cancel: Arc::clone(&speaker_cancel), + activity_app, + }); if let Err(ref e) = monitor { // Degraded but functional: barge-in still works between sentences // via the worker's own checks, just not mid-synthesis. @@ -564,7 +460,8 @@ fn tts_worker( let mut deferred_text = VecDeque::new(); let append_audio = |prepared: PreparedModelAudio, route_id: u64, - speaker_pubkey: Option<&str>| { + speaker_pubkey: Option<&str>, + speaker_generation: u64| { let _ops = lock_player_ops(&player_ops); if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) @@ -582,6 +479,30 @@ fn tts_worker( ); return false; } + let speaker_is_current = speaker_pubkey.is_none_or(|pubkey| { + current_speaker_generation(&speaker_generations, pubkey) == speaker_generation + }); + if !speaker_is_current { + eprintln!( + "buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}" + ); + return false; + } + if let Some(pubkey) = speaker_pubkey { + let mut active = active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()); + if player.empty() { + active.take(); + } + if active + .as_deref() + .is_some_and(|current| !current.eq_ignore_ascii_case(pubkey)) + { + return false; + } + active.get_or_insert_with(|| pubkey.to_ascii_lowercase()); + } if let Some(pubkey) = speaker_pubkey { activity_frames .lock() @@ -604,6 +525,17 @@ fn tts_worker( loop { let mut no_current_text = None; + if consume_speaker_cancel( + &speaker_cancel, + &active_speaker, + &speaker_generations, + &tts_active, + (&text_rx, &mut deferred_text, &mut no_current_text), + Some((&player, &player_ops)), + ) { + first_append = true; + continue; + } if handle_cancel_or_shutdown( (&cancel, &voice_cancel), &shutdown, @@ -647,6 +579,10 @@ fn tts_worker( // lead-in so the next utterance gets a fresh cushion. if player.empty() && !first_append { tts_active.store(false, Ordering::Release); + active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take(); eprintln!( "buzz-desktop: tts stage=player status=drained route_id={last_route_id}" ); @@ -679,6 +615,13 @@ fn tts_worker( let Some(queued_text) = queued_text else { continue; }; + if !queued_speaker_is_current(&speaker_generations, &queued_text) { + eprintln!( + "buzz-desktop: tts stage=queue status=dropped reason=speaker_removed route_id={}", + queued_text.route_id + ); + continue; + } if queued_text.generation < voice_generation.load(Ordering::Acquire) { eprintln!( "buzz-desktop: tts stage=queue status=dropped reason=voice_switch route_id={}", @@ -686,6 +629,22 @@ fn tts_worker( ); continue; } + if !player.empty() + && queued_text + .speaker_pubkey + .as_deref() + .is_some_and(|speaker| { + active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_deref() + .is_some_and(|active| !active.eq_ignore_ascii_case(speaker)) + }) + { + deferred_text.push_front(queued_text); + thread::sleep(RECV_TIMEOUT); + continue; + } let requested_voice = queued_text.voice_reference.unwrap_or_else(|| { selected_voice .lock() @@ -694,9 +653,32 @@ fn tts_worker( }); let raw_text = queued_text.text; let speaker_pubkey = queued_text.speaker_pubkey; + let speaker_generation = queued_text.speaker_generation; let route_id = queued_text.route_id; eprintln!("buzz-desktop: tts stage=synthesis status=started route_id={route_id}"); + // If playback already drained while we were waiting for this item, + // release stale ownership before doing any potentially slow voice or + // synthesis work. Serialize the drain decision with Stop and append so + // those paths observe one coherent utterance boundary. + { + let _ops = lock_player_ops(&player_ops); + if player.empty() && !first_append { + tts_active.store(false, Ordering::Release); + active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take(); + eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}"); + first_append = true; + } + } + + // From this point until the item finishes, an empty player can mean a + // voice-preparation or synthesis gap rather than a drained utterance. + // Stop must remain able to invalidate the in-flight speaker generation. + let _synthesis_flight = playback_probe.begin_synthesis(); + // The selected per-agent voice travels with the queue item, preserving // message order while allowing one warmed Pocket engine to alternate // between cached reference styles. @@ -714,20 +696,6 @@ fn tts_worker( continue; } - // If playback already drained while we were waiting for this item, - // the agent is silent — release the mic gate BEFORE preprocessing/ - // synthesis. Without this, an item arriving inside the recv timeout - // window would run the whole synthesis pass with `tts_active` stuck - // true and nothing playing, making STT discard human speech as - // "echo" during a silent window. (Pipelining is unaffected: when - // audio is still draining, `player.empty()` is false and the flag - // stays set across items.) - if player.empty() && !first_append { - tts_active.store(false, Ordering::Release); - eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}"); - first_append = true; - } - // Preprocess text. let text = preprocess_for_tts(&raw_text); if text.is_empty() { @@ -846,7 +814,12 @@ fn tts_worker( silence_buf_len, player.empty(), ) { - if !append_audio(prepared, route_id, speaker_pubkey.as_deref()) { + if !append_audio( + prepared, + route_id, + speaker_pubkey.as_deref(), + speaker_generation, + ) { first_append = true; synthesis_outcome = "cancelled"; break 'playback_chunks; @@ -872,7 +845,12 @@ fn tts_worker( if let Some(prepared) = playback_audio.finish(&mut first_append, silence_buf_len, player.empty()) { - if !append_audio(prepared, route_id, speaker_pubkey.as_deref()) { + if !append_audio( + prepared, + route_id, + speaker_pubkey.as_deref(), + speaker_generation, + ) { first_append = true; synthesis_outcome = "cancelled"; break 'playback_chunks; diff --git a/desktop/src-tauri/src/huddle/tts_pipeline_controls.rs b/desktop/src-tauri/src/huddle/tts_pipeline_controls.rs new file mode 100644 index 0000000000..0ee472f0fd --- /dev/null +++ b/desktop/src-tauri/src/huddle/tts_pipeline_controls.rs @@ -0,0 +1,100 @@ +use super::*; + +impl TtsPipeline { + /// Queue `text` for TTS synthesis and playback. + /// + /// 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(QueuedText { + generation: self.voice_generation.load(Ordering::Acquire), + route_id: 0, + speaker_pubkey: None, + speaker_generation: 0, + voice_reference: None, + 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) -> TtsTextSender { + TtsTextSender { + text_tx: self.text_tx.clone(), + generation: self.voice_generation.load(Ordering::Acquire), + speaker_generations: Arc::clone(&self.speaker_generations), + } + } + + /// Invalidate speech queued for one agent and cancel the player only when + /// that same agent currently owns it. + pub(crate) fn cancel_speaker(&self, speaker_pubkey: &str) { + request_speaker_cancel( + &self.speaker_generations, + &self.active_speaker, + &self.speaker_cancel, + speaker_pubkey, + ); + } + + /// Cancel exactly the speaker utterance currently owning playback. + /// + /// The speaker generation is advanced while ownership is locked, so a + /// stale Stop click cannot cancel a later utterance that starts after the + /// observed one drains. + pub(crate) fn cancel_active_speaker(&self, expected_speaker_pubkey: &str) -> bool { + request_active_speaker_cancel( + &self.speaker_generations, + &self.active_speaker, + &self.speaker_cancel, + &self.playback_probe, + expected_speaker_pubkey, + ) + } + + /// Select a bundled Pocket voice for subsequent speech. + /// + /// 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) -> Option> { + let acknowledged = begin_voice_change( + &self.voice, + &self.voice_generation, + &self.voice_cancel, + &self.voice_change_ack, + voice, + ); + if acknowledged.is_some() { + eprintln!("buzz-desktop: tts stage=cancellation reason=voice_switch route_id=0"); + } + acknowledged + } + + /// 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(); + } + + /// Signal the worker thread to stop. + pub fn shutdown(&self) { + eprintln!("buzz-desktop: tts stage=cancellation reason=shutdown route_id=0"); + self.shutdown.store(true, Ordering::Release); + } + + /// Returns `true` if the worker thread has exited (init failure, crash, or normal exit). + /// Used by hot-start to detect dead pipelines and clear them for retry. + pub fn is_finished(&self) -> bool { + self.thread.as_ref().is_none_or(|h| h.is_finished()) + } +} diff --git a/desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs b/desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs new file mode 100644 index 0000000000..4b9c2824f7 --- /dev/null +++ b/desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs @@ -0,0 +1,177 @@ +use super::*; + +pub(super) struct TtsMonitorState { + pub(super) player: Arc, + pub(super) cancel: Arc, + pub(super) voice_cancel: Arc, + pub(super) tts_active: Arc, + pub(super) stop: Arc, + pub(super) player_ops: Arc>, + pub(super) activity_frames: Arc>>, + pub(super) active_speaker: ActiveSpeaker, + pub(super) speaker_cancel: SpeakerCancellation, + pub(super) activity_app: Option, +} + +pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result> { + thread::Builder::new() + .name("tts-barge-in-monitor".into()) + .spawn(move || { + let mut last_activity_pubkey: Option = None; + let mut next_activity_tick = Instant::now(); + while !state.stop.load(Ordering::Acquire) { + if state.cancel.load(Ordering::Acquire) + || state.voice_cancel.load(Ordering::Acquire) + { + let _ops = lock_player_ops(&state.player_ops); + if state.cancel.load(Ordering::Acquire) + || state.voice_cancel.load(Ordering::Acquire) + { + state.player.clear(); + state.player.play(); + state + .active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take(); + state.tts_active.store(false, Ordering::Release); + } + } + silence_cancelled_speaker( + &state.speaker_cancel, + &state.active_speaker, + &state.player, + &state.player_ops, + &state.tts_active, + ); + if let Some(ref app) = state.activity_app { + if state.tts_active.load(Ordering::Acquire) { + let now = Instant::now(); + if now >= next_activity_tick { + let frame = state + .activity_frames + .lock() + .unwrap_or_else(|error| error.into_inner()) + .pop_front(); + if let Some(frame) = frame { + use tauri::Emitter; + let _ = app.emit( + "huddle-tts-speaker-level", + TtsSpeakerActivityPayload { + pubkey: Some(frame.pubkey.clone()), + level: frame.level, + }, + ); + last_activity_pubkey = Some(frame.pubkey); + } + next_activity_tick = now + SPEAKER_ACTIVITY_TICK; + } + } else { + let had_activity = last_activity_pubkey.take().is_some(); + state + .activity_frames + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clear(); + if had_activity { + use tauri::Emitter; + let _ = app.emit( + "huddle-tts-speaker-level", + TtsSpeakerActivityPayload { + pubkey: None, + level: 0.0, + }, + ); + } + next_activity_tick = Instant::now(); + } + } + thread::sleep(MONITOR_TICK); + } + }) +} + +pub(super) fn silence_cancelled_speaker( + cancellation: &SpeakerCancellation, + active_speaker: &ActiveSpeaker, + player: &rodio::Player, + player_ops: &Mutex<()>, + tts_active: &AtomicBool, +) { + let Some(cancelled) = cancellation + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone() + else { + return; + }; + let _ops = lock_player_ops(player_ops); + if take_cancelled_active_speaker(&cancelled, active_speaker) { + player.clear(); + player.play(); + tts_active.store(false, Ordering::Release); + } +} + +fn take_cancelled_active_speaker(cancelled: &str, active_speaker: &ActiveSpeaker) -> bool { + let mut active = active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()); + if !active + .as_deref() + .is_some_and(|speaker| speaker.eq_ignore_ascii_case(cancelled)) + { + return false; + } + active.take(); + true +} + +pub(super) fn consume_speaker_cancel( + cancellation: &SpeakerCancellation, + active_speaker: &ActiveSpeaker, + generations: &SpeakerGenerations, + tts_active: &AtomicBool, + text_state: CancelTextState<'_>, + player: Option<(&rodio::Player, &Mutex<()>)>, +) -> bool { + let Some(cancelled) = cancellation + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take() + else { + return false; + }; + let (text_rx, deferred_text, current_text) = text_state; + retain_current_speaker_text(generations, deferred_text, current_text, text_rx); + let mut cleared_player = false; + if let Some((player, player_ops)) = player { + let _ops = lock_player_ops(player_ops); + if take_cancelled_active_speaker(&cancelled, active_speaker) { + player.clear(); + player.play(); + tts_active.store(false, Ordering::Release); + cleared_player = true; + } + } + // The monitor may already have cleared the cancelled speaker while the + // worker was blocked. If another speaker has since claimed the player, + // preserve that speaker's activity flag and lead-in state. + cleared_player +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stale_targeted_cancel_does_not_release_the_next_speaker() { + let active_speaker = Arc::new(Mutex::new(Some("bob".to_string()))); + + assert!(!take_cancelled_active_speaker("alice", &active_speaker)); + assert_eq!( + active_speaker.lock().expect("active speaker").as_deref(), + Some("bob") + ); + } +} 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 044b1acf1e..bff5ab4f76 100644 --- a/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs +++ b/desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs @@ -19,6 +19,10 @@ fn inert_pipeline(cancel: Arc) -> TtsPipeline { voice_cancel: Arc::new(AtomicBool::new(false)), voice: Arc::new(std::sync::Mutex::new("reference_sample".to_string())), voice_generation: Arc::new(AtomicU64::new(1)), + speaker_generations: Arc::new(std::sync::Mutex::new(HashMap::new())), + active_speaker: Arc::new(std::sync::Mutex::new(None)), + speaker_cancel: Arc::new(std::sync::Mutex::new(None)), + playback_probe: PlaybackProbe::new(), voice_change_ack: Arc::new(std::sync::Mutex::new(None)), thread: Some(thread), } @@ -168,6 +172,7 @@ fn an_in_hand_post_change_message_survives_cancellation() { generation: voice_generation.load(Ordering::Acquire), route_id: 1, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "new message".to_string(), }) @@ -181,6 +186,7 @@ fn an_in_hand_post_change_message_survives_cancellation() { generation: 1, route_id: 2, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "old message".to_string(), }, @@ -188,6 +194,7 @@ fn an_in_hand_post_change_message_survives_cancellation() { generation: voice_generation.load(Ordering::Acquire), route_id: 3, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "later new message".to_string(), }, @@ -246,6 +253,7 @@ fn superseding_voice_change_removes_earlier_deferred_messages() { generation: voice_generation.load(Ordering::Acquire), route_id: 4, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "message for Eve".to_string(), }); @@ -294,6 +302,7 @@ fn barge_in_clears_deferred_voice_change_messages() { generation: 2, route_id: 5, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "deferred message".to_string(), }]); @@ -337,6 +346,7 @@ fn barge_in_during_a_voice_change_clears_post_change_messages() { generation: voice_generation.load(Ordering::Acquire), route_id: 6, speaker_pubkey: None, + speaker_generation: 0, voice_reference: None, text: "post-change message".to_string(), }); @@ -365,6 +375,7 @@ fn a_sender_captured_before_voice_change_is_stale_even_if_it_sends_after_drain() let old_sender = TtsTextSender { text_tx, generation: voice_generation.load(Ordering::Acquire), + speaker_generations: Arc::new(std::sync::Mutex::new(HashMap::new())), }; let shutdown = AtomicBool::new(false); let active = AtomicBool::new(true); @@ -392,6 +403,7 @@ fn a_sender_captured_before_voice_change_is_stale_even_if_it_sends_after_drain() .send( 7, "agent".to_string(), + 0, "reference_sample".to_string(), "late old message".to_string(), ) diff --git a/desktop/src-tauri/src/huddle/tts_voice_transition.rs b/desktop/src-tauri/src/huddle/tts_voice_transition.rs index 3a65553756..a60d3506ff 100644 --- a/desktop/src-tauri/src/huddle/tts_voice_transition.rs +++ b/desktop/src-tauri/src/huddle/tts_voice_transition.rs @@ -1,5 +1,6 @@ use std::{ collections::{HashMap, VecDeque}, + fmt, path::Path, sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, @@ -19,6 +20,9 @@ pub(super) struct PendingVoiceChange { pub(super) type VoiceChangeAck = Arc>>; pub(super) type WorkerVoiceState = (Arc>, Arc, VoiceChangeAck); pub(super) type WorkerCancelSignals = (Arc, Arc); +pub(super) type SpeakerGenerations = Arc>>; +pub(super) type ActiveSpeaker = Arc>>; +pub(super) type SpeakerCancellation = Arc>>; pub(super) type CancelTextState<'a> = ( &'a mpsc::Receiver, &'a mut VecDeque, @@ -26,11 +30,73 @@ pub(super) type CancelTextState<'a> = ( ); pub(super) type CancelSignals<'a> = (&'a AtomicBool, &'a AtomicBool); +#[derive(Clone)] +pub(super) struct PlaybackProbe { + player: Arc>>>, + pub(super) player_ops: Arc>, + synthesis_in_flight: Arc, +} + +pub(super) struct SynthesisFlightGuard { + playback_probe: PlaybackProbe, +} + +impl Drop for SynthesisFlightGuard { + fn drop(&mut self) { + self.playback_probe.set_synthesis_in_flight(false); + } +} + +impl PlaybackProbe { + pub(super) fn new() -> Self { + Self { + player: Arc::new(Mutex::new(None)), + player_ops: Arc::new(Mutex::new(())), + synthesis_in_flight: Arc::new(AtomicBool::new(false)), + } + } + + pub(super) fn install(&self, player: Arc) { + self.player + .lock() + .unwrap_or_else(|error| error.into_inner()) + .replace(player); + } + + pub(super) fn set_synthesis_in_flight(&self, in_flight: bool) { + let _ops = super::lock_player_ops(&self.player_ops); + self.synthesis_in_flight.store(in_flight, Ordering::Release); + } + + pub(super) fn begin_synthesis(&self) -> SynthesisFlightGuard { + self.set_synthesis_in_flight(true); + SynthesisFlightGuard { + playback_probe: self.clone(), + } + } + + fn player(&self) -> Option> { + self.player + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone() + } +} + +impl fmt::Debug for PlaybackProbe { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PlaybackProbe") + .finish_non_exhaustive() + } +} + #[derive(Debug)] pub(super) struct QueuedText { pub(super) generation: u64, pub(super) route_id: u64, pub(super) speaker_pubkey: Option, + pub(super) speaker_generation: u64, pub(super) voice_reference: Option, pub(super) text: String, } @@ -39,6 +105,7 @@ pub(super) struct QueuedText { pub(crate) struct TtsTextSender { pub(super) text_tx: SyncSender, pub(super) generation: u64, + pub(super) speaker_generations: SpeakerGenerations, } impl TtsTextSender { @@ -46,6 +113,7 @@ impl TtsTextSender { &self, route_id: u64, speaker_pubkey: String, + speaker_generation: u64, voice_reference: String, text: String, ) -> Result<(), String> { @@ -54,11 +122,156 @@ impl TtsTextSender { generation: self.generation, route_id, speaker_pubkey: Some(speaker_pubkey), + speaker_generation, voice_reference: Some(voice_reference), text, }) .map_err(|error| error.to_string()) } + + pub(crate) fn speaker_generation(&self, speaker_pubkey: &str) -> u64 { + current_speaker_generation(&self.speaker_generations, speaker_pubkey) + } +} + +pub(super) fn current_speaker_generation( + generations: &SpeakerGenerations, + speaker_pubkey: &str, +) -> u64 { + generations + .lock() + .unwrap_or_else(|error| error.into_inner()) + .get(&speaker_pubkey.to_ascii_lowercase()) + .copied() + .unwrap_or(0) +} + +pub(super) fn advance_speaker_generation( + generations: &SpeakerGenerations, + speaker_pubkey: &str, +) -> u64 { + let mut generations = generations + .lock() + .unwrap_or_else(|error| error.into_inner()); + let generation = generations + .entry(speaker_pubkey.to_ascii_lowercase()) + .or_default(); + *generation = generation.saturating_add(1); + *generation +} + +pub(super) fn queued_speaker_is_current( + generations: &SpeakerGenerations, + queued: &QueuedText, +) -> bool { + queued + .speaker_pubkey + .as_deref() + .is_none_or(|speaker_pubkey| { + current_speaker_generation(generations, speaker_pubkey) == queued.speaker_generation + }) +} + +pub(super) fn request_speaker_cancel( + generations: &SpeakerGenerations, + active_speaker: &ActiveSpeaker, + cancellation: &SpeakerCancellation, + speaker_pubkey: &str, +) { + advance_speaker_generation(generations, speaker_pubkey); + let owns_player = active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_deref() + .is_some_and(|active| active.eq_ignore_ascii_case(speaker_pubkey)); + if owns_player { + cancellation + .lock() + .unwrap_or_else(|error| error.into_inner()) + .replace(speaker_pubkey.to_ascii_lowercase()); + } +} + +pub(super) fn request_active_speaker_cancel( + generations: &SpeakerGenerations, + active_speaker: &ActiveSpeaker, + cancellation: &SpeakerCancellation, + playback_probe: &PlaybackProbe, + expected_speaker_pubkey: &str, +) -> bool { + let Some(player) = playback_probe.player() else { + return false; + }; + let _ops = super::lock_player_ops(&playback_probe.player_ops); + let playback_live = + !player.empty() || playback_probe.synthesis_in_flight.load(Ordering::Acquire); + request_active_speaker_cancel_while_locked( + generations, + active_speaker, + cancellation, + playback_live, + expected_speaker_pubkey, + ) +} + +fn request_active_speaker_cancel_while_locked( + generations: &SpeakerGenerations, + active_speaker: &ActiveSpeaker, + cancellation: &SpeakerCancellation, + playback_live: bool, + expected_speaker_pubkey: &str, +) -> bool { + if !playback_live { + return false; + } + let active = active_speaker + .lock() + .unwrap_or_else(|error| error.into_inner()); + let Some(speaker_pubkey) = active.as_deref() else { + return false; + }; + if !speaker_pubkey.eq_ignore_ascii_case(expected_speaker_pubkey) { + return false; + } + + // Keep ownership locked until the generation and cancellation request are + // committed. The drain path takes the same lock, so the request is bound + // to the utterance the Stop action actually observed. + let mut cancellation = cancellation + .lock() + .unwrap_or_else(|error| error.into_inner()); + if cancellation + .as_deref() + .is_some_and(|pending| pending.eq_ignore_ascii_case(speaker_pubkey)) + { + return false; + } + advance_speaker_generation(generations, speaker_pubkey); + cancellation.replace(speaker_pubkey.to_ascii_lowercase()); + true +} + +pub(super) fn retain_current_speaker_text( + generations: &SpeakerGenerations, + deferred_text: &mut VecDeque, + current_text: &mut Option, + text_rx: &mpsc::Receiver, +) { + deferred_text.retain(|text| queued_speaker_is_current(generations, text)); + if let Some(text) = current_text.take() { + if queued_speaker_is_current(generations, &text) { + deferred_text.push_front(text); + } else { + log_cancelled_route(text.route_id, "speaker_removed"); + } + } + while let Ok(text) = text_rx.try_recv() { + if queued_speaker_is_current(generations, &text) { + deferred_text.push_back(text); + } else { + log_cancelled_route(text.route_id, "speaker_removed"); + } + } } pub(super) fn has_pending_voice_change(voice_change_ack: &VoiceChangeAck) -> bool { @@ -258,3 +471,227 @@ pub(super) fn retain_cancelled_text( fn log_cancelled_route(route_id: u64, reason: &str) { eprintln!("buzz-desktop: tts stage=queue status=dropped reason={reason} route_id={route_id}"); } + +#[cfg(test)] +mod speaker_generation_tests { + use super::*; + + fn playback_probe(playback_live: bool) -> PlaybackProbe { + let channels = std::num::NonZero::new(1).expect("non-zero channels"); + let sample_rate = std::num::NonZero::new(24_000).expect("non-zero sample rate"); + let (mixer, _mixer_source) = rodio::mixer::mixer(channels, sample_rate); + let player = Arc::new(rodio::Player::connect_new(&mixer)); + if playback_live { + player.append(rodio::buffer::SamplesBuffer::new( + channels, + sample_rate, + vec![0.0; 24_000], + )); + } + let probe = PlaybackProbe::new(); + probe.install(player); + probe + } + + fn queued_speech(speaker_pubkey: &str, speaker_generation: u64) -> QueuedText { + QueuedText { + generation: 1, + route_id: 1, + speaker_pubkey: Some(speaker_pubkey.to_string()), + speaker_generation, + voice_reference: Some("pocket:mary".to_string()), + text: "Hello".to_string(), + } + } + + #[test] + fn removing_a_speaker_invalidates_only_that_speakers_queued_text() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let alice = queued_speech("ALICE", current_speaker_generation(&generations, "alice")); + let bob = queued_speech("bob", current_speaker_generation(&generations, "bob")); + + advance_speaker_generation(&generations, "alice"); + + assert!(!queued_speaker_is_current(&generations, &alice)); + assert!(queued_speaker_is_current(&generations, &bob)); + + let rejoined_alice = + queued_speech("alice", current_speaker_generation(&generations, "alice")); + assert!(queued_speaker_is_current(&generations, &rejoined_alice)); + } + + #[test] + fn removing_a_silent_speaker_does_not_cancel_the_active_speaker() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + + request_speaker_cancel(&generations, &active_speaker, &cancellation, "bob"); + + assert!(cancellation.lock().expect("cancellation").is_none()); + assert_eq!( + active_speaker.lock().expect("active speaker").as_deref(), + Some("alice") + ); + } + + #[test] + fn targeted_cancellation_preserves_other_speakers_queue_entries() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + let alice = queued_speech("alice", 0); + let bob = queued_speech("bob", 0); + let (_text_tx, text_rx) = mpsc::sync_channel(1); + let mut deferred = VecDeque::from([alice, bob]); + let mut current = None; + + request_speaker_cancel(&generations, &active_speaker, &cancellation, "alice"); + retain_current_speaker_text(&generations, &mut deferred, &mut current, &text_rx); + + assert_eq!(deferred.len(), 1); + assert_eq!(deferred[0].speaker_pubkey.as_deref(), Some("bob")); + } + + #[test] + fn stop_request_is_bound_to_the_observed_speaker_generation() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + + assert!(request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &playback_probe(true), + "alice", + )); + assert_eq!(current_speaker_generation(&generations, "alice"), 1); + assert_eq!( + cancellation.lock().expect("cancellation").as_deref(), + Some("alice") + ); + + active_speaker.lock().expect("active speaker").take(); + cancellation.lock().expect("cancellation").take(); + assert!(!request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &playback_probe(true), + "alice", + )); + + let next_utterance = + queued_speech("alice", current_speaker_generation(&generations, "alice")); + assert!(queued_speaker_is_current(&generations, &next_utterance)); + assert!(cancellation.lock().expect("cancellation").is_none()); + } + + #[test] + fn stop_request_does_not_cancel_a_different_active_speaker() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("bob".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + + assert!(!request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &playback_probe(true), + "alice", + )); + assert_eq!(current_speaker_generation(&generations, "alice"), 0); + assert_eq!(current_speaker_generation(&generations, "bob"), 0); + assert!(cancellation.lock().expect("cancellation").is_none()); + assert_eq!( + active_speaker.lock().expect("active speaker").as_deref(), + Some("bob"), + ); + } + + #[test] + fn stop_request_during_empty_synthesis_gap_cancels_in_flight_speech() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + let next_chunk = queued_speech("alice", 0); + let probe = playback_probe(false); + let _synthesis_flight = probe.begin_synthesis(); + + assert!(request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &probe, + "alice", + )); + + assert_eq!(current_speaker_generation(&generations, "alice"), 1); + assert!(!queued_speaker_is_current(&generations, &next_chunk)); + assert_eq!( + cancellation.lock().expect("cancellation").as_deref(), + Some("alice"), + ); + } + + #[test] + fn repeated_stop_for_same_in_flight_utterance_is_idempotent() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + let probe = playback_probe(false); + let _synthesis_flight = probe.begin_synthesis(); + + assert!(request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &probe, + "alice", + )); + let speech_queued_after_first_stop = queued_speech("alice", 1); + + assert!(!request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &probe, + "alice", + )); + + assert_eq!(current_speaker_generation(&generations, "alice"), 1); + assert!(queued_speaker_is_current( + &generations, + &speech_queued_after_first_stop, + )); + assert_eq!( + cancellation.lock().expect("cancellation").as_deref(), + Some("alice"), + ); + } + + #[test] + fn stop_request_after_playback_drains_preserves_queued_speech() { + let generations = Arc::new(Mutex::new(HashMap::new())); + let active_speaker = Arc::new(Mutex::new(Some("alice".to_string()))); + let cancellation = Arc::new(Mutex::new(None)); + let next_utterance = queued_speech("alice", 0); + + assert!(!request_active_speaker_cancel( + &generations, + &active_speaker, + &cancellation, + &playback_probe(false), + "alice", + )); + + assert_eq!(current_speaker_generation(&generations, "alice"), 0); + assert!(queued_speaker_is_current(&generations, &next_utterance)); + assert!(cancellation.lock().expect("cancellation").is_none()); + assert_eq!( + active_speaker.lock().expect("active speaker").as_deref(), + Some("alice"), + ); + } +} diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index d59936946f..1e73b15232 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -56,9 +56,10 @@ use huddle::reconnect::reconnect_huddle_audio; use huddle::{ add_agent_to_huddle, check_pipeline_hotstart, close_huddle_companion, confirm_huddle_active, download_voice_models, end_huddle, get_huddle_agent_pubkeys, get_huddle_state, - get_model_status, get_voice_input_mode, join_huddle, leave_huddle, open_huddle_window, - push_audio_pcm, set_huddle_transcription_enabled, set_tts_enabled, set_voice_input_mode, - speak_agent_message, start_huddle, start_stt_pipeline, HuddlePhase, + get_model_status, get_voice_input_mode, interrupt_huddle_speech, join_huddle, leave_huddle, + open_huddle_window, push_audio_pcm, remove_agent_from_huddle, set_huddle_manual_mic_unmuted, + set_huddle_transcription_enabled, set_tts_enabled, set_voice_input_mode, speak_agent_message, + start_huddle, start_stt_pipeline, HuddlePhase, }; use initial_window::*; use managed_agents::{ @@ -851,7 +852,9 @@ pub fn run() { huddle::agent_voice::set_huddle_agent_tts_enabled, huddle::agent_voice::set_huddle_agent_voice, speak_agent_message, + interrupt_huddle_speech, add_agent_to_huddle, + remove_agent_from_huddle, huddle::agents::sync_agents_to_active_huddle, check_pipeline_hotstart, confirm_huddle_active, @@ -859,6 +862,7 @@ pub fn run() { get_huddle_agent_pubkeys, set_voice_input_mode, get_voice_input_mode, + set_huddle_manual_mic_unmuted, list_audio_output_devices, set_audio_output_device, get_audio_output_device, diff --git a/desktop/src/app/AppHuddleShell.tsx b/desktop/src/app/AppHuddleShell.tsx index 8370e8efd4..29dcd26cdf 100644 --- a/desktop/src/app/AppHuddleShell.tsx +++ b/desktop/src/app/AppHuddleShell.tsx @@ -48,6 +48,13 @@ export function AppHuddleShell({ data-huddle-open={isDrawerOpen} data-huddle-window={isRoom} > +