//! Huddle (voice) state machine and Tauri commands. //! //! Mental model: //! parent channel → start_huddle → ephemeral channel + audio WS relay //! other clients → join_huddle → audio WS relay //! any client → leave_huddle → lifecycle event, clear local state //! creator → end_huddle → archive ephemeral channel, clear state //! //! HuddleState is stored in AppState and serialized for get_huddle_state. //! //! ## Synchronization Protocol //! //! `HuddleState` lives behind a single `Mutex` in `AppState`. Rules: //! //! 1. **Never hold the outer lock across `.await`** — acquire, read/write, release. //! 2. **Pipeline construction happens outside the lock** — the `stt_starting` / //! `tts_starting` sentinels prevent TOCTOU races during the ~200ms window. //! 3. **`agent_pubkeys` has its own inner `Arc`** — the transcription task //! clones the `Arc` and reads at post time without the outer lock. //! 4. **Atomics for cross-thread signaling** — `tts_active`, `tts_cancel`, //! `ptt_active`, `session_generation` are shared with pipeline worker threads. //! 5. **Pipeline teardown extracts handles before dropping** — `teardown_huddle` //! takes `stt_pipeline`/`tts_pipeline` out of the lock, then calls `shutdown()` //! and drops them outside the lock (thread joins can block ~200ms). mod agent_tts_publisher; mod agent_tts_routing; pub mod agent_voice; pub mod agents; pub mod audio_output; mod commands; mod human_floor; pub mod jitter; #[cfg(test)] mod latency_bench; mod local_barge_in; pub mod models; pub mod pipeline; pub mod playout; pub mod pocket; pub mod preprocessing; pub mod reconnect; pub mod relay_api; pub mod state; pub mod stt; pub mod transcription; pub mod tts; #[path = "tts_playback.rs"] mod tts_playback; pub mod tts_settings; mod tts_voice_import; mod tts_voice_registry; mod window; pub mod wire; // ── Shared utilities ────────────────────────────────────────────────────────── /// Drain and discard all pending messages until shutdown or disconnect. /// Shared by both the STT and TTS worker threads for graceful degradation /// when model files are missing or initialization fails. pub(super) fn drain_until_shutdown( rx: std::sync::mpsc::Receiver, shutdown: &std::sync::atomic::AtomicBool, ) { loop { if shutdown.load(std::sync::atomic::Ordering::Acquire) { break; } match rx.recv_timeout(std::time::Duration::from_millis(100)) { Ok(_) => continue, Err(_) => break, } } } // ── Re-exports ──────────────────────────────────────────────────────────────── pub use commands::{ add_agent_to_huddle, 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; pub use window::{close_huddle_companion, open_huddle_window}; // ── Imports ─────────────────────────────────────────────────────────────────── use std::sync::{atomic::Ordering, Arc}; use tauri::State; use uuid::Uuid; use crate::{app_state::AppState, events, relay::submit_event}; use agent_tts_routing::{ classify_agent_tts_runtime, enqueue_agent_tts_text, normalize_agent_tts_text, AgentTtsRuntimeGate, }; pub use pipeline::check_pipeline_hotstart; use pipeline::{ await_inflight_tts_start, maybe_start_stt_pipeline, maybe_start_tts_pipeline, post_connect_setup, PostConnectOutcome, }; use relay_api::{ count_human_members, fetch_channel_members, parse_channel_uuid, validate_pubkey_hex, MAX_HUDDLE_AGENTS, }; use window::close_huddle_window; fn normalize_huddle_channel_name(candidate: Option, fallback: &str) -> String { let normalized = candidate .unwrap_or_default() .split_whitespace() .collect::>() .join(" "); let name = if normalized.is_empty() { fallback } else { normalized.as_str() }; name.chars().take(80).collect() } // ── Tauri commands ──────────────────────────────────────────────────────────── /// Set the voice input mode (push-to-talk or voice-activity detection). /// /// When switching mid-huddle, restarts the STT pipeline so it picks up the /// new mode (PTT gating vs continuous VAD with barge-in). The pipeline /// captures the mode at construction time, so a restart is required. #[tauri::command] pub async fn set_voice_input_mode( mode: VoiceInputMode, state: State<'_, AppState>, ) -> Result<(), String> { let needs_restart = { let mut hs = state.huddle()?; let old_mode = hs.voice_input_mode.clone(); hs.voice_input_mode = mode.clone(); // Restart STT if mode changed and a huddle is active with a pipeline running. old_mode != mode && matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) && hs.stt_pipeline.is_some() && hs.transcription_enabled }; if needs_restart { let eph_id = { let hs = state.huddle()?; hs.ephemeral_channel_id.clone() }; if let Some(eph_id) = eph_id { // Best-effort restart — if models aren't ready, the pipeline // stays down until the next hotstart cycle picks it up. if let Err(e) = maybe_start_stt_pipeline(&state, &eph_id).await { eprintln!("buzz-desktop: STT pipeline restart on mode switch failed: {e}"); } } } // A mode switch changes whether the global shortcut should be reserved but // does not otherwise emit; emit here so the frontend sees the new mode and // the shortcut registration is re-synced. state.emit_huddle_state_changed(); Ok(()) } /// Return the current voice input mode. #[tauri::command] pub fn get_voice_input_mode(state: State<'_, AppState>) -> Result { let hs = state.huddle()?; Ok(hs.voice_input_mode.clone()) } /// Start a new huddle in the given parent channel. /// /// Steps: /// 1. Create an ephemeral channel (kind 9007, ttl=3600). /// 2. Post voice-mode guidelines (kind 48106). /// 3. Add each invited member to the ephemeral channel (kind 9000). /// 4. Emit KIND_HUDDLE_STARTED to the parent channel (kind 48100). /// 5. Store state and return join info. /// /// If ANY step fails (including channel creation), the orphaned ephemeral /// channel is archived (best-effort) and state is reset to Idle. #[tauri::command] pub async fn start_huddle( parent_channel_id: String, member_pubkeys: Vec, channel_name: Option, app: tauri::AppHandle, state: State<'_, AppState>, ) -> Result { // Validate inputs at the Tauri boundary. if member_pubkeys.len() > MAX_HUDDLE_AGENTS { return Err(format!( "too many agents: {} (max {})", member_pubkeys.len(), MAX_HUDDLE_AGENTS )); } // Dedup and validate pubkey format. let member_pubkeys: Vec = { let mut seen = std::collections::HashSet::new(); let mut deduped = Vec::new(); for pk in member_pubkeys { validate_pubkey_hex(&pk)?; if seen.insert(pk.clone()) { deduped.push(pk); } } deduped }; // Allocate the backing channel ID before the relay work starts. Publishing // it with the Creating state lets the main webview open an immediate // companion window while the channel and audio session are being prepared. let ephemeral_uuid = Uuid::new_v4(); let ephemeral_channel_id = ephemeral_uuid.to_string(); let short_id = &ephemeral_channel_id[..8]; let fallback_channel_name = format!("huddle-{short_id}"); let channel_name = normalize_huddle_channel_name(channel_name, &fallback_channel_name); // Transition to Creating. let huddle_generation = { let mut hs = state.huddle()?; if hs.phase != HuddlePhase::Idle { return Err(format!( "cannot start huddle: already in phase {:?}", hs.phase )); } let generation = hs.begin_huddle_lifetime(); hs.phase = HuddlePhase::Creating; hs.parent_channel_id = Some(parent_channel_id.clone()); hs.ephemeral_channel_id = Some(ephemeral_channel_id.clone()); generation }; state.emit_huddle_state_changed(); // All steps wrapped so we can roll back on ANY failure, including step 1. // channel_was_created tracks whether we need to archive on rollback. let mut channel_was_created = false; let result: Result<(Vec, String), String> = async { // 1. Create ephemeral channel. let create_builder = events::build_create_channel( ephemeral_uuid, &channel_name, "private", "stream", None, Some(3600), )?; submit_event(create_builder, &state).await?; channel_was_created = true; // 2. Post voice-mode guidelines as kind:48106 BEFORE adding agents. // Agents auto-subscribe on membership notification (kind:9000) and may // complete EOSE before guidelines are stored if we post them after. // Best-effort: don't fail the huddle if this fails. let guidelines = agents::voice_mode_guidelines(&parent_channel_id); if let Ok(guidelines_builder) = events::build_huddle_guidelines(&ephemeral_channel_id, &guidelines) { if let Err(e) = submit_event(guidelines_builder, &state).await { eprintln!("buzz-desktop: huddle guidelines (kind:48106) failed: {e}"); } } // 3. Add members to the ephemeral channel; only keep successfully enrolled ones. let mut successful_agents: Vec = Vec::new(); for pubkey in &member_pubkeys { let add_builder = events::build_add_member(ephemeral_uuid, pubkey, Some("bot"))?; match submit_event(add_builder, &state).await { Ok(_) => successful_agents.push(pubkey.clone()), Err(e) => { eprintln!("buzz-desktop: huddle add_member failed for {pubkey}: {e}"); // Intentionally not added — policy rejected this agent. } } } // 4. Emit HUDDLE_STARTED to parent channel. let started_builder = events::build_huddle_started(&parent_channel_id, &ephemeral_channel_id)?; let started_event = submit_event(started_builder, &state).await?; Ok((successful_agents, started_event.event_id)) } .await; match result { Ok((successful_agents, huddle_thread_event_id)) => { // 5. Store active state. let committed = { let mut hs = state.huddle()?; if !hs.owns_huddle_lifetime(huddle_generation, HuddlePhase::Creating) { false } else { hs.phase = HuddlePhase::Connected; hs.is_creator = true; hs.ephemeral_channel_id = Some(ephemeral_channel_id.clone()); hs.huddle_thread_event_id = Some(huddle_thread_event_id); *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = successful_agents.clone(); hs.maybe_auto_enable_transcription_for_agents(); let own_pubkey = state .keys .lock() .map(|k| k.public_key().to_hex()) .unwrap_or_default(); let mut participants = successful_agents.clone(); if !own_pubkey.is_empty() && !participants.contains(&own_pubkey) { participants.insert(0, own_pubkey); } hs.participants = participants; true } }; if !committed { emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state).await; close_huddle_window(&app, &ephemeral_channel_id); return Err("huddle start was superseded".to_owned()); } // 6. Notify frontend of state change. state.emit_huddle_state_changed(); // 7. Hydrate members, download models, start pipelines (incl. audio relay). // Audio relay failure is fatal — no point in a huddle without audio. match post_connect_setup(&state, &ephemeral_channel_id, huddle_generation).await { Ok(PostConnectOutcome::Ready) => {} Ok(PostConnectOutcome::Stale) => { close_huddle_window(&app, &ephemeral_channel_id); return Err("huddle start was superseded".to_owned()); } Err(e) => { // Roll back only if this failed setup still owns the active // huddle. A stale failure must not tear down its replacement. let still_current = state .huddle() .map(|hs| hs.is_current_huddle(&ephemeral_channel_id, huddle_generation)) .unwrap_or(false); if still_current { emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state) .await; if let Ok(mut hs) = state.huddle_state.lock() { if hs.is_current_huddle(&ephemeral_channel_id, huddle_generation) { hs.reset_preserving_generation(); } } state.emit_huddle_state_changed(); } close_huddle_window(&app, &ephemeral_channel_id); return Err(e); } } Ok(HuddleJoinInfo { ephemeral_channel_id, }) } Err(e) => { // Rollback: archive the orphaned ephemeral channel if it was created. if channel_was_created { if let Ok(archive_builder) = events::build_archive(ephemeral_uuid) { if let Err(ae) = submit_event(archive_builder, &state).await { eprintln!( "buzz-desktop: rollback archive of {ephemeral_channel_id} failed: {ae}" ); } } } // Reset only if this failed attempt still owns the Creating state. let reset = if let Ok(mut hs) = state.huddle_state.lock() { if hs.owns_huddle_lifetime(huddle_generation, HuddlePhase::Creating) { hs.reset_preserving_generation(); true } else { false } } else { false }; if reset { state.emit_huddle_state_changed(); close_huddle_window(&app, &ephemeral_channel_id); } Err(e) } } } /// Join an existing huddle in the given parent channel. /// /// Steps: /// 1. Transition to Connecting. /// 2. Store state and return join info. /// 3. Post-connect setup (audio relay WS, pipelines, model hydration). /// /// The relay emits kind:48101 (participant joined) when the audio WS authenticates. #[tauri::command] pub async fn join_huddle( parent_channel_id: String, ephemeral_channel_id: String, huddle_thread_event_id: Option, state: State<'_, AppState>, ) -> Result { // Transition to Connecting. let huddle_generation = { let mut hs = state.huddle()?; if hs.phase != HuddlePhase::Idle { return Err(format!( "cannot join huddle: already in phase {:?}", hs.phase )); } let generation = hs.begin_huddle_lifetime(); hs.phase = HuddlePhase::Connecting; hs.parent_channel_id = Some(parent_channel_id.clone()); hs.ephemeral_channel_id = Some(ephemeral_channel_id.clone()); hs.huddle_thread_event_id = huddle_thread_event_id; generation }; // Seed participant list with own pubkey as a fallback until relay responds. let own_pubkey = state .keys .lock() .map(|k| k.public_key().to_hex()) .unwrap_or_default(); let committed = { let mut hs = state.huddle()?; if !hs.owns_huddle_lifetime(huddle_generation, HuddlePhase::Connecting) { false } else { hs.phase = HuddlePhase::Connected; if !own_pubkey.is_empty() { hs.participants = vec![own_pubkey]; } true } }; if !committed { return Err("huddle join was superseded".to_owned()); } // Notify frontend of state change. state.emit_huddle_state_changed(); // Hydrate members, download models, start pipelines (incl. audio relay). // Audio relay failure is fatal — no point in a huddle without audio. match post_connect_setup(&state, &ephemeral_channel_id, huddle_generation).await { Ok(PostConnectOutcome::Ready) => {} Ok(PostConnectOutcome::Stale) => { return Err("huddle join was superseded".to_owned()); } Err(e) => { // Reset only the huddle lifetime that failed. let mut did_reset = false; if let Ok(mut hs) = state.huddle_state.lock() { if hs.is_current_huddle(&ephemeral_channel_id, huddle_generation) { hs.reset_preserving_generation(); did_reset = true; } } if did_reset { state.emit_huddle_state_changed(); } return Err(e); } } Ok(HuddleJoinInfo { ephemeral_channel_id, }) } /// Shut down all pipelines and reset huddle state to Idle. /// /// Used by both `leave_huddle` and `end_huddle` to avoid duplicating the /// shutdown-then-reset sequence. fn teardown_huddle(state: &AppState) -> Result<(), String> { // Take pipeline handles out of state and drop the lock before shutdown. // Pipeline Drop impls join worker threads — this avoids blocking while // the mutex is held (ONNX inference can take ~200ms). let (old_stt, old_tts, _audio_cancel) = { let mut hs = state.huddle()?; // Increment generation first — this immediately invalidates any // in-flight transcription task, even before pipelines shut down. hs.session_generation.fetch_add(1, Ordering::Release); let stt = hs.take_stt_pipeline(); let tts = hs.tts_pipeline.take(); let cancel = hs.audio_ws_cancel.take(); // Cancel the relay token BEFORE dropping the sender. If we drop // pcm_tx first, the send task sees None from recv() and can exit // the pipeline before is_cancelled() is true — causing a spurious // huddle-audio-disconnected event on intentional teardown. if let Some(ref c) = cancel { c.cancel(); } hs.audio_relay_pcm_tx.take(); // Drop sender — signals the relay task. hs.reset_preserving_generation(); (stt, tts, cancel) }; // Shut down STT/TTS outside the lock — thread joins happen here. if let Some(ref p) = old_stt { p.shutdown(); } if let Some(ref p) = old_tts { p.shutdown(); } // Drop the Arcs here (implicit) — triggers thread join via Drop. drop(old_stt); drop(old_tts); // Notify frontend that we're back to Idle. state.emit_huddle_state_changed(); Ok(()) } /// Emit HUDDLE_ENDED to the parent channel and archive the ephemeral channel. /// /// Both steps are best-effort — failures are logged but do not propagate. /// Called from `leave_huddle` (auto-end path) and `end_huddle`. async fn emit_end_and_archive( parent_channel_id: &str, ephemeral_channel_id: &str, state: &AppState, ) { if !parent_channel_id.is_empty() && !ephemeral_channel_id.is_empty() { if let Ok(ended_builder) = events::build_huddle_ended(parent_channel_id, ephemeral_channel_id) { if let Err(e) = submit_event(ended_builder, state).await { eprintln!("buzz-desktop: huddle_ended event failed: {e}"); } } } remove_huddle_agents(ephemeral_channel_id, state).await; if !ephemeral_channel_id.is_empty() { if let Ok(uuid) = parse_channel_uuid(ephemeral_channel_id) { if let Ok(archive_builder) = events::build_archive(uuid) { if let Err(e) = submit_event(archive_builder, state).await { eprintln!("buzz-desktop: archive ephemeral channel failed: {e}"); } } } } } /// Best-effort agent removal before archiving an ended huddle. /// /// Archive should make the ephemeral channel unavailable on its own, but removing /// bot-role members first prevents an agent-only huddle from lingering if the /// archive publish is delayed or rejected. async fn remove_huddle_agents(ephemeral_channel_id: &str, state: &AppState) { if ephemeral_channel_id.is_empty() { return; } let Ok(eph_uuid) = parse_channel_uuid(ephemeral_channel_id) else { return; }; let agent_pubkeys = match fetch_channel_members(ephemeral_channel_id, Some("bot"), state).await { Ok(pubkeys) => pubkeys, Err(e) => { eprintln!("buzz-desktop: fetch huddle agents for cleanup failed: {e}"); return; } }; for pubkey in agent_pubkeys { let Ok(remove_builder) = events::build_remove_member(eph_uuid, &pubkey) else { continue; }; if let Err(e) = submit_event(remove_builder, state).await { eprintln!("buzz-desktop: remove huddle agent {pubkey} failed: {e}"); } } } /// Leave the current huddle. /// /// Steps: /// 1. Transition to Leaving. /// 2. Auto-end check: if last human, emit HUDDLE_ENDED + archive. /// 3. Shut down pipelines and audio relay. /// /// The relay emits kind:48102 (participant left) when the audio WS disconnects. #[tauri::command] pub async fn leave_huddle(app: tauri::AppHandle, state: State<'_, AppState>) -> Result<(), String> { let (parent_channel_id, ephemeral_channel_id) = { let mut hs = state.huddle()?; if hs.phase == HuddlePhase::Idle { return Ok(()); // Nothing to leave. } hs.phase = HuddlePhase::Leaving; ( hs.parent_channel_id.clone().unwrap_or_default(), hs.ephemeral_channel_id.clone().unwrap_or_default(), ) }; // Auto-end: check if any human participants remain. If not, end the huddle // (emit HUDDLE_ENDED + archive). If others remain, just remove self from // membership so the participant roster stays accurate. // // We check BEFORE removing self — the relay counts us as a member until // we leave. So "1 human remaining" means WE are the last one. if !parent_channel_id.is_empty() && !ephemeral_channel_id.is_empty() { let humans_remaining = count_human_members(&ephemeral_channel_id, &state) .await // On fetch failure, assume 2 humans remain (safe default). // unwrap_or(1) would mean "I'm the last human" → triggers auto-archive, // ending the huddle for everyone on a transient REST failure. Using 2 // means we skip the auto-end path and just remove ourselves — the huddle // stays alive and the next real leave will clean up correctly. .unwrap_or(2); if humans_remaining <= 1 { // We're the last human — end the huddle entirely. // Archive subsumes leave (the channel is gone, membership is moot). // This avoids the "cannot remove the last owner" relay error that // build_leave hits when the creator is the sole remaining member. eprintln!("buzz-desktop: last human left huddle — auto-ending"); emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state).await; } else { // Other humans still in the huddle — just remove self from membership. if let Ok(eph_uuid) = parse_channel_uuid(&ephemeral_channel_id) { if let Ok(leave_builder) = events::build_leave(eph_uuid) { if let Err(e) = submit_event(leave_builder, &state).await { eprintln!("buzz-desktop: huddle leave ephemeral channel failed: {e}"); } } } } } teardown_huddle(&state)?; close_huddle_window(&app, &ephemeral_channel_id); Ok(()) } /// End the current huddle (creator only). /// /// Steps: /// 1. Emit KIND_HUDDLE_ENDED to the parent channel. /// 2. Archive the ephemeral channel. /// 3. Shut down the STT pipeline (Fix 5). /// 4. Clear local huddle state. #[tauri::command] pub async fn end_huddle( force: Option, app: tauri::AppHandle, state: State<'_, AppState>, ) -> Result<(), String> { let (parent_channel_id, ephemeral_channel_id) = { let mut hs = state.huddle()?; if hs.phase == HuddlePhase::Idle { return Ok(()); // Nothing to end. } // Only the creator can end the huddle for everyone. Non-creators // should use leave_huddle (which auto-ends if they're the last human). // The `force` flag allows recovery when the creator has disconnected // ungracefully — the UI should gate this behind a confirmation dialog. if !hs.is_creator && !force.unwrap_or(false) { return Err("only the huddle creator can end it — use leave_huddle instead".into()); } hs.phase = HuddlePhase::Leaving; ( hs.parent_channel_id.clone().unwrap_or_default(), hs.ephemeral_channel_id.clone().unwrap_or_default(), ) }; emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state).await; teardown_huddle(&state)?; close_huddle_window(&app, &ephemeral_channel_id); Ok(()) } /// Confirm that the frontend has established mic + AudioWorklet. /// Transitions from Connected → Active. No-op if already Active. #[tauri::command] pub async fn confirm_huddle_active(state: State<'_, AppState>) -> Result<(), String> { let transitioned = { let mut hs = state.huddle()?; match hs.phase { HuddlePhase::Connected => { hs.phase = HuddlePhase::Active; true } HuddlePhase::Active => false, // Already active — idempotent. ref other => return Err(format!("cannot confirm active: phase is {:?}", other)), } }; if transitioned { state.emit_huddle_state_changed(); } Ok(()) } /// Return the current HuddleState (serialized for the frontend). #[tauri::command] pub fn get_huddle_state(state: State<'_, AppState>) -> Result { let hs = state.huddle()?; Ok(hs.clone()) } /// Return the authoritative list of agent (bot-role) pubkeys in the active huddle. /// /// Fetches from the relay's channel membership API — works for both creators /// and joiners. Returns `Ok(Vec::new())` if no huddle is active. Returns /// `Err` on relay fetch failure so the frontend can keep `agentsLoaded = false` /// rather than treating a failed lookup as "zero agents". #[tauri::command] pub async fn get_huddle_agent_pubkeys(state: State<'_, AppState>) -> Result, String> { let eph_id = { let hs = state.huddle()?; hs.ephemeral_channel_id.clone() }; match eph_id { Some(id) => fetch_channel_members(&id, Some("bot"), &state).await, None => Ok(Vec::new()), } } /// Maximum IPC audio batch size: 100 KB. /// A 100 ms batch at 48 kHz mono f32 is ~19 KB; 100 KB allows headroom /// without letting a malformed IPC call allocate unbounded memory. const MAX_AUDIO_BATCH_BYTES: usize = 100 * 1024; /// Receive raw PCM audio bytes from the AudioWorklet and feed the STT pipeline. /// /// Expects a raw binary body of f32 LE samples at 48 kHz mono. /// If no STT pipeline is active, the bytes are silently discarded. #[tauri::command] pub fn push_audio_pcm( request: tauri::ipc::Request<'_>, state: State<'_, AppState>, ) -> Result<(), String> { match request.body() { tauri::ipc::InvokeBody::Raw(bytes) => { if bytes.len() > MAX_AUDIO_BATCH_BYTES { return Err(format!( "audio batch too large: {} bytes (max {})", bytes.len(), MAX_AUDIO_BATCH_BYTES )); } if let Ok(hs) = state.huddle() { // Fan out to STT pipeline. if let Some(ref pipeline) = hs.stt_pipeline { pipeline.push_audio(bytes.to_vec())?; } // Fan out to audio relay encoder (best-effort, non-blocking). if let Some(ref pcm_tx) = hs.audio_relay_pcm_tx { let _ = pcm_tx.try_send(bytes.to_vec()); } } Ok(()) } _ => Err("expected raw binary body".to_string()), } } /// Trigger a background download of voice models (Parakeet STT + Pocket TTS). /// /// Returns immediately — downloads run in tokio background tasks. /// Poll `get_model_status` to track progress. /// Safe to call multiple times: no-op if already downloading or ready. #[tauri::command] pub async fn download_voice_models(state: State<'_, AppState>) -> Result<(), String> { let manager = models::global_model_manager() .ok_or("model manager unavailable (home directory could not be resolved)")?; manager.start_stt_download(state.http_client.clone()); manager.start_tts_download(state.http_client.clone()); Ok(()) } /// Return the current download status for all voice models. #[tauri::command] pub fn get_model_status(_state: State<'_, AppState>) -> Result { let manager = models::global_model_manager() .ok_or("model manager unavailable (home directory could not be resolved)")?; Ok(models::VoiceModelStatus { stt: manager.stt_status(), tts: manager.tts_status(), }) } /// Speak an agent message via TTS. /// /// Called by the WebView when it receives an eligible live agent message. /// Lazily starts the TTS pipeline if models are ready but the pipeline hasn't /// been created yet (e.g. models finished downloading after huddle started). /// /// Disabled is the only intentional no-op. Enabled-but-unavailable speech /// returns an error so the caller cannot mistake a dropped message for success. #[tauri::command] pub async fn speak_agent_message( text: String, route_id: u64, speaker_pubkey: String, app: tauri::AppHandle, state: State<'_, AppState>, ) -> Result<(), String> { eprintln!("buzz-desktop: tts stage=invoke status=started route_id={route_id}"); // Truncate oversized messages — agents shouldn't monologue in a voice huddle. // Use char count (not byte length) to avoid panicking on multi-byte UTF-8. let text = normalize_agent_tts_text(text); if !state.huddle()?.tts_enabled { eprintln!( "buzz-desktop: tts stage=invoke status=no_op reason=disabled route_id={route_id}" ); return Ok(()); } let Some(voice_reference) = agent_voice::voice_reference_for_agent(&app, &state, &speaker_pubkey)? else { eprintln!( "buzz-desktop: tts stage=invoke status=no_op reason=agent_disabled route_id={route_id}" ); return Ok(()); }; let needs_pipeline = { let mut hs = state.huddle()?; if hs .tts_pipeline .as_ref() .is_some_and(|pipeline| pipeline.is_finished()) { hs.tts_pipeline = None; } match classify_agent_tts_runtime(hs.tts_enabled, &hs.phase, hs.tts_pipeline.is_some()) { AgentTtsRuntimeGate::Disabled => { eprintln!( "buzz-desktop: tts stage=invoke status=no_op reason=disabled route_id={route_id}" ); return Ok(()); } AgentTtsRuntimeGate::Inactive => { eprintln!( "buzz-desktop: tts stage=invoke status=failed reason=inactive_huddle route_id={route_id}" ); return Err( "Agent text to speech is unavailable outside an active huddle".to_string(), ); } AgentTtsRuntimeGate::NeedsPipeline => true, AgentTtsRuntimeGate::Ready => false, } }; // Lazy-start: models may have finished downloading after the huddle began. if needs_pipeline { maybe_start_tts_pipeline(&state).await.inspect_err(|_| { eprintln!( "buzz-desktop: tts stage=invoke status=failed reason=startup_failed route_id={route_id}" ); })?; await_inflight_tts_start(&state).await.inspect_err(|_| { eprintln!( "buzz-desktop: tts stage=invoke status=failed reason=startup_timeout route_id={route_id}" ); })?; } let pipeline = { 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(Arc::clone) }; let Some(pipeline) = pipeline else { eprintln!( "buzz-desktop: tts stage=invoke status=failed reason=unavailable route_id={route_id}" ); return Err("Agent text to speech is enabled but its audio pipeline is unavailable".into()); }; match agent_tts_publisher::ensure(&app, &state, &pipeline, &speaker_pubkey).await { Ok(true) => eprintln!( "buzz-desktop: tts broadcast status=ready route_id={route_id}" ), Ok(false) => eprintln!( "buzz-desktop: tts broadcast status=unavailable reason=agent_identity_not_local route_id={route_id}" ), Err(error) => eprintln!( "buzz-desktop: tts broadcast status=unavailable reason=publisher_setup_failed route_id={route_id} error={error}" ), } let sender = pipeline.text_sender(); let speaker_generation = sender.speaker_generation(&speaker_pubkey); enqueue_agent_tts_text(route_id, text, move |route_id, text| { sender .send( route_id, speaker_pubkey, speaker_generation, voice_reference, text, ) .map_err(|error| format!("TTS queue closed while waiting to enqueue: {error}")) }) .await .inspect(|_| eprintln!("buzz-desktop: tts stage=queue status=accepted route_id={route_id}")) .inspect_err(|_| { eprintln!("buzz-desktop: tts stage=queue status=failed reason=closed route_id={route_id}") }) }