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