diff --git a/desktop/src-tauri/src/huddle/agent_tts_publisher.rs b/desktop/src-tauri/src/huddle/agent_tts_publisher.rs index a1d4269266..ff47db7c58 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_publisher.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_publisher.rs @@ -2,7 +2,7 @@ use std::sync::Arc; -use super::{relay_api, tts}; +use super::{relay_api, roster, tts}; use crate::app_state::AppState; pub(super) async fn ensure( @@ -51,15 +51,13 @@ pub(super) async fn ensure( Arc::clone(&huddle.local_tts_publishers), ) }; - let has_bot_membership = - relay_api::fetch_channel_members_with_roles(&ephemeral_channel_id, state) - .await? - .into_iter() - .any(|(pubkey, role)| { - pubkey.eq_ignore_ascii_case(speaker_pubkey) && role.as_deref() == Some("bot") - }); - if !has_bot_membership { - return Err("agent is not an active bot member of the Huddle".to_string()); + let is_huddle_agent = roster::fetch_huddle_roster(&ephemeral_channel_id, state) + .await? + .agents + .iter() + .any(|pubkey| pubkey.eq_ignore_ascii_case(speaker_pubkey)); + if !is_huddle_agent { + return Err("agent is not an active agent member of the Huddle".to_string()); } let publisher = relay_api::connect_tts_audio_publisher( &ephemeral_channel_id, diff --git a/desktop/src-tauri/src/huddle/agents.rs b/desktop/src-tauri/src/huddle/agents.rs index 3213badf89..9af624e992 100644 --- a/desktop/src-tauri/src/huddle/agents.rs +++ b/desktop/src-tauri/src/huddle/agents.rs @@ -18,10 +18,8 @@ use uuid::Uuid; use crate::{ app_state::AppState, events, - huddle::relay_api::{ - fetch_channel_members, fetch_channel_members_with_roles, validate_pubkey_hex, - MAX_HUDDLE_AGENTS, - }, + huddle::relay_api::{fetch_channel_members_with_roles, validate_pubkey_hex, MAX_HUDDLE_AGENTS}, + huddle::roster::fetch_huddle_roster, relay::submit_event, }; @@ -216,8 +214,9 @@ pub(crate) async fn sync_agents_for_active_huddle( // Membership reads can lag a just-accepted write, so merge the relay view // with local state instead of allowing a stale snapshot to remove agents. - let fresh_agents = fetch_channel_members(&ephemeral_channel_id, Some("bot"), state) + let fresh_agents = fetch_huddle_roster(&ephemeral_channel_id, state) .await + .map(|roster| roster.agents) .unwrap_or_default(); let mut known_agents = HashSet::new(); let mut merged_agents = Vec::new(); diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index e219b2f75f..554ac649e0 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -41,6 +41,7 @@ pub mod pocket; pub mod preprocessing; pub mod reconnect; pub mod relay_api; +mod roster; pub mod state; pub mod stt; pub mod transcription; @@ -99,12 +100,10 @@ use agent_tts_routing::{ 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, + post_connect_setup, start_huddle_media, PostConnectOutcome, }; +use relay_api::{parse_channel_uuid, validate_pubkey_hex, MAX_HUDDLE_AGENTS}; +use roster::fetch_huddle_roster; use window::close_huddle_window; fn normalize_huddle_channel_name(candidate: Option, fallback: &str) -> String { @@ -332,7 +331,14 @@ 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. - match post_connect_setup(&state, &ephemeral_channel_id, huddle_generation).await { + match post_connect_setup( + &state, + &ephemeral_channel_id, + huddle_generation, + start_huddle_media, + ) + .await + { Ok(PostConnectOutcome::Ready) => {} Ok(PostConnectOutcome::Stale) => { close_huddle_window(&app, &ephemeral_channel_id); @@ -455,7 +461,14 @@ 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. - match post_connect_setup(&state, &ephemeral_channel_id, huddle_generation).await { + match post_connect_setup( + &state, + &ephemeral_channel_id, + huddle_generation, + start_huddle_media, + ) + .await + { Ok(PostConnectOutcome::Ready) => {} Ok(PostConnectOutcome::Stale) => { return Err("huddle join was superseded".to_owned()); @@ -566,9 +579,8 @@ async fn remove_huddle_agents(ephemeral_channel_id: &str, state: &AppState) { return; }; - let agent_pubkeys = match fetch_channel_members(ephemeral_channel_id, Some("bot"), state).await - { - Ok(pubkeys) => pubkeys, + let agent_pubkeys = match fetch_huddle_roster(ephemeral_channel_id, state).await { + Ok(roster) => roster.agents, Err(e) => { eprintln!("buzz-desktop: fetch huddle agents for cleanup failed: {e}"); return; @@ -614,9 +626,12 @@ pub async fn leave_huddle(app: tauri::AppHandle, state: State<'_, AppState>) -> // 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) + let humans_remaining = fetch_huddle_roster(&ephemeral_channel_id, &state) .await - // On fetch failure, assume 2 humans remain (safe default). + .ok() + .and_then(|roster| roster.humans_remaining()) + // On fetch failure or an incomplete roster, 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 @@ -729,7 +744,9 @@ pub async fn get_huddle_agent_pubkeys(state: State<'_, AppState>) -> Result fetch_channel_members(&id, Some("bot"), &state).await, + Some(id) => fetch_huddle_roster(&id, &state) + .await + .map(|roster| roster.agents), None => Ok(Vec::new()), } } diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 47d4aeb43d..1649a0a5e3 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -4,6 +4,7 @@ //! the voice pipelines. Extracted from mod.rs to keep the command layer thin. use std::{ + ops::AsyncFnOnce, sync::{ atomic::{AtomicU64, Ordering}, Arc, Mutex, @@ -20,6 +21,7 @@ use crate::events; use super::models; use super::relay_api::{self, fetch_channel_members, parse_channel_uuid}; +use super::roster::fetch_huddle_roster; use super::state::{HuddlePhase, HuddleState, VoiceInputMode}; use super::stt; use super::tts; @@ -36,6 +38,15 @@ pub(crate) enum PostConnectOutcome { /// the huddle is not active or pipelines are already running. #[tauri::command] pub async fn check_pipeline_hotstart(state: State<'_, AppState>) -> Result<(), String> { + pipeline_hotstart(&state, start_ready_pipelines).await +} + +/// Hot-start body. `start_pipelines` starts the voice pipelines whose models +/// are ready; tests replace it to keep model files and audio out of the run. +async fn pipeline_hotstart( + state: &AppState, + start_pipelines: impl AsyncFnOnce(&AppState, Option<&str>) -> Result<(), String>, +) -> Result<(), String> { let (is_active, ephemeral_channel_id, huddle_generation) = { let hs = state.huddle()?; ( @@ -49,6 +60,40 @@ pub async fn check_pipeline_hotstart(state: State<'_, AppState>) -> Result<(), S return Ok(()); } + start_pipelines(state, ephemeral_channel_id.as_deref()).await?; + + // 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 a complete roster: replace (even with empty — agents may have been + // removed). On an incomplete roster (managed-agent records unreadable): + // add the role-proven agents and keep stored agents that are still + // members. On Err: preserve the existing list (transient failure shouldn't + // zero it). + if let Some(eph_id) = &ephemeral_channel_id { + refresh_huddle_roster(state, eph_id, huddle_generation).await?; + } + + Ok(()) +} + +/// Start the STT and TTS pipelines whose models are ready, and clear handles +/// of pipelines whose worker has exited. +async fn start_ready_pipelines( + state: &AppState, + ephemeral_channel_id: Option<&str>, +) -> Result<(), String> { // 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. { @@ -84,127 +129,104 @@ pub async fn check_pipeline_hotstart(state: State<'_, AppState>) -> Result<(), S // Start TTS first so STT can observe its active-playback gate. if !has_tts && (tts_ready || models::is_tts_ready()) { - if let Err(e) = maybe_start_tts_pipeline(&state).await { + 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 { + 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 (roster_changed, 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(()); - } - let mut roster_changed = false; - if let Some(agents) = fresh_agents { - let mut current_agents = - hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); - if *current_agents != agents { - *current_agents = agents; - roster_changed = true; - } - } - if let Some(members) = fresh_members { - if hs.participants != members { - hs.participants = members; - roster_changed = true; - } - } - hs.last_agent_refresh = Some(std::time::Instant::now()); - ( - roster_changed, - hs.maybe_auto_enable_transcription_for_agents(), - ) - } else { - (false, false) - }; - if transcription_auto_enabled { - start_auto_enabled_transcription(&state, eph_id).await; - } - // Audio authentication auto-adds a joining human to the ephemeral - // channel. Emit whenever that authoritative roster changes so the - // desktop participant strip updates immediately instead of waiting - // for its slow fallback IPC read. - if roster_changed || transcription_auto_enabled { - state.emit_huddle_state_changed(); - } - } - } - Ok(()) } -pub(crate) async fn post_connect_setup( +/// Periodic roster refresh for an active Huddle, at most every 15 s: store +/// the agents and participants read from the relay. +async fn refresh_huddle_roster( + state: &AppState, + eph_id: &str, + huddle_generation: u64, +) -> Result<(), String> { + 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_huddle_roster(eph_id, state).await.ok(); + let fresh_members = fetch_channel_members(eph_id, state).await.ok(); + let (roster_changed, 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(()); + } + let mut roster_changed = false; + if let Some(roster) = fresh_agents { + let mut current_agents = hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); + roster_changed |= roster.apply_agents(&mut current_agents); + } + if let Some(members) = fresh_members { + if hs.participants != members { + hs.participants = members; + roster_changed = true; + } + } + hs.last_agent_refresh = Some(std::time::Instant::now()); + ( + roster_changed, + hs.maybe_auto_enable_transcription_for_agents(), + ) + } else { + (false, false) + }; + if transcription_auto_enabled { + start_auto_enabled_transcription(state, eph_id).await; + } + // Audio authentication auto-adds a joining human to the ephemeral + // channel. Emit whenever that authoritative roster changes so the + // desktop participant strip updates immediately instead of waiting + // for its slow fallback IPC read. + if roster_changed || transcription_auto_enabled { + state.emit_huddle_state_changed(); + } + } + Ok(()) +} + +/// First roster read after connecting: store the agents and participants read +/// from the relay. Returns false when the Huddle is no longer current. +async fn hydrate_huddle_roster( state: &AppState, ephemeral_channel_id: &str, huddle_generation: u64, -) -> Result { - { - let hs = state.huddle()?; - if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { - return Ok(PostConnectOutcome::Stale); - } - } - +) -> Result { // 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), + async { fetch_huddle_roster(ephemeral_channel_id, state).await.ok() }, + fetch_channel_members(ephemeral_channel_id, state), ); let (roster_changed, transcription_auto_enabled) = { let mut hs = state.huddle()?; if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { - return Ok(PostConnectOutcome::Stale); + return Ok(false); } let mut roster_changed = false; - if let Ok(agents) = agents_result { + if let Some(roster) = agents_result { let mut current_agents = hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); - if *current_agents != agents { - *current_agents = agents; - roster_changed = true; - } + roster_changed |= roster.apply_agents(&mut current_agents); } if let Ok(all_members) = all_members_result { if !all_members.is_empty() && hs.participants != all_members { @@ -221,7 +243,39 @@ pub(crate) async fn post_connect_setup( if roster_changed || transcription_auto_enabled { state.emit_huddle_state_changed(); } + Ok(true) +} +/// Setup after the Huddle connects: read the roster, then run `start_media`. +/// Callers pass [`start_huddle_media`]; tests replace it to keep model files +/// and audio out of the run. +pub(crate) async fn post_connect_setup( + state: &AppState, + ephemeral_channel_id: &str, + huddle_generation: u64, + start_media: impl AsyncFnOnce(&AppState, &str, u64) -> Result, +) -> Result { + { + let hs = state.huddle()?; + if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) { + return Ok(PostConnectOutcome::Stale); + } + } + + if !hydrate_huddle_roster(state, ephemeral_channel_id, huddle_generation).await? { + return Ok(PostConnectOutcome::Stale); + } + + start_media(state, ephemeral_channel_id, huddle_generation).await +} + +/// Download voice models, connect the audio relay, and start the voice +/// pipelines. Audio relay failure is an error. +pub(crate) async fn start_huddle_media( + state: &AppState, + ephemeral_channel_id: &str, + huddle_generation: u64, +) -> Result { // Prepare voice models. Agent presence may have auto-enabled transcription; // explicit user choices remain authoritative. if let Some(mgr) = models::global_model_manager() { @@ -735,6 +789,10 @@ pub(crate) fn spawn_transcription_task( }); } +#[cfg(test)] +#[path = "roster_hydration_tests.rs"] +mod roster_hydration_tests; + #[cfg(test)] mod tts_start_race_tests { use std::sync::{ diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 53c757b986..67f205c2db 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -488,30 +488,14 @@ pub(crate) async fn fetch_channel_members_with_roles( Ok(members) } -/// Fetch channel members, optionally filtered by role (e.g., "bot" for agents). +/// Fetch all channel member pubkeys. Use `roster::fetch_huddle_roster` to +/// tell agents from people. pub(crate) async fn fetch_channel_members( channel_id: &str, - role_filter: Option<&str>, state: &AppState, ) -> Result, String> { let all = fetch_channel_members_with_roles(channel_id, state).await?; - Ok(all - .into_iter() - .filter(|(_, role)| role_filter.is_none_or(|r| role.as_deref() == Some(r))) - .map(|(pubkey, _)| pubkey) - .collect()) -} - -/// Count human (non-bot) members remaining in a channel. -pub(crate) async fn count_human_members( - channel_id: &str, - state: &AppState, -) -> Result { - let all = fetch_channel_members_with_roles(channel_id, state).await?; - Ok(all - .iter() - .filter(|(_, role)| role.as_deref() != Some("bot")) - .count()) + Ok(all.into_iter().map(|(pubkey, _)| pubkey).collect()) } #[cfg(test)] diff --git a/desktop/src-tauri/src/huddle/roster.rs b/desktop/src-tauri/src/huddle/roster.rs new file mode 100644 index 0000000000..3c045f09d3 --- /dev/null +++ b/desktop/src-tauri/src/huddle/roster.rs @@ -0,0 +1,331 @@ +//! Which Huddle members are agents. +//! +//! The channel role `bot` is the first signal. The relay stores it only for +//! keys with a recorded NIP-OA owner, so an agent without an owner +//! attestation joins as `member`. Desktop still knows such an agent when a +//! trusted owner published a kind:30177 managed-agent record for it. Desktop +//! publishes that record for every agent it manages, from every machine. +//! +//! Trusted record authors are the current user and the Huddle channel's +//! owners. A record from anyone else is only that author's claim, so it does +//! not make a member an agent. This is a Huddle compatibility policy, not +//! identity proof: a record mints no owner. It does let an owner mark a +//! member as an agent for every Huddle behavior that reads this roster: +//! spoken text, the voice publisher, remote transcription and floor +//! arbitration (which skip agent audio), cleanup, and the last-person check. +//! An owner already had that power when the client chose the `bot` role. +//! +//! A participant who does not own the channel can still add an agent without +//! NIP-OA. Only that participant's own machines recognize it. + +use std::collections::HashSet; + +use nostr::Event; + +use super::relay_api::fetch_channel_members_with_roles; +use crate::{app_state::AppState, nostr_convert, relay::query_relay}; + +const KIND_MANAGED_AGENT: u16 = 30177; + +/// Huddle channel members, split into agents and people. +/// +/// When the managed-agent records cannot be read, `complete` is false. Then +/// `agents` holds only the members proven by role `bot`, and `humans` can +/// include agents without NIP-OA. +#[derive(Debug, Default, PartialEq, Eq)] +pub(crate) struct HuddleRoster { + pub(crate) agents: Vec, + pub(crate) humans: Vec, + pub(crate) complete: bool, +} + +impl HuddleRoster { + /// Update a stored agent list from this roster. Returns true when the + /// list changed. + /// + /// A complete roster replaces the list. An incomplete roster adds its + /// role-proven agents and keeps stored agents that are still members, so + /// a failed record read can neither hide a `bot` agent nor drop an agent + /// that only a record proved. Stored agents that left the channel drop out. + pub(crate) fn apply_agents(self, current: &mut Vec) -> bool { + let next = if self.complete { + self.agents + } else { + let mut next: Vec = current + .iter() + .filter(|pubkey| self.agents.contains(pubkey) || self.humans.contains(pubkey)) + .cloned() + .collect(); + for agent in self.agents { + if !next.contains(&agent) { + next.push(agent); + } + } + next + }; + if *current == next { + return false; + } + *current = next; + true + } + + /// The number of people, only when every member was classified. An + /// incomplete roster can count an agent as a person, so it must not + /// decide that the Huddle can end. + pub(crate) fn humans_remaining(&self) -> Option { + self.complete.then_some(self.humans.len()) + } +} + +/// Read the Huddle channel's members and decide which are agents. +pub(crate) async fn fetch_huddle_roster( + channel_id: &str, + state: &AppState, +) -> Result { + let members = fetch_channel_members_with_roles(channel_id, state).await?; + let viewer = state + .keys + .lock() + .map(|keys| keys.public_key().to_hex()) + .map_err(|error| error.to_string())?; + let authors = trusted_record_authors(&members, &viewer); + let candidates = record_candidates(&members); + let records = if candidates.is_empty() { + Ok(Vec::new()) + } else { + query_relay( + state, + &[serde_json::json!({ + "kinds": [KIND_MANAGED_AGENT], + "authors": authors, + "#d": candidates, + "limit": authors.len() * candidates.len(), + })], + ) + .await + }; + Ok(roster_from(&members, records, &authors)) +} + +/// Classify members. A failed record read keeps the role-proven agents and +/// marks the roster incomplete. +fn roster_from( + members: &[(String, Option)], + records: Result, String>, + trusted_authors: &[String], +) -> HuddleRoster { + match records { + Ok(records) => classify(members, &records, trusted_authors), + Err(error) => { + eprintln!("buzz-desktop: Huddle managed-agent query failed: {error}"); + HuddleRoster { + complete: false, + ..classify(members, &[], trusted_authors) + } + } + } +} + +/// The current user and the channel's owners, normalized and deduplicated. +fn trusted_record_authors(members: &[(String, Option)], viewer: &str) -> Vec { + let mut authors = vec![viewer.to_ascii_lowercase()]; + for (pubkey, role) in members { + let pubkey = pubkey.to_ascii_lowercase(); + if role.as_deref() == Some("owner") && !authors.contains(&pubkey) { + authors.push(pubkey); + } + } + authors +} + +/// Members without the `bot` role. Only these need a managed-agent record. +fn record_candidates(members: &[(String, Option)]) -> Vec { + members + .iter() + .filter(|(_, role)| role.as_deref() != Some("bot")) + .map(|(pubkey, _)| pubkey.to_ascii_lowercase()) + .collect() +} + +fn classify( + members: &[(String, Option)], + records: &[Event], + trusted_authors: &[String], +) -> HuddleRoster { + let trusted: Vec = records + .iter() + .filter(|event| { + trusted_authors + .iter() + .any(|author| author.eq_ignore_ascii_case(&event.pubkey.to_hex())) + }) + .cloned() + .collect(); + let managed: HashSet = nostr_convert::managed_agent_pubkeys_from_events(&trusted); + let mut roster = HuddleRoster { + complete: true, + ..HuddleRoster::default() + }; + for (pubkey, role) in members { + if role.as_deref() == Some("bot") || managed.contains(&pubkey.to_ascii_lowercase()) { + roster.agents.push(pubkey.clone()); + } else { + roster.humans.push(pubkey.clone()); + } + } + roster +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::{EventBuilder, JsonUtil, Keys, Kind, Tag}; + + fn record(owner: &Keys, agent: &str) -> Event { + EventBuilder::new(Kind::Custom(KIND_MANAGED_AGENT), "{}") + .tags([Tag::parse(["d", agent]).unwrap()]) + .sign_with_keys(owner) + .unwrap() + } + + fn member(pubkey: &str, role: &str) -> (String, Option) { + (pubkey.to_string(), Some(role.to_string())) + } + + #[test] + fn bot_role_and_trusted_records_mark_agents() { + let viewer = Keys::generate(); + let creator = Keys::generate(); + let stranger = Keys::generate(); + let human = Keys::generate().public_key().to_hex(); + let attested = Keys::generate().public_key().to_hex(); + let mine = Keys::generate().public_key().to_hex(); + let creators = Keys::generate().public_key().to_hex(); + let claimed = Keys::generate().public_key().to_hex(); + let members = vec![ + member(&creator.public_key().to_hex(), "owner"), + member(&viewer.public_key().to_hex(), "member"), + member(&human, "member"), + member(&attested, "bot"), + member(&mine, "member"), + member(&creators, "member"), + member(&claimed, "member"), + (stranger.public_key().to_hex(), None), + ]; + let authors = trusted_record_authors(&members, &viewer.public_key().to_hex()); + let records = vec![ + record(&viewer, &mine), + record(&creator, &creators), + // A member who does not own the channel cannot vouch. + record(&stranger, &claimed), + ]; + + let roster = classify(&members, &records, &authors); + + assert_eq!(roster.agents, vec![attested, mine, creators]); + assert_eq!( + roster.humans, + vec![ + creator.public_key().to_hex(), + viewer.public_key().to_hex(), + human, + claimed, + stranger.public_key().to_hex(), + ] + ); + } + + #[test] + fn untrusted_or_forged_records_are_ignored() { + let viewer = Keys::generate(); + let other = Keys::generate(); + let agent = Keys::generate().public_key().to_hex(); + let members = vec![member(&agent, "member")]; + let authors = trusted_record_authors(&members, &viewer.public_key().to_hex()); + + // The relay may return an event from an author outside the filter. + let untrusted = record(&other, &agent); + assert!(classify(&members, &[untrusted], &authors).agents.is_empty()); + + // A record whose signature does not verify is not the viewer's claim. + let mut forged: serde_json::Value = + serde_json::from_str(&record(&other, &agent).as_json()).unwrap(); + forged["pubkey"] = serde_json::json!(viewer.public_key().to_hex()); + let forged: Event = serde_json::from_value(forged).unwrap(); + assert!(classify(&members, &[forged], &authors).agents.is_empty()); + } + + #[test] + fn failed_record_read_keeps_bot_agents_and_blocks_auto_end() { + let viewer = Keys::generate(); + let attested = Keys::generate().public_key().to_hex(); + let legacy = Keys::generate().public_key().to_hex(); + let members = vec![ + member(&viewer.public_key().to_hex(), "owner"), + member(&attested, "bot"), + member(&legacy, "member"), + ]; + let authors = trusted_record_authors(&members, &viewer.public_key().to_hex()); + + let failed = roster_from(&members, Err("relay down".to_string()), &authors); + assert_eq!(failed.agents, vec![attested.clone()]); + assert!(!failed.complete); + // Viewer + legacy agent look like two people, but the roster is + // incomplete, so the leave path must not treat this as "last person". + assert_eq!(failed.humans_remaining(), None); + + // The next successful read classifies the legacy agent again. + let recovered = roster_from(&members, Ok(vec![record(&viewer, &legacy)]), &authors); + assert_eq!(recovered.humans_remaining(), Some(1)); + assert_eq!(recovered.agents, vec![attested, legacy]); + } + + #[test] + fn failed_record_read_adds_bot_agents_to_stored_state() { + let viewer = Keys::generate(); + let attested = Keys::generate().public_key().to_hex(); + let legacy = Keys::generate().public_key().to_hex(); + let members = vec![ + member(&viewer.public_key().to_hex(), "owner"), + member(&attested, "bot"), + member(&legacy, "member"), + ]; + let authors = trusted_record_authors(&members, &viewer.public_key().to_hex()); + let failed = || roster_from(&members, Err("relay down".to_string()), &authors); + + // A joiner starts with no stored agents. The role-proven agent must + // still reach state, while the roster cannot end the Huddle. + let mut joiner = Vec::new(); + assert_eq!(failed().humans_remaining(), None); + assert!(failed().apply_agents(&mut joiner)); + assert_eq!(joiner, vec![attested.clone()]); + + // An agent that only a record proved stays through a failed read. + let mut stored = vec![legacy.clone(), attested.clone()]; + assert!(!failed().apply_agents(&mut stored)); + assert_eq!(stored, vec![legacy.clone(), attested.clone()]); + + // A stored agent that left the channel drops out. + let departed = Keys::generate().public_key().to_hex(); + let mut stale = vec![departed, legacy.clone()]; + assert!(failed().apply_agents(&mut stale)); + assert_eq!(stale, vec![legacy.clone(), attested.clone()]); + + // A complete roster replaces the list. + let complete = roster_from(&members, Ok(Vec::new()), &authors); + assert!(complete.apply_agents(&mut stored)); + assert_eq!(stored, vec![attested]); + } + + #[test] + fn candidates_skip_bot_members() { + let members = vec![ + member("aa", "bot"), + member("BB", "member"), + member("cc", "owner"), + ]; + assert_eq!(record_candidates(&members), vec!["bb", "cc"]); + assert_eq!(trusted_record_authors(&members, "DD"), vec!["dd", "cc"]); + } +} diff --git a/desktop/src-tauri/src/huddle/roster_hydration_tests.rs b/desktop/src-tauri/src/huddle/roster_hydration_tests.rs new file mode 100644 index 0000000000..101c52348f --- /dev/null +++ b/desktop/src-tauri/src/huddle/roster_hydration_tests.rs @@ -0,0 +1,132 @@ +//! Huddle roster hydration through the post-connect and hot-start entrypoints, +//! with the production relay reads, when the managed-agent records cannot be +//! read. The voice-model and audio steps are replaced. +use super::*; +use axum::{extract::State as AxumState, http::StatusCode, routing::post, Json, Router}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; + +use crate::app_state::build_app_state; + +const CHANNEL: &str = "4b1d1f0a-6a7e-4c47-9a49-2f3c55c0b0a1"; + +struct Relay { + url: String, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Relay { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Serves one kind:39002 roster and fails every kind:30177 record read. +async fn relay(roster: nostr::Event) -> Relay { + let router = + Router::new() + .route( + "/query", + post( + |AxumState(roster): AxumState, + Json(filters): Json>| async move { + if filters.iter().any(|f| f["kinds"] == json!([30177])) { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"error": "records unavailable"})), + ); + } + (StatusCode::OK, Json(json!([roster]))) + }, + ), + ) + .with_state(roster); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + Relay { url, task } +} + +/// An active Huddle with no known agents, whose relay roster has the viewer, +/// a `bot` agent and an ordinary member. +async fn huddle() -> (Relay, AppState, u64, String, String) { + let relay_keys = Keys::generate(); + let viewer = Keys::generate(); + let bot = Keys::generate().public_key().to_hex(); + let member = Keys::generate().public_key().to_hex(); + let viewer_hex = viewer.public_key().to_hex(); + let tags = [ + vec!["d", CHANNEL], + vec!["p", &viewer_hex, "", "owner"], + vec!["p", &bot, "", "bot"], + vec!["p", &member, "", "member"], + ]; + let roster = EventBuilder::new(Kind::from_u16(39002), "") + .tags(tags.into_iter().map(|tag| Tag::parse(tag).unwrap())) + .sign_with_keys(&relay_keys) + .unwrap(); + let relay = relay(roster).await; + let state = build_app_state(); + *state.keys.lock().unwrap() = viewer; + *state.relay_url_override.lock().unwrap() = Some(relay.url.clone()); + let generation = { + let mut hs = state.huddle().unwrap(); + let generation = hs.begin_huddle_lifetime(); + hs.phase = HuddlePhase::Active; + hs.ephemeral_channel_id = Some(CHANNEL.to_owned()); + // An explicit choice keeps the refresh from starting transcription. + hs.set_transcription_enabled_by_user(false); + generation + }; + (relay, state, generation, bot, member) +} + +fn agents(state: &AppState) -> Vec { + state + .huddle() + .unwrap() + .agent_pubkeys + .lock() + .unwrap() + .clone() +} + +#[tokio::test] +async fn post_connect_keeps_bot_agents_when_records_are_unreadable() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + let (_relay, state, generation, bot, _member) = huddle().await; + + let mut media_started = false; + let outcome = post_connect_setup(&state, CHANNEL, generation, async |state, _, _| { + // Media setup reads the agents, so they must be stored first. + assert_eq!(agents(state), std::slice::from_ref(&bot)); + media_started = true; + Ok(PostConnectOutcome::Ready) + }) + .await + .unwrap(); + assert!(matches!(outcome, PostConnectOutcome::Ready)); + assert!(media_started); + assert_eq!(agents(&state), [bot]); + assert_eq!(state.huddle().unwrap().participants.len(), 3); + + // The same incomplete roster must not decide that the Huddle can end. + let roster = fetch_huddle_roster(CHANNEL, &state).await.unwrap(); + assert!(!roster.complete); + assert_eq!(roster.humans_remaining(), None); +} + +#[tokio::test] +async fn hotstart_refresh_keeps_known_agents_when_records_are_unreadable() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + let departed = Keys::generate().public_key().to_hex(); + let (_relay, state, _generation, bot, member) = huddle().await; + // `member` was proven an agent by an earlier record read. + *state.huddle().unwrap().agent_pubkeys.lock().unwrap() = vec![member.clone(), departed]; + + pipeline_hotstart(&state, async |_, _| Ok(())) + .await + .unwrap(); + assert_eq!(agents(&state), [member, bot]); + assert!(state.huddle().unwrap().last_agent_refresh.is_some()); +} diff --git a/desktop/src/features/huddle/lib/useTtsSubscription.ts b/desktop/src/features/huddle/lib/useTtsSubscription.ts index 5b3a844698..2e15636488 100644 --- a/desktop/src/features/huddle/lib/useTtsSubscription.ts +++ b/desktop/src/features/huddle/lib/useTtsSubscription.ts @@ -48,8 +48,9 @@ export function useTtsSubscription( // ── Agent identity (authoritative, fail-closed) ─────────────────────── // - // Fetch the ephemeral channel's member list from the relay REST API and - // identify agents by their "bot" role. This is authoritative — it works + // Fetch the ephemeral channel's agents from the backend. It counts a + // member as an agent by its "bot" role or by a trusted kind:30177 + // managed-agent record (see huddle/roster.rs). This works // for both creators and joiners, and reflects mid-huddle agent additions. // // FAIL-CLOSED: agentsLoaded starts false. Until the fetch succeeds and