jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(desktop): recognize agents without NIP-OA in Huddles (#8544)

Browse Source 
🤖
## Summary
In a Huddle (a voice call in a channel), Buzz Desktop decides which
participants are agents. It uses that answer to choose whose text it
speaks aloud, which agents can publish voice, and when the last person
leaves and the Huddle ends.

Desktop currently treats a participant as an agent only when its channel
role is `bot`. #8494 changes the relay so that it gives `bot` only to
agents with a verified owner attestation (NIP-OA). After that change,
agents without an attestation join as ordinary members, and Desktop
treats them as people:
- their messages are not spoken,
- their voice publisher is refused,
- the Huddle stays open after the last person leaves while such an agent
is still connected.

With this PR, Desktop also counts a member as an agent when a trusted
person has published a managed-agent record for it. Desktop already
publishes such a record for every agent it runs, on every machine.
Trusted people are:
- **you**, so your own agents count as agents on all of your machines;
- **the Huddle's owner**, usually the person who started it, so the
agents they added count as agents for everyone who joins.

Some agents without an attestation still count as people: an agent added
by someone who does not own the Huddle, as seen from any machine other
than that person's. The only way to recognize those agents is an owner
attestation. This PR does not sign one.

### Details
- `huddle/roster.rs` (`fetch_huddle_roster`) is now the only place that
decides agent or person for Huddles. A member is an agent if its role is
`bot`, or if the current user or a channel `owner` published a
kind:30177 managed-agent record whose `d` tag is that member's key.
Records from other authors are ignored. Records whose signature does not
verify are ignored.
- All Huddle checks use the roster: voice-publisher authorization,
`get_huddle_agent_pubkeys` (spoken-text gating and the participant
list), the periodic roster refresh, adding an agent, cleanup at the end
of the Huddle, and the last-person check.
- Trusting the owner's records is a compatibility policy for Huddles,
not proof of identity. A record creates no owner. It lets the channel
owner mark a member as an agent for every Huddle behavior that reads the
roster: spoken text, voice publishing, remote transcription and
turn-taking (both skip agent audio), cleanup, and the last-person check.
Before #8494 the client chose the role, so an owner could already do
this by labeling a member `bot`.
- If the managed-agent records cannot be read, Desktop does not drop
agents that have the `bot` role. Spoken text, voice publishing, adding
agents and cleanup keep working for those agents. The Huddle does not
end automatically, because an agent without an attestation could be
counted as a person. When you join, and on each periodic refresh,
Desktop adds the `bot` agents to its stored list and keeps stored agents
that are still members.

### Related issue
Follow-up to #8494 (the relay stores `bot` only for attested agents). No
existing issue found.

### Testing
- New helper-level unit tests in `huddle/roster.rs`. They do not run the
relay query, the pipeline writers or the leave path.
- the `bot` role, the user's own record and the owner's record each mark
a member as an agent;
  - records from other members are ignored;
  - a record with a forged signature is ignored;
- when the record read fails, `bot` agents stay agents, the Huddle
cannot end automatically, and the next successful read restores the full
roster.
- when the record read fails, a joiner with no stored agents still
stores the `bot` agents, an agent known only by its record stays stored,
and a stored agent that left drops out.
- New tests in `huddle/roster_hydration_tests.rs` run the real
`post_connect_setup` and hot-start entrypoints against a fake relay
whose managed-agent record read fails. The voice-model, audio and
pipeline steps are passed in, so the tests replace them. Post-connect
stores the `bot` agent before media setup starts; the hot-start refresh
also keeps a still-member agent known from an earlier record and drops
one that left; the same incomplete roster gives no human count for the
last-person check. The tests fail when either entrypoint skips its
roster read, and when the roster code accepts only complete rosters.
- `cargo test --lib huddle::` (Huddle tests only), `cargo clippy --tests
-D warnings` and `cargo fmt --check` pass in `desktop/src-tauri`.
- Not yet tested in a live Huddle.

---------

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-10-26 16:00:13 -04:00
committed by GitHub
parent 834dca26bf
commit a38fedb234
8 changed files with 666 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
desktop/src-tauri/src/huddle/agent_tts_publisher.rs
View File

@@ -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,

9
desktop/src-tauri/src/huddle/agents.rs
View File

@@ -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();

43
desktop/src-tauri/src/huddle/mod.rs
View File

@@ -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<String>, 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<Vec<
hs.ephemeral_channel_id.clone()
};
match eph_id {
Some(id) => fetch_channel_members(&id, Some("bot"), &state).await,
Some(id) => fetch_huddle_roster(&id, &state)
.await
.map(|roster| roster.agents),
None => Ok(Vec::new()),
}
}

252
desktop/src-tauri/src/huddle/pipeline.rs
View File

@@ -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<PostConnectOutcome, String> {
{
let hs = state.huddle()?;
if !hs.is_current_huddle(ephemeral_channel_id, huddle_generation) {
return Ok(PostConnectOutcome::Stale);
}
}
) -> Result<bool, String> {
// 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<PostConnectOutcome, String>,
) -> Result<PostConnectOutcome, String> {
{
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<PostConnectOutcome, String> {
// 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::{

22
desktop/src-tauri/src/huddle/relay_api.rs
View File

@@ -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<Vec<String>, 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<usize, String> {
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)]

331
desktop/src-tauri/src/huddle/roster.rs Normal file
View File

@@ -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<String>,
pub(crate) humans: Vec<String>,
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<String>) -> bool {
let next = if self.complete {
self.agents
} else {
let mut next: Vec<String> = 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<usize> {
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<HuddleRoster, String> {
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<String>)],
records: Result<Vec<Event>, 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<String>)], viewer: &str) -> Vec<String> {
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<String>)]) -> Vec<String> {
members
.iter()
.filter(|(_, role)| role.as_deref() != Some("bot"))
.map(|(pubkey, _)| pubkey.to_ascii_lowercase())
.collect()
}
fn classify(
members: &[(String, Option<String>)],
records: &[Event],
trusted_authors: &[String],
) -> HuddleRoster {
let trusted: Vec<Event> = records
.iter()
.filter(|event| {
trusted_authors
.iter()
.any(|author| author.eq_ignore_ascii_case(&event.pubkey.to_hex()))
})
.cloned()
.collect();
let managed: HashSet<String> = 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<String>) {
(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"]);
}
}

132
desktop/src-tauri/src/huddle/roster_hydration_tests.rs Normal file
View File

@@ -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<nostr::Event>,
Json(filters): Json<Vec<Value>>| 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<String> {
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());
}

5
desktop/src/features/huddle/lib/useTtsSubscription.ts
View File

@@ -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