diff --git a/crates/buzz-acp/src/directory.rs b/crates/buzz-acp/src/directory.rs new file mode 100644 index 00000000000..56c4120e39a --- /dev/null +++ b/crates/buzz-acp/src/directory.rs @@ -0,0 +1,385 @@ +//! Kind:10100 agent-directory record publisher. +//! +//! The harness advertises the `respond_to` policy it actually enforces and the +//! channels it actually listens on, so member clients can decide whether this +//! agent is @-mentionable (`agentAutocompleteEligibility` in the desktop app, +//! `agent_identity_provider` on mobile). Publishing from the harness — rather +//! than the owner's client — guarantees the advertisement can never drift from +//! enforcement: a deployment that clamps `--respond-to` also clamps what gets +//! advertised. +//! +//! The record is a replaceable event authored by the agent key; consumers +//! (`agents_from_events` in the desktop Tauri backend) treat the event author +//! as the authoritative pubkey and parse the content as a flat JSON object. +//! `channel_add_policy` is carried forward from the current head record so a +//! policy set via `buzz channels set-add-policy` survives our republish (the +//! relay's kind:10100 side effect also requires the field). + +use std::time::Duration; + +use buzz_core::kind::{KIND_AGENT_PROFILE, KIND_NIP29_GROUP_METADATA}; +use nostr::{Alphabet, EventBuilder, Filter, Keys, Kind, SingleLetterTag}; +use serde_json::{json, Value}; +use tokio::task::JoinHandle; +use uuid::Uuid; + +use crate::config::RespondTo; +use crate::relay::{merge_discovered_channels, ChannelInfo, RestClient}; + +/// Trailing-edge debounce for membership-churn republishes. +pub(crate) const DIRECTORY_DEBOUNCE: Duration = Duration::from_secs(2); +/// Bound on each relay round-trip (head fetch, metadata fetch, submit). Wide +/// enough for the RestClient's internal transient-error retries to complete. +const DIRECTORY_STEP_TIMEOUT: Duration = Duration::from_secs(10); +/// Fallback add-policy when no head record exists — matches the relay's +/// database default for users without a kind:10100 record. +const DEFAULT_CHANNEL_ADD_POLICY: &str = "anyone"; + +/// Static identity and policy inputs for directory publishes, built once at +/// startup. Cheap to clone into spawned publish tasks. +#[derive(Clone)] +pub(crate) struct DirectoryPublisher { + /// HTTP bridge client (NIP-98 signed as the agent). + pub rest: RestClient, + /// Agent keys — the record must be authored by the agent itself. + pub keys: Keys, + /// The author gate mode this harness enforces. + pub respond_to: RespondTo, + /// Allowlist pubkeys (relevant when `respond_to` is `allowlist`). + pub respond_to_allowlist: Vec, + /// Normalized harness/runtime identity (e.g. `claude-agent-acp`). + pub agent_type: String, + /// Display name for the record; `None` omits the key so consumers fall + /// back to the agent's npub / kind:0 profile. + pub name: Option, +} + +impl DirectoryPublisher { + /// Spawn a debounced, best-effort publish of the directory record. + /// + /// Callers hold at most one outstanding handle and abort it before + /// spawning a replacement (trailing-edge debounce, same pattern as the + /// presence heartbeat task). + pub(crate) fn spawn_publish( + &self, + channel_ids: Vec, + status: &'static str, + debounce: Duration, + ) -> JoinHandle<()> { + let publisher = self.clone(); + tokio::spawn(async move { + if !debounce.is_zero() { + tokio::time::sleep(debounce).await; + } + publisher.publish_once(channel_ids, status).await; + }) + } + + /// One publish attempt: carry forward `channel_add_policy` from the head + /// record, resolve channel names, sign, submit. Every failure is logged + /// and swallowed — directory publishing must never take the harness down. + pub(crate) async fn publish_once(&self, channel_ids: Vec, status: &str) { + let channel_add_policy = match self.fetch_head_add_policy().await { + Ok(policy) => policy, + Err(e) => { + // Never overwrite a policy we could not read — skip entirely. + tracing::warn!("directory publish skipped (head fetch failed): {e}"); + return; + } + }; + + let channels = if channel_ids.is_empty() { + Vec::new() + } else { + match self.resolve_channels(channel_ids).await { + Ok(map) => advertisable_channels(map), + Err(e) => { + tracing::warn!("directory publish skipped (channel metadata failed): {e}"); + return; + } + } + }; + + let content = build_directory_content( + self.name.as_deref(), + &self.agent_type, + status, + &self.respond_to, + &self.respond_to_allowlist, + &channels, + &channel_add_policy, + ); + + let event = + match EventBuilder::new(Kind::Custom(KIND_AGENT_PROFILE as u16), content.to_string()) + .tags([]) + .sign_with_keys(&self.keys) + { + Ok(event) => event, + Err(e) => { + tracing::warn!("directory record sign failed: {e}"); + return; + } + }; + + match tokio::time::timeout(DIRECTORY_STEP_TIMEOUT, self.rest.submit_event(&event)).await { + Ok(Ok(_)) => tracing::info!( + status, + channels = channels.len(), + respond_to = %self.respond_to, + "published kind:10100 directory record" + ), + Ok(Err(e)) => tracing::warn!("directory record publish failed: {e}"), + Err(_) => tracing::warn!("directory record publish timed out"), + } + } + + /// Fetch the agent's current head kind:10100 record and extract its + /// `channel_add_policy`, defaulting when no record (or no field) exists. + async fn fetch_head_add_policy(&self) -> Result { + let filter = Filter::new() + .kind(Kind::Custom(KIND_AGENT_PROFILE as u16)) + .author(self.keys.public_key()) + .limit(1); + let head = tokio::time::timeout(DIRECTORY_STEP_TIMEOUT, self.rest.query(&[filter])) + .await + .map_err(|_| "head query timed out".to_string())? + .map_err(|e| e.to_string())?; + Ok(parse_add_policy(&head)) + } + + /// Resolve channel names/types via kind:39000 metadata, reusing the + /// discovery-time merge (which also drops archived channels). + async fn resolve_channels( + &self, + channel_ids: Vec, + ) -> Result, String> { + let d_tag = SingleLetterTag::lowercase(Alphabet::D); + let d_values: Vec = channel_ids.iter().map(|u| u.to_string()).collect(); + let filter = Filter::new() + .kind(Kind::Custom(KIND_NIP29_GROUP_METADATA as u16)) + .custom_tags(d_tag, d_values); + let meta = tokio::time::timeout(DIRECTORY_STEP_TIMEOUT, self.rest.query(&[filter])) + .await + .map_err(|_| "metadata query timed out".to_string())? + .map_err(|e| e.to_string())?; + Ok(merge_discovered_channels(channel_ids, &meta)) + } +} + +/// Reduce discovered channels to the advertisable `(id, name)` set, sorted by +/// id for deterministic output. +/// +/// DM channels are excluded (non-owner DM invocation is rejected by the author +/// gate regardless of `respond_to`), and so are channels whose type could not +/// be resolved — the author gate fails closed on unresolved types, so +/// advertising them would invite mentions the harness then silently drops. +fn advertisable_channels(map: std::collections::HashMap) -> Vec<(Uuid, String)> { + let mut channels: Vec<(Uuid, String)> = map + .into_iter() + .filter(|(_, info)| info.channel_type != "dm" && info.channel_type != "unknown") + .map(|(id, info)| (id, info.name)) + .collect(); + channels.sort_by_key(|(id, _)| *id); + channels +} + +/// Extract `content.channel_add_policy` from the newest head event in a +/// `/query` response, defaulting to the relay's database default. +fn parse_add_policy(head_events: &Value) -> String { + head_events + .as_array() + .and_then(|events| events.first()) + .and_then(|event| event.get("content")) + .and_then(Value::as_str) + .and_then(|content| serde_json::from_str::(content).ok()) + .and_then(|content| { + content + .get("channel_add_policy") + .and_then(Value::as_str) + .map(str::to_string) + }) + .unwrap_or_else(|| DEFAULT_CHANNEL_ADD_POLICY.to_string()) +} + +/// Build the kind:10100 content object exactly as the desktop consumer +/// (`agents_from_events` → `RelayAgentInfo` → `RawRelayAgent`) parses it. +/// +/// The allowlist is re-sorted and `channels`/`channel_ids` stay index-aligned; +/// `name` is omitted (not null) when absent so the consumer's npub fallback +/// engages. +fn build_directory_content( + name: Option<&str>, + agent_type: &str, + status: &str, + respond_to: &RespondTo, + respond_to_allowlist: &[String], + channels: &[(Uuid, String)], + channel_add_policy: &str, +) -> Value { + let mut allowlist: Vec<&str> = respond_to_allowlist.iter().map(String::as_str).collect(); + allowlist.sort_unstable(); + let mut content = serde_json::Map::new(); + if let Some(name) = name { + content.insert("name".into(), json!(name)); + } + content.insert("agent_type".into(), json!(agent_type)); + content.insert("status".into(), json!(status)); + content.insert("respond_to".into(), json!(respond_to.to_string())); + content.insert("respond_to_allowlist".into(), json!(allowlist)); + content.insert( + "channels".into(), + json!(channels.iter().map(|(_, name)| name).collect::>()), + ); + content.insert( + "channel_ids".into(), + json!(channels + .iter() + .map(|(id, _)| id.to_string()) + .collect::>()), + ); + content.insert("capabilities".into(), json!([])); + content.insert("channel_add_policy".into(), json!(channel_add_policy)); + Value::Object(content) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn uuid(n: u8) -> Uuid { + Uuid::from_u128(n as u128) + } + + #[test] + fn content_matches_consumer_schema() { + let channels = vec![ + (uuid(1), "general".to_string()), + (uuid(2), "hq".to_string()), + ]; + let content = build_directory_content( + Some("Scout"), + "claude-agent-acp", + "online", + &RespondTo::Anyone, + &[], + &channels, + "anyone", + ); + + assert_eq!(content["name"], "Scout"); + assert_eq!(content["agent_type"], "claude-agent-acp"); + assert_eq!(content["status"], "online"); + assert_eq!(content["respond_to"], "anyone"); + assert_eq!(content["respond_to_allowlist"], json!([])); + assert_eq!(content["channels"], json!(["general", "hq"])); + assert_eq!( + content["channel_ids"], + json!([uuid(1).to_string(), uuid(2).to_string()]) + ); + assert_eq!(content["capabilities"], json!([])); + assert_eq!(content["channel_add_policy"], "anyone"); + } + + #[test] + fn respond_to_serializes_to_ts_literals() { + for (mode, expected) in [ + (RespondTo::OwnerOnly, "owner-only"), + (RespondTo::Allowlist, "allowlist"), + (RespondTo::Anyone, "anyone"), + (RespondTo::Nobody, "nobody"), + ] { + let content = build_directory_content(None, "acp", "online", &mode, &[], &[], "anyone"); + assert_eq!(content["respond_to"], expected); + } + } + + #[test] + fn name_key_omitted_when_none() { + let content = build_directory_content( + None, + "acp", + "online", + &RespondTo::Anyone, + &[], + &[], + "anyone", + ); + assert!(!content.as_object().is_some_and(|o| o.contains_key("name"))); + } + + #[test] + fn allowlist_always_present_and_sorted() { + let allowlist = vec!["bb".repeat(32), "aa".repeat(32)]; + let content = build_directory_content( + None, + "acp", + "online", + &RespondTo::Allowlist, + &allowlist, + &[], + "anyone", + ); + assert_eq!( + content["respond_to_allowlist"], + json!(["aa".repeat(32), "bb".repeat(32)]) + ); + } + + #[test] + fn advertisable_channels_excludes_dm_and_unknown_and_sorts() { + let mut map = std::collections::HashMap::new(); + map.insert( + uuid(3), + ChannelInfo { + name: "later".into(), + channel_type: "stream".into(), + }, + ); + map.insert( + uuid(1), + ChannelInfo { + name: "first".into(), + channel_type: "private".into(), + }, + ); + map.insert( + uuid(2), + ChannelInfo { + name: "direct".into(), + channel_type: "dm".into(), + }, + ); + map.insert( + uuid(4), + ChannelInfo { + name: "mystery".into(), + channel_type: "unknown".into(), + }, + ); + + let channels = advertisable_channels(map); + assert_eq!( + channels, + vec![(uuid(1), "first".into()), (uuid(3), "later".into())] + ); + } + + #[test] + fn parse_add_policy_defaults_and_carries() { + assert_eq!(parse_add_policy(&json!([])), "anyone"); + assert_eq!(parse_add_policy(&json!("not-an-array")), "anyone"); + assert_eq!( + parse_add_policy(&json!([{ "content": "{\"channel_add_policy\":\"nobody\"}" }])), + "nobody" + ); + assert_eq!( + parse_add_policy(&json!([{ "content": "not json" }])), + "anyone" + ); + assert_eq!( + parse_add_policy(&json!([{ "content": "{\"name\":\"x\"}" }])), + "anyone" + ); + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index fa348eeb3cf..818cdc78282 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2,6 +2,7 @@ mod acp; mod config; +mod directory; mod engram_fetch; mod filter; mod observer; @@ -1797,6 +1798,24 @@ async fn tokio_main() -> Result<()> { } } + // Advertise this agent in the kind:10100 directory so other members' + // clients can offer it in @-mention autocomplete. Published after channel + // subscriptions so the advertised channel set matches what we listen on. + let directory_publisher = directory::DirectoryPublisher { + rest: relay.rest_client(), + keys: config.keys.clone(), + respond_to: config.respond_to.clone(), + respond_to_allowlist: config.respond_to_allowlist.iter().cloned().collect(), + agent_type: config::normalize_agent_command_identity(&config.agent_command), + name: config.session_title.clone(), + }; + let mut directory_task: Option> = + Some(directory_publisher.spawn_publish( + subscribed_channel_ids.iter().copied().collect(), + "online", + Duration::ZERO, + )); + if config.lazy_pool { emit_runtime_lifecycle( observer.as_ref(), @@ -2262,6 +2281,7 @@ async fn tokio_main() -> Result<()> { } membership_newest_ts.insert(ch, ts); + let mut subscribed_set_changed = false; if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION { // Clear removal tracking so sessions are not // stripped for a legitimately re-added channel. @@ -2275,12 +2295,13 @@ async fn tokio_main() -> Result<()> { tracing::warn!("failed to subscribe to new channel {ch}: {e}"); } else { subscribed_channel_ids.insert(ch); + subscribed_set_changed = true; } } else { tracing::debug!(channel_id = %ch, "membership notification: no matching rules — skipping"); } } else { - subscribed_channel_ids.remove(&ch); + subscribed_set_changed = subscribed_channel_ids.remove(&ch); tracing::info!(channel_id = %ch, "membership notification: unsubscribing from channel"); if let Err(e) = relay.unsubscribe_channel(ch).await { tracing::warn!("failed to unsubscribe from channel {ch}: {e}"); @@ -2323,6 +2344,18 @@ async fn tokio_main() -> Result<()> { ); } } + if subscribed_set_changed { + // Re-advertise the updated channel set, + // debounced to absorb membership churn. + if let Some(handle) = directory_task.take() { + handle.abort(); + } + directory_task = Some(directory_publisher.spawn_publish( + subscribed_channel_ids.iter().copied().collect(), + "online", + directory::DIRECTORY_DEBOUNCE, + )); + } continue; } @@ -3052,6 +3085,23 @@ async fn tokio_main() -> Result<()> { } } + // Best-effort: flip the kind:10100 directory record to offline. Status is + // cosmetic to mention eligibility (presence stays the liveness signal), so + // a crash that leaves "online" behind is acceptable. + if let Some(handle) = directory_task.take() { + handle.abort(); + } + if tokio::time::timeout( + Duration::from_secs(5), + directory_publisher + .publish_once(subscribed_channel_ids.iter().copied().collect(), "offline"), + ) + .await + .is_err() + { + tracing::warn!("offline directory update timed out"); + } + if let Some(handle) = relay_observer_publisher_task.take() { handle.abort(); } diff --git a/desktop/src-tauri/tests/agent_directory_schema.rs b/desktop/src-tauri/tests/agent_directory_schema.rs new file mode 100644 index 00000000000..c25313e05cd --- /dev/null +++ b/desktop/src-tauri/tests/agent_directory_schema.rs @@ -0,0 +1,71 @@ +//! Pins the kind:10100 agent-directory content contract between the +//! harness-side producer (`crates/buzz-acp/src/directory.rs`, +//! `build_directory_content`) and the desktop consumer +//! (`nostr_convert::agents_from_events`). The fixture below mirrors the +//! producer's output shape — keep the two in sync so harness-advertised +//! agents stay parseable by the discovery pipeline. +//! +//! Kept as an integration test (rather than growing the `nostr_convert` +//! in-module test suite) so the schema pin lives beside the contract it +//! guards without pushing an already-at-limit file past the size ratchet. +//! Deserialization of this JSON into `RelayAgentInfo` — including the +//! `respond_to` / `respond_to_allowlist` fields — is pinned by the +//! in-module `agents_*_for_directory_parse` tests. + +use nostr::{EventBuilder, Keys, Kind}; +use serde_json::Value; + +const HARNESS_CONTENT: &str = r#"{"name":"Scout","agent_type":"claude-agent-acp","status":"online","respond_to":"anyone","respond_to_allowlist":[],"channels":["general","hq"],"channel_ids":["00000000-0000-0000-0000-000000000001","00000000-0000-0000-0000-000000000002"],"capabilities":[],"channel_add_policy":"anyone"}"#; + +#[test] +fn agents_from_events_parses_harness_published_directory_record() { + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(10100), HARNESS_CONTENT) + .tags([]) + .sign_with_keys(&keys) + .expect("sign fixture event"); + + let value = buzz_lib::nostr_convert::agents_from_events(std::slice::from_ref(&event)); + let agents = value + .get("agents") + .and_then(Value::as_array) + .expect("agents array"); + assert_eq!(agents.len(), 1); + + let agent = &agents[0]; + // Event author is authoritative — content has no pubkey, converter adds it. + assert_eq!( + agent.get("pubkey").and_then(Value::as_str), + Some(keys.public_key().to_hex().as_str()) + ); + assert_eq!(agent.get("name").and_then(Value::as_str), Some("Scout")); + assert_eq!( + agent.get("agent_type").and_then(Value::as_str), + Some("claude-agent-acp") + ); + assert_eq!(agent.get("status").and_then(Value::as_str), Some("online")); + assert_eq!( + agent.get("respond_to").and_then(Value::as_str), + Some("anyone") + ); + assert_eq!( + agent.get("respond_to_allowlist"), + Some(&serde_json::json!([])) + ); + assert_eq!( + agent.get("channels"), + Some(&serde_json::json!(["general", "hq"])) + ); + assert_eq!( + agent.get("channel_ids"), + Some(&serde_json::json!([ + "00000000-0000-0000-0000-000000000001", + "00000000-0000-0000-0000-000000000002" + ])) + ); + assert_eq!(agent.get("capabilities"), Some(&serde_json::json!([]))); + assert_eq!( + agent.get("channel_add_policy").and_then(Value::as_str), + Some("anyone") + ); +} diff --git a/desktop/src/features/messages/lib/useMentions.ts b/desktop/src/features/messages/lib/useMentions.ts index b9737f1f398..cb92ffc755b 100644 --- a/desktop/src/features/messages/lib/useMentions.ts +++ b/desktop/src/features/messages/lib/useMentions.ts @@ -17,7 +17,6 @@ import { filterCachedAgentSuggestions, getMentionableAgentPubkeys, getSharedChannelIds, - isAgentIdentityInAllowedList, isAgentMentionChannelType, shouldHideAgentFromMentions, uniqueAutocompleteLabels, @@ -257,9 +256,6 @@ export function useMentions( if (isArchivedDiscovery(pubkey)) { return; } - if (!isAgentIdentityInAllowedList(candidate, mentionableAgentPubkeys)) { - return; - } if ( shouldHideAgentFromMentions({ isAgent: candidate.isAgent === true, diff --git a/desktop/tests/e2e/mentions.spec.ts b/desktop/tests/e2e/mentions.spec.ts index ed00e8c3556..4ee75fb3b20 100644 --- a/desktop/tests/e2e/mentions.spec.ts +++ b/desktop/tests/e2e/mentions.spec.ts @@ -848,9 +848,12 @@ test("other-owned agents without a shared channel are hidden from mentions", asy await expect(input.locator(".mention-chip")).toHaveCount(0); }); -test("stale channel-member agents absent from managed and relay directories stay hidden", async ({ +test("channel-member agents absent from managed and relay directories stay mentionable", async ({ page, }) => { + // Unknown invocability: a member agent with no kind:10100 directory record + // is shown rather than hidden, so agents whose harness has not (yet) + // published a directory record remain addressable in their channels. await installMockBridge(page, { userSearchDelayMs: 1_000 }); await page.goto("/"); await page.getByTestId("channel-general").click(); @@ -859,7 +862,15 @@ test("stale channel-member agents absent from managed and relay directories stay const input = page.getByTestId("message-input"); await input.fill("@mira"); - await expect(autocomplete(page)).toHaveCount(0); + const dropdown = autocomplete(page); + await expect(dropdown.getByText("mira")).toBeVisible(); + await input.press("Enter"); + + const agentMentionChip = input.locator(".agent-mention-highlight", { + hasText: "mira", + }); + await expect(agentMentionChip).toBeVisible(); + await expect(agentMentionChip).toHaveText("mira"); }); test("managed relay agents are visible in channel mentions regardless of relay policy", async ({