diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index 459fa75743..4a07cf2d9e 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -127,6 +127,7 @@ export default defineConfig({ "**/harness-management.spec.ts", "**/harness-catalog-screenshots.spec.ts", "**/inline-custom-harness.spec.ts", + "**/huddle-transcription.spec.ts", ], use: { ...devices["Desktop Chrome"], diff --git a/desktop/src-tauri/src/huddle/agents.rs b/desktop/src-tauri/src/huddle/agents.rs index 02c4045410..2de22f99d8 100644 --- a/desktop/src-tauri/src/huddle/agents.rs +++ b/desktop/src-tauri/src/huddle/agents.rs @@ -2,7 +2,8 @@ //! //! Mental model: //! add_agent_to_huddle → kind:9000 to ephemeral channel -//! → kind:9000 to parent channel (best-effort) +//! → preserve existing parent membership, or +//! kind:9000 to parent channel (best-effort) //! //! ACP spawning is NOT needed here: the running agent process auto-subscribes //! when it receives the kind:9000 membership notification. Huddle-specific @@ -11,7 +12,10 @@ use serde::Serialize; use uuid::Uuid; -use crate::{app_state::AppState, events, relay::submit_event}; +use crate::{ + app_state::AppState, events, huddle::relay_api::fetch_channel_members_with_roles, + relay::submit_event, +}; // ── Constants ───────────────────────────────────────────────────────────────── @@ -61,8 +65,9 @@ with the next one. /// The field exists for forward compatibility with future batch-add operations /// where partial success may be meaningful. /// -/// `parent_added` reflects whether the parent-channel add succeeded; -/// `parent_error` carries the error string when it didn't. +/// `parent_added` reflects whether the parent already contained the agent or +/// the parent-channel add succeeded; `parent_error` carries the error string +/// when neither condition could be confirmed. #[derive(Debug, Serialize)] pub struct AgentAddResult { /// Always `true` — invariant guaranteed by [`add_agent_to_huddle`]. @@ -91,17 +96,33 @@ pub async fn add_agent_to_huddle( let add_eph = events::build_add_member(ephemeral_channel_id, agent_pubkey, Some("bot"))?; submit_event(add_eph, state).await?; - // 2. Add agent to parent channel — so agent has full context. - // Best-effort: capture the error but don't propagate it. - let (parent_added, parent_error) = { + // 2. Preserve any active parent membership, regardless of role. Rewriting + // an existing DM member as `bot` is both unnecessary and forbidden for + // non-admins. Otherwise add the agent so it has full context. + // Best-effort: capture a real error but don't propagate it. + let parent_channel_id_string = parent_channel_id.to_string(); + let parent_already_contains_agent = + fetch_channel_members_with_roles(&parent_channel_id_string, state) + .await + .is_ok_and(|members| contains_member(&members, agent_pubkey)); + + let (parent_added, parent_error) = if parent_already_contains_agent { + (true, None) + } else { let add_parent = events::build_add_member(parent_channel_id, agent_pubkey, Some("bot"))?; match submit_event(add_parent, state).await { Ok(_) => (true, None), Err(e) => { - eprintln!( - "buzz-desktop: add agent to parent channel failed (may already be member): {e}" - ); - (false, Some(e)) + let active_after_error = + fetch_channel_members_with_roles(&parent_channel_id_string, state) + .await + .is_ok_and(|members| contains_member(&members, agent_pubkey)); + if active_after_error { + (true, None) + } else { + eprintln!("buzz-desktop: add agent to parent channel failed: {e}"); + (false, Some(e)) + } } } }; @@ -112,3 +133,26 @@ pub async fn add_agent_to_huddle( parent_error, }) } + +fn contains_member(members: &[(String, Option)], pubkey: &str) -> bool { + members + .iter() + .any(|(member_pubkey, _)| member_pubkey.eq_ignore_ascii_case(pubkey)) +} + +#[cfg(test)] +mod tests { + use super::contains_member; + + #[test] + fn existing_parent_membership_is_preserved_regardless_of_role() { + let members = vec![ + ("agent-member".to_owned(), Some("member".to_owned())), + ("agent-bot".to_owned(), Some("bot".to_owned())), + ]; + + assert!(contains_member(&members, "AGENT-MEMBER")); + assert!(contains_member(&members, "agent-bot")); + assert!(!contains_member(&members, "missing")); + } +} diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index a815bf2d06..0a84cffe00 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -66,13 +66,16 @@ pub use transcription::{set_huddle_transcription_enabled, start_stt_pipeline}; // ── Imports ─────────────────────────────────────────────────────────────────── -use std::sync::{atomic::Ordering, Arc}; +use std::sync::atomic::Ordering; use tauri::State; use uuid::Uuid; use crate::{app_state::AppState, events, relay::submit_event}; - -use pipeline::{maybe_start_stt_pipeline, maybe_start_tts_pipeline, post_connect_setup}; +pub use pipeline::check_pipeline_hotstart; +use pipeline::{ + maybe_start_stt_pipeline, maybe_start_tts_pipeline, post_connect_setup, + start_auto_enabled_transcription, PostConnectOutcome, +}; use relay_api::{ count_human_members, fetch_channel_members, parse_channel_uuid, validate_pubkey_hex, MAX_HUDDLE_AGENTS, @@ -186,7 +189,7 @@ pub async fn start_huddle( }; // Transition to Creating. - { + let huddle_generation = { let mut hs = state.huddle()?; if hs.phase != HuddlePhase::Idle { return Err(format!( @@ -194,9 +197,11 @@ pub async fn start_huddle( hs.phase )); } + let generation = hs.begin_huddle_lifetime(); hs.phase = HuddlePhase::Creating; hs.parent_channel_id = Some(parent_channel_id.clone()); - } + generation + }; let ephemeral_uuid = Uuid::new_v4(); let ephemeral_channel_id = ephemeral_uuid.to_string(); @@ -259,27 +264,33 @@ pub async fn start_huddle( match result { Ok(successful_agents) => { // 5. Store active state. - { + let committed = { let mut hs = state.huddle()?; - hs.phase = HuddlePhase::Connected; - hs.is_creator = true; - hs.ephemeral_channel_id = Some(ephemeral_channel_id.clone()); - // Only store agents that were successfully enrolled. - *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = - successful_agents.clone(); - // Include the current user + successfully enrolled agents as participants. - // Use successful_agents (not member_pubkeys) so failed enrollments - // are not reflected in the participant list. - 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); + 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.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 } - hs.participants = participants; + }; + if !committed { + emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state).await; + return Err("huddle start was superseded".to_owned()); } // 6. Notify frontend of state change. @@ -287,16 +298,30 @@ pub async fn start_huddle( // 7. Hydrate members, download models, start pipelines (incl. audio relay). // Audio relay failure is fatal — no point in a huddle without audio. - if let Err(e) = post_connect_setup(&state, &ephemeral_channel_id).await { - // Rollback: audio relay failed after state was committed. - // Publish the terminal lifecycle event before archiving so - // other clients do not reconstruct a phantom active huddle. - emit_end_and_archive(&parent_channel_id, &ephemeral_channel_id, &state).await; - if let Ok(mut hs) = state.huddle_state.lock() { - hs.reset_preserving_generation(); + match post_connect_setup(&state, &ephemeral_channel_id, huddle_generation).await { + Ok(PostConnectOutcome::Ready) => {} + Ok(PostConnectOutcome::Stale) => { + 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(); + } + return Err(e); } - state.emit_huddle_state_changed(); - return Err(e); } Ok(HuddleJoinInfo { @@ -314,11 +339,11 @@ pub async fn start_huddle( } } } - // Reset state to Idle so the user can retry. - // Preserve session_generation so in-flight transcription tasks - // from a prior session still see a stale generation and exit. + // Reset only if this failed attempt still owns the Creating state. if let Ok(mut hs) = state.huddle_state.lock() { - hs.reset_preserving_generation(); + if hs.owns_huddle_lifetime(huddle_generation, HuddlePhase::Creating) { + hs.reset_preserving_generation(); + } } Err(e) } @@ -340,7 +365,7 @@ pub async fn join_huddle( state: State<'_, AppState>, ) -> Result { // Transition to Connecting. - { + let huddle_generation = { let mut hs = state.huddle()?; if hs.phase != HuddlePhase::Idle { return Err(format!( @@ -348,10 +373,12 @@ pub async fn join_huddle( 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()); - } + generation + }; // Seed participant list with own pubkey as a fallback until relay responds. let own_pubkey = state @@ -360,12 +387,20 @@ pub async fn join_huddle( .map(|k| k.public_key().to_hex()) .unwrap_or_default(); - { + let committed = { let mut hs = state.huddle()?; - hs.phase = HuddlePhase::Connected; - if !own_pubkey.is_empty() { - hs.participants = vec![own_pubkey]; + 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. @@ -373,15 +408,25 @@ pub async fn join_huddle( // Hydrate members, download models, start pipelines (incl. audio relay). // Audio relay failure is fatal — no point in a huddle without audio. - if let Err(e) = post_connect_setup(&state, &ephemeral_channel_id).await { - // Rollback: audio relay failed after state was committed. - // Reset state to Idle so the user can retry. The ephemeral channel - // has a TTL and will expire — no manual archive needed for joiners. - if let Ok(mut hs) = state.huddle_state.lock() { - hs.reset_preserving_generation(); + 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); } - state.emit_huddle_state_changed(); - return Err(e); } Ok(HuddleJoinInfo { @@ -675,123 +720,6 @@ pub fn push_audio_pcm( } } -/// Hot-start: check if voice models just finished downloading during an active -/// huddle and start the corresponding pipelines. -/// -/// Called by the frontend on a timer or after model status changes. No-op if -/// the huddle is not active or pipelines are already running. -#[tauri::command] -pub async fn check_pipeline_hotstart(state: State<'_, AppState>) -> Result<(), String> { - let (is_active, ephemeral_channel_id) = { - let hs = state.huddle()?; - ( - matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active), - hs.ephemeral_channel_id.clone(), - ) - }; - - if !is_active { - return Ok(()); - } - - // Detect dead pipelines: if the worker thread has exited (init failure or crash), - // clear the pipeline handle so hot-start can retry on the next cycle. - { - let mut hs = state.huddle()?; - if let Some(ref p) = hs.stt_pipeline { - if p.is_finished() { - hs.stt_pipeline = None; - } - } - if let Some(ref p) = hs.tts_pipeline { - if p.is_finished() { - hs.tts_pipeline = None; - } - } - } - // Re-read after potential cleanup. - let (has_stt, has_tts, transcription_enabled) = { - let hs = state.huddle()?; - ( - hs.stt_pipeline.is_some(), - hs.tts_pipeline.is_some(), - hs.transcription_enabled, - ) - }; - - // Check if models just became ready (one-shot flags). - let stt_ready = models::global_model_manager() - .map(|m| m.take_stt_ready()) - .unwrap_or(false); - let tts_ready = models::global_model_manager() - .map(|m| m.take_tts_ready()) - .unwrap_or(false); - - // Start TTS first (so STT can capture tts_cancel). - if !has_tts && (tts_ready || models::is_tts_ready()) { - if let Err(e) = maybe_start_tts_pipeline(&state).await { - eprintln!("buzz-desktop: TTS hotstart failed: {e}"); - } - } - - if transcription_enabled && !has_stt && (stt_ready || models::is_stt_ready()) { - if let Some(eph_id) = &ephemeral_channel_id { - if let Err(e) = maybe_start_stt_pipeline(&state, eph_id).await { - eprintln!("buzz-desktop: STT hotstart failed: {e}"); - } - } - } - - // Periodically refresh agent_pubkeys from relay membership. - // This catches mid-huddle agent additions/removals by other participants, - // keeping STT p-tags authoritative throughout the session. - // Throttled to every 15 s (not on every 5 s hotstart poll). - // - // NOTE: The frontend ALSO polls agent membership independently (every 10 s - // via get_huddle_agent_pubkeys). This is intentional — the two polls have - // different failure semantics: - // - Rust (here): preserves stale list on failure (STT p-tags should not - // disappear on a transient network blip). - // - React (HuddleContext.tsx): clears list on failure (TTS authorization - // must fail-closed — never speak from a stale agent list). - // - // On Ok: always replace (even with empty — agents may have been removed). - // On Err: preserve the existing list (transient failure shouldn't zero it). - if let Some(eph_id) = &ephemeral_channel_id { - let should_refresh = { - let hs = state.huddle()?; - match hs.last_agent_refresh { - None => true, - Some(t) => t.elapsed() >= std::time::Duration::from_secs(15), - } - }; - if should_refresh { - // Fetch agents (for STT p-tags) and all members (for participant list). - // Sequential — tokio::join! requires the `macros` feature. - // Only update the throttle timestamp when at least one fetch succeeds, - // so transient failures retry immediately on the next poll cycle. - // Fetch both lists before acquiring the lock — no lock held across await. - let fresh_agents = fetch_channel_members(eph_id, Some("bot"), &state) - .await - .ok(); - let fresh_members = fetch_channel_members(eph_id, None, &state).await.ok(); - - if fresh_agents.is_some() || fresh_members.is_some() { - let mut hs = state.huddle()?; - if let Some(agents) = fresh_agents { - *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = agents; - } - if let Some(members) = fresh_members { - hs.participants = members; - } - hs.last_agent_refresh = Some(std::time::Instant::now()); - } - } - } - - Ok(()) -} - /// Trigger a background download of voice models (Parakeet STT + Pocket TTS). /// /// Returns immediately — downloads run in tokio background tasks. @@ -924,7 +852,7 @@ pub async fn add_agent_to_huddle( ) -> Result { validate_pubkey_hex(&agent_pubkey)?; - let (eph_id, parent_id) = { + let (eph_id, parent_id, huddle_generation) = { let hs = state.huddle()?; if !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) { return Err("no active huddle".to_string()); @@ -948,7 +876,7 @@ pub async fn add_agent_to_huddle( .clone() .ok_or("no ephemeral channel")?; let parent = hs.parent_channel_id.clone().ok_or("no parent channel")?; - (eph, parent) + (eph, parent, hs.huddle_generation) }; let eph_uuid = Uuid::parse_str(&eph_id).map_err(|e| e.to_string())?; @@ -957,29 +885,30 @@ pub async fn add_agent_to_huddle( // Returns Err only if the ephemeral add fails — parent failure is in the result. let result = agents::add_agent_to_huddle(eph_uuid, parent_uuid, &agent_pubkey, &state).await?; - // Ephemeral add succeeded — safe to register for p-tagging. - // Clone the Arc first so we can drop the outer HuddleState lock before - // acquiring the inner pubkeys lock (avoids the E0597 borrow-checker error). - { - let agent_pubkeys_arc = { - let hs = state.huddle()?; - Arc::clone(&hs.agent_pubkeys) - }; - let mut pubkeys = agent_pubkeys_arc.lock().unwrap_or_else(|e| e.into_inner()); + // Ephemeral add succeeded — register it only if this is still the huddle + // that initiated the relay operation. + let transcription_auto_enabled = { + let mut hs = state.huddle()?; + if !hs.is_current_huddle(&eph_id, huddle_generation) { + return Ok(result); + } + let mut pubkeys = hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); if !pubkeys.contains(&agent_pubkey) { pubkeys.push(agent_pubkey.clone()); } - } + drop(pubkeys); + if !hs.participants.contains(&agent_pubkey) { + hs.participants.push(agent_pubkey.clone()); + } + hs.maybe_auto_enable_transcription_for_agents() + }; // No guidelines re-post needed — the agent sees the original kind:48106 // guidelines via EOSE replay when it subscribes to the ephemeral channel. - - // Also add the agent to the visible participants list. - { - let mut hs = state.huddle()?; - if !hs.participants.contains(&agent_pubkey) { - hs.participants.push(agent_pubkey); - } + if transcription_auto_enabled { + start_auto_enabled_transcription(&state, &eph_id).await; + } else { + state.emit_huddle_state_changed(); } Ok(result) diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index ceccedd8b6..48073bc16c 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -9,6 +9,7 @@ use std::sync::{ }; use nostr::JsonUtil; +use tauri::State; use uuid::Uuid; use crate::app_state::AppState; @@ -20,55 +21,224 @@ use super::state::{HuddlePhase, VoiceInputMode}; use super::stt; use super::tts; +pub(crate) enum PostConnectOutcome { + Ready, + Stale, +} + +/// Hot-start: check if voice models just finished downloading during an active +/// huddle and start the corresponding pipelines. +/// +/// Called by the frontend on a timer or after model status changes. No-op if +/// the huddle is not active or pipelines are already running. +#[tauri::command] +pub async fn check_pipeline_hotstart(state: State<'_, AppState>) -> Result<(), String> { + let (is_active, ephemeral_channel_id, huddle_generation) = { + let hs = state.huddle()?; + ( + matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active), + hs.ephemeral_channel_id.clone(), + hs.huddle_generation, + ) + }; + + if !is_active { + return Ok(()); + } + + // Detect dead pipelines: if the worker thread has exited (init failure or crash), + // clear the pipeline handle so hot-start can retry on the next cycle. + { + let mut hs = state.huddle()?; + if let Some(ref p) = hs.stt_pipeline { + if p.is_finished() { + hs.stt_pipeline = None; + } + } + if let Some(ref p) = hs.tts_pipeline { + if p.is_finished() { + hs.tts_pipeline = None; + } + } + } + // Re-read after potential cleanup. + let (has_stt, has_tts, transcription_enabled) = { + let hs = state.huddle()?; + ( + hs.stt_pipeline.is_some(), + hs.tts_pipeline.is_some(), + hs.transcription_enabled, + ) + }; + + // Check if models just became ready (one-shot flags). + let stt_ready = models::global_model_manager() + .map(|m| m.take_stt_ready()) + .unwrap_or(false); + let tts_ready = models::global_model_manager() + .map(|m| m.take_tts_ready()) + .unwrap_or(false); + + // Start TTS first (so STT can capture tts_cancel). + if !has_tts && (tts_ready || models::is_tts_ready()) { + if let Err(e) = maybe_start_tts_pipeline(&state).await { + eprintln!("buzz-desktop: TTS hotstart failed: {e}"); + } + } + if transcription_enabled && !has_stt && (stt_ready || models::is_stt_ready()) { + if let Some(eph_id) = &ephemeral_channel_id { + if let Err(e) = maybe_start_stt_pipeline(&state, eph_id).await { + eprintln!("buzz-desktop: STT hotstart failed: {e}"); + } + } + } + + // Periodically refresh agent membership from the relay. + // This catches mid-huddle additions/removals by other participants, keeps + // STT p-tags authoritative, and auto-enables transcription when the first + // agent appears unless the user has already chosen a transcription state. + // Throttled independently from the more frequent hotstart poll. + // + // NOTE: The frontend ALSO polls agent membership independently via + // get_huddle_agent_pubkeys. This is intentional — the two polls have + // different failure semantics: + // - Rust (here): preserves stale list on failure (STT p-tags should not + // disappear on a transient network blip). + // - React (HuddleContext.tsx): clears list on failure (TTS authorization + // must fail-closed — never speak from a stale agent list). + // + // On Ok: always replace (even with empty — agents may have been removed). + // On Err: preserve the existing list (transient failure shouldn't zero it). + if let Some(eph_id) = &ephemeral_channel_id { + let should_refresh = { + let hs = state.huddle()?; + match hs.last_agent_refresh { + None => true, + Some(t) => t.elapsed() >= std::time::Duration::from_secs(15), + } + }; + if should_refresh { + // Fetch agents (for STT p-tags) before all members (for participant + // list) so relay membership queries remain ordered. + // Only update the throttle timestamp when at least one fetch succeeds, + // so transient failures retry immediately on the next poll cycle. + // Fetch both lists before acquiring the lock — no lock held across await. + let fresh_agents = fetch_channel_members(eph_id, Some("bot"), &state) + .await + .ok(); + let fresh_members = fetch_channel_members(eph_id, None, &state).await.ok(); + let transcription_auto_enabled = if fresh_agents.is_some() || fresh_members.is_some() { + let mut hs = state.huddle()?; + if !hs.is_current_huddle(eph_id, huddle_generation) { + return Ok(()); + } + if let Some(agents) = fresh_agents { + *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = agents; + } + if let Some(members) = fresh_members { + hs.participants = members; + } + hs.last_agent_refresh = Some(std::time::Instant::now()); + hs.maybe_auto_enable_transcription_for_agents() + } else { + false + }; + if transcription_auto_enabled { + start_auto_enabled_transcription(&state, eph_id).await; + } + } + } + + Ok(()) +} + pub(crate) async fn post_connect_setup( state: &AppState, ephemeral_channel_id: &str, -) -> Result<(), String> { + huddle_generation: u64, +) -> Result { + { + let hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + return Ok(PostConnectOutcome::Stale); + } + } + // Hydrate agent pubkeys and participants from relay in parallel // (authoritative — overrides local guesses). let (agents_result, all_members_result) = tokio::join!( fetch_channel_members(ephemeral_channel_id, Some("bot"), state), fetch_channel_members(ephemeral_channel_id, None, state), ); - if let Ok(agents) = agents_result { - let hs = state.huddle()?; - *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = agents; - } - - if let Ok(all_members) = all_members_result { - if !all_members.is_empty() { - let mut hs = state.huddle()?; - hs.participants = all_members; + let transcription_auto_enabled = { + let mut hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + return Ok(PostConnectOutcome::Stale); } + if let Ok(agents) = agents_result { + *hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()) = agents; + } + if let Ok(all_members) = all_members_result { + if !all_members.is_empty() { + hs.participants = all_members; + } + } + hs.maybe_auto_enable_transcription_for_agents() + }; + + if transcription_auto_enabled { + state.emit_huddle_state_changed(); } - // Prepare TTS for agent voice. STT is transcript-specific and starts only - // when transcription is explicitly enabled. + // Prepare voice models. Agent presence may have auto-enabled transcription; + // explicit user choices remain authoritative. if let Some(mgr) = models::global_model_manager() { mgr.start_tts_download(state.http_client.clone()); + if state.huddle()?.transcription_enabled { + mgr.start_stt_download(state.http_client.clone()); + } } // Connect audio relay WebSocket (Opus encode/decode pipeline). // This is the core audio path — failure is fatal for the huddle. let parent_id = { let hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + return Ok(PostConnectOutcome::Stale); + } hs.parent_channel_id.clone() }; - let (cancel, pcm_tx) = - relay_api::connect_audio_relay(ephemeral_channel_id, parent_id.as_deref(), state).await?; + let audio_result = + relay_api::connect_audio_relay(ephemeral_channel_id, parent_id.as_deref(), state).await; { let mut hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + if let Ok((cancel, _)) = audio_result { + cancel.cancel(); + } + return Ok(PostConnectOutcome::Stale); + } + let (cancel, pcm_tx) = audio_result?; hs.audio_ws_cancel = Some(cancel); hs.audio_relay_pcm_tx = Some(pcm_tx); } - // Start TTS immediately. STT/transcript posting is opt-in and starts only - // after the user explicitly enables transcription. + // Start TTS immediately, then STT when transcription is enabled either by + // the user or by authoritative agent membership. + if !state + .huddle()? + .is_current_huddle(ephemeral_channel_id, huddle_generation) + { + return Ok(PostConnectOutcome::Stale); + } if let Err(e) = maybe_start_tts_pipeline(state).await { eprintln!("buzz-desktop: TTS pipeline failed to start: {e}"); } + if let Err(e) = maybe_start_stt_pipeline(state, ephemeral_channel_id).await { + eprintln!("buzz-desktop: STT pipeline failed to start: {e}"); + } - Ok(()) + Ok(PostConnectOutcome::Ready) } /// Attempt to start the STT pipeline if models are present. @@ -83,12 +253,16 @@ pub(crate) async fn maybe_start_stt_pipeline( state: &AppState, ephemeral_channel_id: &str, ) -> Result { - { + let huddle_generation = { let hs = state.huddle()?; - if !hs.transcription_enabled { + if !hs.transcription_enabled + || !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) + || hs.ephemeral_channel_id.as_deref() != Some(ephemeral_channel_id) + { return Ok(false); } - } + hs.huddle_generation + }; if !models::is_stt_ready() { return Ok(false); // Models not downloaded yet — voice-only mode. @@ -97,21 +271,29 @@ pub(crate) async fn maybe_start_stt_pipeline( let channel_uuid = parse_channel_uuid(ephemeral_channel_id)?; - // Atomically claim the construction slot (mirrors tts_starting pattern). - { - let hs = state.huddle()?; - if hs.stt_starting.swap(true, Ordering::AcqRel) { - return Ok(false); // Another caller is already constructing. - } - } - - // Grab shared flags, agent pubkeys, and session generation from HuddleState. + // Atomically claim construction and grab shared state under one lock. // If replacing an existing pipeline, bump generation first so the old // transcription task's next POST sees a stale generation and exits. // Take the old pipeline OUT of the lock before dropping — Drop joins // the worker thread (~200ms) and must not block under the mutex. - let (tts_active, tts_cancel, agent_pubkeys_arc, session_gen, ptt_active_for_stt, old_stt) = { + let ( + tts_active, + tts_cancel, + agent_pubkeys_arc, + session_gen, + expected_generation, + stt_starting, + ptt_active_for_stt, + old_stt, + ) = { let mut hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + return Ok(false); + } + if hs.stt_starting.swap(true, Ordering::AcqRel) { + return Ok(false); + } + let stt_starting = Arc::clone(&hs.stt_starting); // Invalidate any existing transcription task before replacing the pipeline. if hs.stt_pipeline.is_some() { hs.session_generation.fetch_add(1, Ordering::Release); @@ -130,6 +312,8 @@ pub(crate) async fn maybe_start_stt_pipeline( Some(Arc::clone(&hs.tts_cancel)), Arc::clone(&hs.agent_pubkeys), Arc::clone(&hs.session_generation), + hs.session_generation.load(Ordering::Acquire), + stt_starting, ptt, old, ) @@ -144,13 +328,11 @@ pub(crate) async fn maybe_start_stt_pipeline( let (pipeline, text_rx) = match constructed { Ok(Ok(p)) => p, Ok(Err(e)) => { - let hs = state.huddle()?; - hs.stt_starting.store(false, Ordering::Release); + stt_starting.store(false, Ordering::Release); return Err(e); } Err(e) => { - let hs = state.huddle()?; - hs.stt_starting.store(false, Ordering::Release); + stt_starting.store(false, Ordering::Release); return Err(format!("spawn_blocking failed: {e}")); } }; @@ -158,10 +340,14 @@ pub(crate) async fn maybe_start_stt_pipeline( { let mut hs = state.huddle()?; - hs.stt_starting.store(false, Ordering::Release); + stt_starting.store(false, Ordering::Release); // Phase check: huddle may have been torn down during construction. if !hs.transcription_enabled - || !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) + || !hs.is_current_transcription_generation( + ephemeral_channel_id, + huddle_generation, + expected_generation, + ) { return Ok(false); } @@ -172,6 +358,17 @@ pub(crate) async fn maybe_start_stt_pipeline( Ok(true) } +/// Start STT after agent presence automatically enables transcription. +pub(crate) async fn start_auto_enabled_transcription(state: &AppState, ephemeral_channel_id: &str) { + if let Some(manager) = models::global_model_manager() { + manager.start_stt_download(state.http_client.clone()); + } + if let Err(error) = maybe_start_stt_pipeline(state, ephemeral_channel_id).await { + eprintln!("buzz-desktop: auto-enabled STT failed to start: {error}"); + } + state.emit_huddle_state_changed(); +} + /// Attempt to start the TTS pipeline if TTS models are present and TTS is enabled. /// /// Returns `Ok(true)` if the pipeline was started, `Ok(false)` if preconditions diff --git a/desktop/src-tauri/src/huddle/state.rs b/desktop/src-tauri/src/huddle/state.rs index 876c2d688b..f0a2227ca8 100644 --- a/desktop/src-tauri/src/huddle/state.rs +++ b/desktop/src-tauri/src/huddle/state.rs @@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize}; use std::sync::{ - atomic::{AtomicBool, AtomicU64}, + atomic::{AtomicBool, AtomicU64, Ordering}, Arc, Mutex, }; @@ -80,6 +80,15 @@ pub struct HuddleState { pub tts_enabled: bool, /// Whether STT transcript posting is enabled for this huddle. pub transcription_enabled: bool, + /// Whether the user has explicitly used the transcription control in this + /// huddle. Agent presence may auto-enable transcription only while this is + /// false, so membership refreshes never undo an explicit user choice. + /// + /// This is backend-only session state: keeping it in `HuddleState` makes it + /// survive frontend remounts and audio reconnects, while huddle teardown + /// resets it for the next session. + #[serde(skip)] + pub transcription_user_controlled: bool, /// Shared flag: true while TTS is playing audio. /// Shared with the STT pipeline for barge-in / echo gating. #[serde(skip)] @@ -103,6 +112,10 @@ pub struct HuddleState { /// Used to throttle the refresh in check_pipeline_hotstart to every 15 s. #[serde(skip)] pub last_agent_refresh: Option, + /// Monotonic identity for a local huddle lifetime. Unlike transcript + /// generation, this changes only when a new start/join attempt begins. + #[serde(skip)] + pub huddle_generation: u64, /// Session generation — incremented on every teardown. The transcription /// task captures this at spawn time and checks before each POST. If the /// generation has changed, the task silently drops the transcript. @@ -157,11 +170,13 @@ impl Clone for HuddleState { is_creator: self.is_creator, tts_enabled: self.tts_enabled, transcription_enabled: self.transcription_enabled, + transcription_user_controlled: self.transcription_user_controlled, tts_active: Arc::clone(&self.tts_active), tts_cancel: Arc::clone(&self.tts_cancel), tts_starting: Arc::clone(&self.tts_starting), stt_starting: Arc::clone(&self.stt_starting), last_agent_refresh: self.last_agent_refresh, + huddle_generation: self.huddle_generation, session_generation: Arc::clone(&self.session_generation), voice_input_mode: self.voice_input_mode.clone(), ptt_active: Arc::clone(&self.ptt_active), @@ -184,11 +199,13 @@ impl Default for HuddleState { is_creator: false, tts_enabled: true, transcription_enabled: false, + transcription_user_controlled: false, tts_active: Arc::new(AtomicBool::new(false)), tts_cancel: Arc::new(AtomicBool::new(false)), tts_starting: Arc::new(AtomicBool::new(false)), stt_starting: Arc::new(AtomicBool::new(false)), last_agent_refresh: None, + huddle_generation: 0, session_generation: Arc::new(AtomicU64::new(0)), voice_input_mode: VoiceInputMode::default(), ptt_active: Arc::new(AtomicBool::new(false)), @@ -197,13 +214,233 @@ impl Default for HuddleState { } impl HuddleState { + /// Begin a new local huddle lifetime and return its identity. + pub(crate) fn begin_huddle_lifetime(&mut self) -> u64 { + self.huddle_generation = self.huddle_generation.wrapping_add(1); + self.huddle_generation + } + + pub(crate) fn owns_huddle_lifetime(&self, huddle_generation: u64, phase: HuddlePhase) -> bool { + self.huddle_generation == huddle_generation && self.phase == phase + } + + /// Whether an async result still belongs to the active huddle that + /// initiated it. The channel id is the huddle-session identity; transcript + /// generation changes within the same huddle must not invalidate it. + pub(crate) fn is_current_huddle( + &self, + ephemeral_channel_id: &str, + huddle_generation: u64, + ) -> bool { + matches!(self.phase, HuddlePhase::Connected | HuddlePhase::Active) + && self.ephemeral_channel_id.as_deref() == Some(ephemeral_channel_id) + && self.huddle_generation == huddle_generation + } + + /// Whether an STT construction still belongs to the current transcript + /// generation within the active huddle. + pub(crate) fn is_current_transcription_generation( + &self, + ephemeral_channel_id: &str, + huddle_generation: u64, + session_generation: u64, + ) -> bool { + self.is_current_huddle(ephemeral_channel_id, huddle_generation) + && self.session_generation.load(Ordering::Acquire) == session_generation + } + + /// Invalidate in-flight transcription work and give the next constructor a + /// fresh sentinel that stale constructors cannot clear. + pub(crate) fn invalidate_transcription_pipeline(&mut self) { + self.session_generation.fetch_add(1, Ordering::Release); + self.stt_starting = Arc::new(AtomicBool::new(false)); + } + + /// Record an explicit transcription choice made through the existing user + /// control. Later agent membership refreshes must preserve this choice. + pub(crate) fn set_transcription_enabled_by_user(&mut self, enabled: bool) { + self.transcription_enabled = enabled; + self.transcription_user_controlled = true; + } + + /// Enable transcription when an agent is present and the user has not + /// explicitly chosen a transcription state for this huddle. + /// + /// Returns true only for the transition from disabled to enabled, allowing + /// callers to start models/pipelines and emit state exactly once. Removing + /// the last agent deliberately leaves the current state unchanged. + pub(crate) fn maybe_auto_enable_transcription_for_agents(&mut self) -> bool { + let has_agent = !self + .agent_pubkeys + .lock() + .unwrap_or_else(|e| e.into_inner()) + .is_empty(); + if has_agent && !self.transcription_user_controlled && !self.transcription_enabled { + self.transcription_enabled = true; + return true; + } + false + } + /// Reset to default state while preserving the session generation counter. /// Used by start_huddle rollback, join_huddle rollback, and teardown_huddle /// to invalidate in-flight transcription tasks without losing the generation. pub(crate) fn reset_preserving_generation(&mut self) { let gen = Arc::clone(&self.session_generation); + let huddle_generation = self.huddle_generation; *self = Self::default(); self.session_generation = gen; + self.huddle_generation = huddle_generation; + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::Ordering; + + use super::HuddleState; + + fn set_agents(state: &HuddleState, agents: &[&str]) { + *state + .agent_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) = + agents.iter().map(|agent| (*agent).to_owned()).collect(); + } + + #[test] + fn first_agent_auto_enables_transcription_once() { + let mut state = HuddleState::default(); + set_agents(&state, &["agent"]); + + assert!(state.maybe_auto_enable_transcription_for_agents()); + assert!(state.transcription_enabled); + assert!(!state.maybe_auto_enable_transcription_for_agents()); + } + + #[test] + fn explicit_user_disable_is_not_undone_by_agent_presence() { + let mut state = HuddleState::default(); + set_agents(&state, &["agent"]); + assert!(state.maybe_auto_enable_transcription_for_agents()); + + state.set_transcription_enabled_by_user(false); + + assert!(!state.maybe_auto_enable_transcription_for_agents()); + assert!(!state.transcription_enabled); + } + + #[test] + fn last_agent_leaving_preserves_current_transcription_state() { + let mut state = HuddleState::default(); + set_agents(&state, &["agent"]); + assert!(state.maybe_auto_enable_transcription_for_agents()); + + set_agents(&state, &[]); + + assert!(!state.maybe_auto_enable_transcription_for_agents()); + assert!(state.transcription_enabled); + } + + #[test] + fn clone_preserves_user_control_across_frontend_state_reads() { + let mut state = HuddleState::default(); + state.set_transcription_enabled_by_user(false); + + let mut clone = state.clone(); + set_agents(&clone, &["agent"]); + + assert!(clone.transcription_user_controlled); + assert!(!clone.maybe_auto_enable_transcription_for_agents()); + } + + #[test] + fn stale_huddle_identity_is_rejected_after_replacement() { + let mut state = HuddleState { + phase: super::HuddlePhase::Active, + ephemeral_channel_id: Some("huddle-a".to_owned()), + ..HuddleState::default() + }; + let huddle_generation = state.begin_huddle_lifetime(); + let generation = state.session_generation.load(Ordering::Acquire); + assert!(state.is_current_huddle("huddle-a", huddle_generation)); + assert!(state.is_current_transcription_generation( + "huddle-a", + huddle_generation, + generation + )); + + state.session_generation.fetch_add(1, Ordering::Release); + assert!(state.is_current_huddle("huddle-a", huddle_generation)); + assert!(!state.is_current_transcription_generation( + "huddle-a", + huddle_generation, + generation + )); + + state.ephemeral_channel_id = Some("huddle-b".to_owned()); + + assert!(!state.is_current_huddle("huddle-a", huddle_generation)); + } + + #[test] + fn same_channel_rejoin_gets_a_new_huddle_lifetime() { + let mut state = HuddleState { + phase: super::HuddlePhase::Active, + ephemeral_channel_id: Some("huddle".to_owned()), + ..HuddleState::default() + }; + let first_generation = state.begin_huddle_lifetime(); + state.reset_preserving_generation(); + state.phase = super::HuddlePhase::Active; + state.ephemeral_channel_id = Some("huddle".to_owned()); + let second_generation = state.begin_huddle_lifetime(); + + assert_ne!(first_generation, second_generation); + assert!(!state.is_current_huddle("huddle", first_generation)); + assert!(state.is_current_huddle("huddle", second_generation)); + } + + #[test] + fn superseded_create_cannot_commit_or_reset_replacement_lifetime() { + let mut state = HuddleState::default(); + let first_generation = state.begin_huddle_lifetime(); + state.phase = super::HuddlePhase::Creating; + assert!(state.owns_huddle_lifetime(first_generation, super::HuddlePhase::Creating)); + + state.reset_preserving_generation(); + let replacement_generation = state.begin_huddle_lifetime(); + state.phase = super::HuddlePhase::Creating; + + assert!(!state.owns_huddle_lifetime(first_generation, super::HuddlePhase::Creating)); + assert!(state.owns_huddle_lifetime(replacement_generation, super::HuddlePhase::Creating)); + } + + #[test] + fn superseded_join_cannot_commit_replacement_lifetime() { + let mut state = HuddleState::default(); + let first_generation = state.begin_huddle_lifetime(); + state.phase = super::HuddlePhase::Connecting; + + state.reset_preserving_generation(); + let replacement_generation = state.begin_huddle_lifetime(); + state.phase = super::HuddlePhase::Connecting; + + assert!(!state.owns_huddle_lifetime(first_generation, super::HuddlePhase::Connecting)); + assert!(state.owns_huddle_lifetime(replacement_generation, super::HuddlePhase::Connecting)); + } + + #[test] + fn stale_constructor_cannot_clear_replacement_sentinel() { + let mut state = HuddleState::default(); + let stale_sentinel = std::sync::Arc::clone(&state.stt_starting); + stale_sentinel.store(true, Ordering::Release); + + state.invalidate_transcription_pipeline(); + state.stt_starting.store(true, Ordering::Release); + stale_sentinel.store(false, Ordering::Release); + + assert!(state.stt_starting.load(Ordering::Acquire)); } } diff --git a/desktop/src-tauri/src/huddle/transcription.rs b/desktop/src-tauri/src/huddle/transcription.rs index 0d752c1de7..5962f57cf4 100644 --- a/desktop/src-tauri/src/huddle/transcription.rs +++ b/desktop/src-tauri/src/huddle/transcription.rs @@ -1,5 +1,3 @@ -use std::sync::atomic::Ordering; - use tauri::State; use crate::app_state::AppState; @@ -15,10 +13,12 @@ use super::{models, pipeline::maybe_start_stt_pipeline}; pub async fn start_stt_pipeline(state: State<'_, AppState>) -> Result<(), String> { let ephemeral_channel_id = { let mut hs = state.huddle()?; - hs.transcription_enabled = true; - hs.ephemeral_channel_id + let ephemeral_channel_id = hs + .ephemeral_channel_id .clone() - .ok_or("no active huddle — start or join a huddle first")? + .ok_or("no active huddle — start or join a huddle first")?; + hs.set_transcription_enabled_by_user(true); + ephemeral_channel_id }; match maybe_start_stt_pipeline(&state, &ephemeral_channel_id).await { @@ -41,14 +41,17 @@ pub async fn set_huddle_transcription_enabled( ) -> Result<(), String> { let (ephemeral_channel_id, old_stt) = { let mut hs = state.huddle()?; - hs.transcription_enabled = enabled; + let ephemeral_channel_id = hs + .ephemeral_channel_id + .clone() + .ok_or("no active huddle — start or join a huddle first")?; + hs.set_transcription_enabled_by_user(enabled); if enabled { - (hs.ephemeral_channel_id.clone(), None) + (ephemeral_channel_id, None) } else { - hs.session_generation.fetch_add(1, Ordering::Release); - hs.stt_starting.store(false, Ordering::Release); - (hs.ephemeral_channel_id.clone(), hs.stt_pipeline.take()) + hs.invalidate_transcription_pipeline(); + (ephemeral_channel_id, hs.stt_pipeline.take()) } }; @@ -58,12 +61,10 @@ pub async fn set_huddle_transcription_enabled( drop(old_stt); if enabled { - let eph_id = - ephemeral_channel_id.ok_or("no active huddle — start or join a huddle first")?; if let Some(manager) = models::global_model_manager() { manager.start_stt_download(state.http_client.clone()); } - if let Err(e) = maybe_start_stt_pipeline(&state, &eph_id).await { + if let Err(e) = maybe_start_stt_pipeline(&state, &ephemeral_channel_id).await { eprintln!("buzz-desktop: STT transcript start failed: {e}"); } } diff --git a/desktop/src/features/channels/ui/ChannelMembersBar.tsx b/desktop/src/features/channels/ui/ChannelMembersBar.tsx index c87617ce9b..7b9bf2b79f 100644 --- a/desktop/src/features/channels/ui/ChannelMembersBar.tsx +++ b/desktop/src/features/channels/ui/ChannelMembersBar.tsx @@ -11,9 +11,15 @@ import { useManagedAgentsQuery, useRelayAgentsQuery, } from "@/features/agents/hooks"; +import { mergeChannelKnownAgentPubkeys } from "@/features/agents/knownAgentPubkeys"; import { requestOpenCreateAgent } from "@/features/agents/openCreateAgentEvent"; import { useChannelMembersQuery } from "@/features/channels/hooks"; +import { + getDmHuddleMemberPubkeys, + hasOtherDmParticipant, +} from "@/features/channels/lib/dmHuddleMembers"; import { canStartHuddleInChannel } from "@/features/channels/lib/huddleAvailability"; +import { useUsersBatchQuery } from "@/features/profile/hooks"; import type { Channel } from "@/shared/api/types"; import { normalizePubkey } from "@/shared/lib/pubkey"; import { Button } from "@/shared/ui/button"; @@ -64,6 +70,41 @@ export function ChannelMembersBar({ const managedAgentsQuery = useManagedAgentsQuery(); const relayAgentsQuery = useRelayAgentsQuery(); const members = membersQuery.data ?? []; + const dmProfilesQuery = useUsersBatchQuery( + channel.channelType === "dm" ? channel.participantPubkeys : [], + { enabled: channel.channelType === "dm" }, + ); + const huddleAgentPubkeys = React.useMemo(() => { + const pubkeys = new Set( + mergeChannelKnownAgentPubkeys( + membersQuery.data, + managedAgentsQuery.data, + relayAgentsQuery.data, + ), + ); + for (const [pubkey, profile] of Object.entries( + dmProfilesQuery.data?.profiles ?? {}, + )) { + if (profile.isAgent) pubkeys.add(normalizePubkey(pubkey)); + } + return pubkeys; + }, [ + dmProfilesQuery.data?.profiles, + managedAgentsQuery.data, + membersQuery.data, + relayAgentsQuery.data, + ]); + const huddleMemberPubkeys = React.useMemo( + () => getDmHuddleMemberPubkeys(channel, huddleAgentPubkeys, currentPubkey), + [channel, currentPubkey, huddleAgentPubkeys], + ); + const huddleMemberPubkeysPending = + hasOtherDmParticipant(channel, currentPubkey) && + (membersQuery.isPending || + managedAgentsQuery.isPending || + relayAgentsQuery.isPending || + dmProfilesQuery.isPending || + dmProfilesQuery.isPlaceholderData); const memberCount = membersQuery.data?.length ?? channel.memberCount; const providers = React.useMemo( () => @@ -117,7 +158,7 @@ export function ChannelMembersBar({ try { await startHuddle( channel.id, - [], + [...huddleMemberPubkeys], buildHuddleChannelName({ channel, currentPubkey, @@ -133,7 +174,9 @@ export function ChannelMembersBar({ } }} renderMode={variant === "compact" ? "menu-item" : "button"} - startDisabled={!canStartHuddle || isStartingHuddle} + startDisabled={ + !canStartHuddle || isStartingHuddle || huddleMemberPubkeysPending + } /> ); diff --git a/desktop/src/features/huddle/HuddleContext.tsx b/desktop/src/features/huddle/HuddleContext.tsx index 804eb4bc0e..03620838a3 100644 --- a/desktop/src/features/huddle/HuddleContext.tsx +++ b/desktop/src/features/huddle/HuddleContext.tsx @@ -339,6 +339,28 @@ export function HuddleProvider({ children }: { children: React.ReactNode }) { [], ); + /** + * Clean up only this provider's media after its start token is superseded. + * The action that changed the token owns backend teardown; issuing a global + * leave here could terminate a replacement huddle started by a new provider. + */ + const cleanupSupersededStart = React.useCallback( + (worklet: AudioWorkletHandle | null) => { + try { + worklet?.stop(); + } catch { + /* best-effort */ + } + workletRef.current = null; + rustActiveRef.current = false; + setLocalAudioTrack(null); + setMicConnected(false); + setEphemeralChannelId(null); + setActiveSpeakers([]); + }, + [], + ); + /** Shared media setup: get mic, setup AudioWorklet, confirm active. * Used by both startHuddle and joinHuddle after the Rust backend call succeeds. */ const connectAndSetupMedia = React.useCallback( @@ -443,7 +465,7 @@ export function HuddleProvider({ children }: { children: React.ReactNode }) { await connectAndSetupMedia(joinInfo, myToken); } catch (e) { if (e instanceof Error && e.message === "superseded") { - await cleanupFailedStart(workletRef.current, true); + cleanupSupersededStart(workletRef.current); return; } throw e; @@ -466,7 +488,7 @@ export function HuddleProvider({ children }: { children: React.ReactNode }) { busyRef.current = false; } }, - [cleanupFailedStart, connectAndSetupMedia], + [cleanupFailedStart, cleanupSupersededStart, connectAndSetupMedia], ); const joinHuddle = React.useCallback( @@ -489,7 +511,7 @@ export function HuddleProvider({ children }: { children: React.ReactNode }) { await connectAndSetupMedia(joinInfo, myToken); } catch (e) { if (e instanceof Error && e.message === "superseded") { - await cleanupFailedStart(workletRef.current, false); + cleanupSupersededStart(workletRef.current); return; } throw e; @@ -512,7 +534,7 @@ export function HuddleProvider({ children }: { children: React.ReactNode }) { busyRef.current = false; } }, - [cleanupFailedStart, connectAndSetupMedia], + [cleanupFailedStart, cleanupSupersededStart, connectAndSetupMedia], ); useTtsSubscription(ephemeralChannelId, selfPubkeyRef); diff --git a/desktop/src/features/huddle/components/HuddleBar.tsx b/desktop/src/features/huddle/components/HuddleBar.tsx index 35d660e475..758726cf58 100644 --- a/desktop/src/features/huddle/components/HuddleBar.tsx +++ b/desktop/src/features/huddle/components/HuddleBar.tsx @@ -242,7 +242,10 @@ export function HuddleBar({ // Primary: listen for Rust-emitted state change events listen("huddle-state-changed", (event) => { - if (!cancelled) applyIncomingState(event.payload); + if (!cancelled) { + stateGenerationRef.current += 1; + applyIncomingState(event.payload); + } }).then((fn) => { if (cancelled) fn(); else unlisten = fn; @@ -757,14 +760,11 @@ export function HuddleBar({ transcriptionEnabled ? "Stop transcript" : "Start transcript" } aria-pressed={transcriptionEnabled} - className={cn( - "buzz-huddle-control-button h-12 w-12 shrink-0 rounded-md", - transcriptionEnabled && "text-foreground", - )} + className="buzz-huddle-control-button h-12 w-12 shrink-0 rounded-md" onClick={() => void handleToggleTranscript()} size="icon" type="button" - variant={transcriptionEnabled ? "secondary" : "ghost"} + variant="ghost" > diff --git a/desktop/src/features/profile/ui/UserProfilePopover.tsx b/desktop/src/features/profile/ui/UserProfilePopover.tsx index f2739088a6..da51ab1b88 100644 --- a/desktop/src/features/profile/ui/UserProfilePopover.tsx +++ b/desktop/src/features/profile/ui/UserProfilePopover.tsx @@ -260,11 +260,18 @@ export function UserProfilePopover({ const selfProfileQuery = useProfileQuery(open && showProfileActions); const isCurrentUserOwner = ownsAuthorAgent(profile, currentPubkey); const viewerIsOwner = isCurrentUserOwner || isOwner === true; + const showHuddleAction = + showHumanProfileActions || + (showProfileActions && + isBotProfile && + viewerIsOwner && + !isAgentClassificationPending); const showMessageAction = showProfileActions && !isAgentClassificationPending && (!isBotProfile || viewerIsOwner); - const showAnyProfileActions = showHumanProfileActions || showMessageAction; + const showAnyProfileActions = + showHumanProfileActions || showMessageAction || showHuddleAction; const canViewActivity = isBotProfile && viewerIsOwner && canOpenAgentActivity(pubkey); const presenceStatus = presenceQuery.data?.[pubkey.toLowerCase()]; @@ -356,7 +363,7 @@ export function UserProfilePopover({ const handleHuddle = React.useCallback(async () => { if ( !showProfileActions || - !showHumanProfileActions || + !showHuddleAction || pendingAction !== null || isStartingHuddle ) { @@ -369,7 +376,7 @@ export function UserProfilePopover({ try { const dm = await openDmMutation.mutateAsync({ pubkeys: [pubkey] }); await goChannel(dm.id); - await startHuddle(dm.id, []); + await startHuddle(dm.id, isBotProfile ? [pubkey] : []); await queryClient.invalidateQueries({ queryKey: channelsQueryKey }); if (isMountedRef.current) { setOpen(false); @@ -389,7 +396,8 @@ export function UserProfilePopover({ pendingAction, pubkey, queryClient, - showHumanProfileActions, + isBotProfile, + showHuddleAction, showProfileActions, startHuddle, ]); @@ -722,7 +730,7 @@ export function UserProfilePopover({ Message ) : null} - {showHumanProfileActions ? ( + {showHuddleAction ? (