diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index a2dcdce6d2..1be6ccc26e 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -2,6 +2,9 @@ Agent-first command-line interface for Buzz relay. JSON in, JSON out. +Resident harness integrations should also follow the +[external-agent CLI contract](../../docs/cli-external-agents.md). + ## Install ```bash @@ -20,6 +23,26 @@ export BUZZ_PRIVATE_KEY="nsec1..." buzz channels list ``` +### Minting an identity for a self-hosted agent + +`buzz keys generate` creates a keypair without a relay connection and without +an existing `BUZZ_PRIVATE_KEY`. Run it **on the machine that will use the +identity** — the secret is then created where it is used and never has to be +copied from an operator workstation. + +```bash +# On the agent's own host +buzz keys generate --out ~/.config/buzz/agent.nsec +# → {"pubkey":"<64-hex>","npub":"npub1...","secret_key_path":"/home/agent/.config/buzz/agent.nsec"} + +export BUZZ_PRIVATE_KEY="$(cat ~/.config/buzz/agent.nsec)" +``` + +The secret is written with mode `0600` and is **not** printed unless `--stdout` +is passed; stdout carries only the public half, so the pubkey can be registered +without the secret ever passing through another process. An existing `--out` +file is never overwritten without `--force`. + ## Usage All output is JSON on stdout. Errors are JSON on stderr. Exit codes: 0=ok, 1=user error, 2=network, 3=auth, 4=other, 5=write conflict. @@ -28,6 +51,9 @@ All output is JSON on stdout. Errors are JSON on stderr. Exit codes: 0=ok, 1=use # Set relay URL (defaults to http://localhost:3000) export BUZZ_RELAY_URL="https://relay.example.com" +# Realtime external-agent ingress +buzz listen --channel --mentions-of-me --envelope v1 --no-reconnect + # Messages buzz messages send --channel --content "Hello" buzz messages send --channel --content "Reply" --reply-to --broadcast @@ -53,6 +79,7 @@ buzz reactions add --event --emoji "👍" buzz reactions get --event # Users & Presence +buzz users me # local identity; no relay request buzz users get # your own profile buzz users get --pubkey # single user buzz users get --pubkey --pubkey # batch (max 200) @@ -102,6 +129,7 @@ stored rules in `validation_error` so an owner can remove and repair them. | Group | Subcommand | Description | |-------|-----------|-------------| +| `listen` | | Stream channel events as NDJSON | | `messages` | `send` | Send a message to a channel | | | `send-diff` | Send a code diff with metadata | | | `edit` | Edit a message you sent | @@ -133,6 +161,7 @@ stored rules in `validation_error` so an owner can remove and repair them. | | `open` | Open a DM (1–8 pubkeys) | | | `add-member` | Add member to DM group | | `users` | `get` | Get user profile(s) | +| | `me` | Print the active local identity | | | `set-profile` | Update your profile | | | `presence` | Get presence status | | | `set-presence` | Set presence status | @@ -160,6 +189,7 @@ stored rules in `validation_error` so an owner can remove and repair them. | `upload` | `file` | Upload a file to the Blossom store | | `pack` | `validate` | Validate a persona pack (local, no relay) | | | `inspect` | Inspect a persona pack (local, no relay) | +| `keys` | `generate` | Mint a new agent identity (local, no relay, no key required) | | `mem` | `ls` | List non-tombstoned memories | | | `get` | Print memory value to stdout | | | `hash` | Print SHA-256 hex of memory value | @@ -167,6 +197,37 @@ stored rules in `validation_error` so an owner can remove and repair them. | | `patch` | Apply unified diff to memory value | | | `rm` | Publish a tombstone to delete memory | +## Agent profile (kind:10100) + +`kind:10100` is the agent-authored directory record. Buzz Desktop discovers +agents by querying it, so a self-hosted agent publishes its own profile to +become visible and mentionable in a workspace — no Desktop-side ownership of +the process required. + +```bash +buzz agents profile get +buzz agents profile set --display-name Scout --agent-type researcher --policy owner_only +buzz agents profile set --capabilities search,summarize # policy inherited +buzz agents profile set --status online +``` + +**`kind:10100` is a replaceable event** — the relay keeps only the newest one +per author. Writes are therefore read-modify-write: the current profile is +fetched and your changes are layered onto it, so a partial update cannot drop +fields you did not mention. Fields absent from an existing profile are +preserved even when this CLI build does not recognize them. + +`channel_add_policy` (`--policy`: `anyone`, `owner_only`, `nobody`) is +required. It is inherited from the existing profile when present; a first +profile must pass `--policy` explicitly. The reason is that the relay derives +a stored policy from this event in a side effect, and side-effect failures are +logged rather than rejected — so a profile published without the field would +replace the visible record while leaving the relay's stored policy untouched, +leaving the event log and the database silently disagreeing. + +`buzz channels set-add-policy` writes the same event through the same +read-modify-write path. + ## Architecture ``` diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index d0dd2677a9..92c209bbb8 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -563,6 +563,11 @@ impl BuzzClient { &self.keys } + /// Get the parsed NIP-OA auth tag, if configured. + pub fn auth_tag(&self) -> Option<&Tag> { + self.auth_tag.as_ref() + } + /// Get the relay base URL. #[allow(dead_code)] pub fn relay_url(&self) -> &str { diff --git a/crates/buzz-cli/src/commands/agent_profile.rs b/crates/buzz-cli/src/commands/agent_profile.rs new file mode 100644 index 0000000000..de22589f51 --- /dev/null +++ b/crates/buzz-cli/src/commands/agent_profile.rs @@ -0,0 +1,457 @@ +//! Agent profile (kind:10100) read and write. +//! +//! Kind:10100 is the agent-authored directory record. Buzz Desktop discovers +//! agents by querying it unfiltered, so it is how an agent that runs on its own +//! machine becomes visible, mentionable, and addressable in a workspace without +//! the Desktop owning or supervising it. +//! +//! # Why every write is read-modify-write +//! +//! Kind:10100 is a **replaceable** event: the relay keeps only the newest one +//! per author. A writer that publishes a partial profile does not merge into +//! the previous one, it *replaces* it, and every field it omitted is gone. +//! +//! That has a second, quieter consequence. The relay derives a stored +//! `channel_add_policy` column from this event in a side effect, and side +//! effect failures are logged rather than rejected — the event is still +//! accepted and still becomes the author's profile. So a profile published +//! without `channel_add_policy` replaces the visible record *and* leaves the +//! relay's stored policy at its previous value, with nothing but a relay-side +//! warning to show for it. The event log and the database disagree, silently. +//! +//! Both problems have the same fix, applied here: read the current profile, +//! layer the caller's changes onto it, and always emit a complete document. + +use nostr::{EventBuilder, Kind}; + +use crate::client::BuzzClient; +use crate::error::CliError; + +/// Policy values the relay's `channel_add_policy` side effect accepts. +pub const VALID_ADD_POLICIES: [&str; 3] = ["anyone", "owner_only", "nobody"]; + +/// Fields a caller may set on an agent profile. +/// +/// Every field is optional: `None` means "leave whatever the current profile +/// has", which is what makes partial updates safe against the replaceable-event +/// clobber described in the module docs. +#[derive(Debug, Default, Clone)] +pub struct ProfileUpdate { + pub display_name: Option, + pub agent_type: Option, + pub capabilities: Option>, + pub status: Option, + pub channel_add_policy: Option, +} + +impl ProfileUpdate { + fn is_empty(&self) -> bool { + self.display_name.is_none() + && self.agent_type.is_none() + && self.capabilities.is_none() + && self.status.is_none() + && self.channel_add_policy.is_none() + } +} + +/// Status values the Desktop renders. `agents_from_events` defaults an +/// absent or non-string status to `offline`, so anything outside this set +/// would round-trip to something the caller did not write. +const VALID_STATUSES: [&str; 3] = ["online", "away", "offline"]; + +// Note on unknown content keys: an unrecognized key in an existing profile is +// preserved untouched rather than filtered out. A newer Buzz may have added a +// field this CLI build does not know about, and dropping it would be the very +// clobber this module exists to prevent. Only *incoming* values are restricted, +// via `ProfileUpdate`'s typed fields. + +/// Validate a policy string against the set the relay side effect accepts. +pub fn validate_add_policy(policy: &str) -> Result<(), CliError> { + if VALID_ADD_POLICIES.contains(&policy) { + return Ok(()); + } + Err(CliError::Usage(format!( + "--policy must be one of {} (got: {policy})", + VALID_ADD_POLICIES.join(", ") + ))) +} + +fn validate_status(status: &str) -> Result<(), CliError> { + if VALID_STATUSES.contains(&status) { + return Ok(()); + } + Err(CliError::Usage(format!( + "--status must be one of {} (got: {status})", + VALID_STATUSES.join(", ") + ))) +} + +/// Validate the caller's own values, independent of any existing profile. +/// +/// Split out from [`merge_profile`] so the write path can reject bad input +/// *before* it queries the relay. Ordering the fetch first would turn a typo in +/// `--policy` into a network error (exit 2) instead of an input error (exit 1), +/// and would spend a round trip discovering something already knowable. +fn validate_update(update: &ProfileUpdate) -> Result<(), CliError> { + if let Some(policy) = &update.channel_add_policy { + validate_add_policy(policy)?; + } + if let Some(status) = &update.status { + validate_status(status)?; + } + Ok(()) +} + +/// Merge `update` onto `current`, returning the complete profile document to +/// publish. +/// +/// `current` is the author's existing profile content, or `None` when they have +/// never published one. Pure so the merge semantics — the whole point of this +/// module — are testable without a relay. +/// +/// Fails when the result would carry no `channel_add_policy`: publishing such a +/// profile desyncs the relay's stored policy from the event log (see module +/// docs), so it is refused rather than written. +pub fn merge_profile( + current: Option<&serde_json::Value>, + update: &ProfileUpdate, +) -> Result { + validate_update(update)?; + + // Start from the existing document so unknown-to-this-build fields survive. + let mut merged = match current.and_then(|v| v.as_object()) { + Some(obj) => obj.clone(), + None => serde_json::Map::new(), + }; + + // A stored `pubkey` is never authoritative — the Desktop overwrites it with + // the event author on read. Carrying it forward would preserve a stale or + // forged value in the document for no benefit. + merged.remove("pubkey"); + + if let Some(v) = &update.display_name { + merged.insert("display_name".into(), serde_json::json!(v)); + } + if let Some(v) = &update.agent_type { + merged.insert("agent_type".into(), serde_json::json!(v)); + } + if let Some(v) = &update.capabilities { + merged.insert("capabilities".into(), serde_json::json!(v)); + } + if let Some(v) = &update.status { + merged.insert("status".into(), serde_json::json!(v)); + } + if let Some(v) = &update.channel_add_policy { + merged.insert("channel_add_policy".into(), serde_json::json!(v)); + } + + // Refuse rather than guess. Defaulting to a policy the caller did not + // choose would silently widen or narrow who may add this agent to channels. + let policy_ok = merged + .get("channel_add_policy") + .and_then(serde_json::Value::as_str) + .is_some_and(|p| VALID_ADD_POLICIES.contains(&p)); + if !policy_ok { + return Err(CliError::Usage(format!( + "channel_add_policy is required and must be one of {}. This identity has no \ + existing profile to inherit it from, so pass --policy explicitly. Publishing \ + a profile without it would replace the stored record while leaving the relay's \ + policy unchanged.", + VALID_ADD_POLICIES.join(", ") + ))); + } + + Ok(serde_json::Value::Object(merged)) +} + +/// Fetch the signing identity's current kind:10100 content, if any. +/// +/// Returns `None` when the identity has never published a profile. A stored +/// profile whose content is not a JSON object is also treated as `None`: it +/// carries nothing mergeable, and preserving unparseable bytes would just +/// propagate the corruption into the next write. +pub async fn fetch_current_profile( + client: &BuzzClient, +) -> Result, CliError> { + let me = client.keys().public_key().to_hex(); + let filter = serde_json::json!({ + "kinds": [buzz_sdk::kind::KIND_AGENT_PROFILE], + "authors": [me], + "limit": 1, + }); + let events = client.query_paginated(filter, 1).await?; + let Some(event) = events.first() else { + return Ok(None); + }; + let Some(content) = event.get("content").and_then(serde_json::Value::as_str) else { + return Ok(None); + }; + match serde_json::from_str::(content) { + Ok(value) if value.is_object() => Ok(Some(value)), + _ => Ok(None), + } +} + +/// Sign and submit a complete profile document as kind:10100. +async fn publish_profile( + client: &BuzzClient, + content: &serde_json::Value, +) -> Result { + let builder = EventBuilder::new( + Kind::Custom(buzz_sdk::kind::KIND_AGENT_PROFILE as u16), + content.to_string(), + ) + .tags([]); + let event = client.sign_event(builder)?; + client.submit_event(event).await +} + +/// Read-modify-write entry point shared by `agents profile set` and +/// `channels set-add-policy`, so the two can never disagree about how a +/// partial update is applied. +pub async fn apply_profile_update( + client: &BuzzClient, + update: &ProfileUpdate, +) -> Result { + // Validate before the fetch: bad input must fail as an input error without + // a network round trip. `merge_profile` re-checks, which is cheap and keeps + // it safe to call directly. + validate_update(update)?; + let current = fetch_current_profile(client).await?; + let merged = merge_profile(current.as_ref(), update)?; + publish_profile(client, &merged).await +} + +/// Run `buzz agents profile get`. +pub async fn cmd_profile_get(client: &BuzzClient) -> Result<(), CliError> { + let current = fetch_current_profile(client).await?; + let report = serde_json::json!({ + "pubkey": client.keys().public_key().to_hex(), + "profile": current, + }); + println!("{report}"); + Ok(()) +} + +/// Run `buzz agents profile set`. +pub async fn cmd_profile_set(client: &BuzzClient, update: &ProfileUpdate) -> Result<(), CliError> { + if update.is_empty() { + return Err(CliError::Usage( + "no fields to set: pass at least one of --display-name, --agent-type, \ + --capabilities, --status, --policy" + .into(), + )); + } + let resp = apply_profile_update(client, update).await?; + println!("{}", crate::client::normalize_write_response(&resp)); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn profile(json: serde_json::Value) -> serde_json::Value { + json + } + + #[test] + fn policy_only_update_preserves_every_other_field() { + // This is the regression that motivates the module. Before the + // read-modify-write refactor, `channels set-add-policy` published a + // document containing only `channel_add_policy`, wiping the agent's + // identity fields from the replaceable event. + let current = profile(serde_json::json!({ + "display_name": "Scout", + "agent_type": "researcher", + "capabilities": ["search", "summarize"], + "status": "online", + "channel_add_policy": "owner_only", + })); + let update = ProfileUpdate { + channel_add_policy: Some("anyone".into()), + ..Default::default() + }; + + let merged = merge_profile(Some(¤t), &update).unwrap(); + assert_eq!(merged["channel_add_policy"], "anyone"); + assert_eq!(merged["display_name"], "Scout"); + assert_eq!(merged["agent_type"], "researcher"); + assert_eq!( + merged["capabilities"], + serde_json::json!(["search", "summarize"]) + ); + assert_eq!(merged["status"], "online"); + } + + #[test] + fn field_update_preserves_existing_policy() { + // The mirror case: renaming the agent must not drop the policy and + // desync the relay's stored value. + let current = profile(serde_json::json!({ + "display_name": "Scout", + "channel_add_policy": "nobody", + })); + let update = ProfileUpdate { + display_name: Some("Scout II".into()), + ..Default::default() + }; + + let merged = merge_profile(Some(¤t), &update).unwrap(); + assert_eq!(merged["display_name"], "Scout II"); + assert_eq!(merged["channel_add_policy"], "nobody"); + } + + #[test] + fn unknown_fields_from_a_newer_buzz_survive() { + // A field this CLI build does not know about must not be dropped — + // dropping it is the same clobber, one release later. + let current = profile(serde_json::json!({ + "channel_add_policy": "anyone", + "some_future_field": {"nested": true}, + })); + let update = ProfileUpdate { + display_name: Some("Scout".into()), + ..Default::default() + }; + + let merged = merge_profile(Some(¤t), &update).unwrap(); + assert_eq!( + merged["some_future_field"], + serde_json::json!({"nested": true}) + ); + assert_eq!(merged["display_name"], "Scout"); + } + + #[test] + fn first_profile_requires_an_explicit_policy() { + let update = ProfileUpdate { + display_name: Some("Scout".into()), + ..Default::default() + }; + let err = merge_profile(None, &update).expect_err("expected usage error"); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn first_profile_succeeds_with_a_policy() { + let update = ProfileUpdate { + display_name: Some("Scout".into()), + channel_add_policy: Some("owner_only".into()), + ..Default::default() + }; + let merged = merge_profile(None, &update).unwrap(); + assert_eq!(merged["display_name"], "Scout"); + assert_eq!(merged["channel_add_policy"], "owner_only"); + } + + #[test] + fn a_current_profile_with_an_invalid_policy_is_not_inherited() { + // Garbage in the stored document must not be laundered into a new + // write just because it was already there. + let current = profile(serde_json::json!({ + "display_name": "Scout", + "channel_add_policy": "everyone", + })); + let update = ProfileUpdate { + display_name: Some("Scout II".into()), + ..Default::default() + }; + let err = merge_profile(Some(¤t), &update).expect_err("expected usage error"); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn rejects_invalid_policy_and_status() { + let base = serde_json::json!({"channel_add_policy": "anyone"}); + + let bad_policy = ProfileUpdate { + channel_add_policy: Some("everyone".into()), + ..Default::default() + }; + assert!(matches!( + merge_profile(Some(&base), &bad_policy), + Err(CliError::Usage(_)) + )); + + let bad_status = ProfileUpdate { + status: Some("busy".into()), + ..Default::default() + }; + assert!(matches!( + merge_profile(Some(&base), &bad_status), + Err(CliError::Usage(_)) + )); + } + + #[test] + fn stale_pubkey_in_content_is_dropped() { + // The Desktop overwrites `pubkey` with the event author on read, so a + // stored value is at best redundant and at worst a forged claim. + let current = profile(serde_json::json!({ + "pubkey": "deadbeef", + "channel_add_policy": "anyone", + })); + let update = ProfileUpdate { + display_name: Some("Scout".into()), + ..Default::default() + }; + let merged = merge_profile(Some(¤t), &update).unwrap(); + assert!(merged.get("pubkey").is_none()); + } + + #[test] + fn bad_input_is_rejected_without_needing_a_current_profile() { + // Regression: `apply_profile_update` used to fetch before validating, so + // a typo in --policy surfaced as a network error (exit 2) rather than an + // input error (exit 1). `validate_update` is what the write path calls + // first, so it must reject on the caller's values alone. + assert!(matches!( + validate_update(&ProfileUpdate { + channel_add_policy: Some("everyone".into()), + ..Default::default() + }), + Err(CliError::Usage(_)) + )); + assert!(matches!( + validate_update(&ProfileUpdate { + status: Some("busy".into()), + ..Default::default() + }), + Err(CliError::Usage(_)) + )); + // A valid update passes with no profile and no relay in sight. + assert!(validate_update(&ProfileUpdate { + display_name: Some("Scout".into()), + channel_add_policy: Some("anyone".into()), + ..Default::default() + }) + .is_ok()); + } + + #[test] + fn empty_update_is_detected() { + assert!(ProfileUpdate::default().is_empty()); + assert!(!ProfileUpdate { + status: Some("online".into()), + ..Default::default() + } + .is_empty()); + } + + #[test] + fn capabilities_are_replaced_not_appended() { + // Set semantics, not merge semantics: a caller passing --capabilities + // is stating the full list. Appending would make removal impossible. + let current = profile(serde_json::json!({ + "capabilities": ["old"], + "channel_add_policy": "anyone", + })); + let update = ProfileUpdate { + capabilities: Some(vec!["new".into()]), + ..Default::default() + }; + let merged = merge_profile(Some(¤t), &update).unwrap(); + assert_eq!(merged["capabilities"], serde_json::json!(["new"])); + } +} diff --git a/crates/buzz-cli/src/commands/agents.rs b/crates/buzz-cli/src/commands/agents.rs index 58564a45c2..8f23c75dad 100644 --- a/crates/buzz-cli/src/commands/agents.rs +++ b/crates/buzz-cli/src/commands/agents.rs @@ -11,6 +11,29 @@ use crate::{AgentsCmd, RespondToArg}; pub async fn dispatch(command: AgentsCmd, client: &BuzzClient) -> Result<(), CliError> { match command { + AgentsCmd::Profile(sub) => { + use crate::commands::agent_profile::{self, ProfileUpdate}; + use crate::AgentProfileCmd; + match sub { + AgentProfileCmd::Get => agent_profile::cmd_profile_get(client).await, + AgentProfileCmd::Set { + display_name, + agent_type, + capabilities, + status, + policy, + } => { + let update = ProfileUpdate { + display_name, + agent_type, + capabilities, + status, + channel_add_policy: policy, + }; + agent_profile::cmd_profile_set(client, &update).await + } + } + } AgentsCmd::DraftCreate { channel, display_name, diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 42844bf1e0..8a01e0d2fa 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -8,6 +8,7 @@ use crate::client::{ extract_d_tag, extract_p_tags, extract_tag_value, normalize_write_response, print_create_response, BuzzClient, }; +use crate::commands::agent_profile; use crate::commands::agents::fetch_archived_snapshot; use crate::commands::channel_templates::{self, ChannelTemplateRecord, TemplateAgentRoster}; use crate::error::CliError; @@ -1001,16 +1002,16 @@ pub async fn cmd_remove_channel_member( Ok(()) } -/// Set the channel addition policy — sign and submit a kind:10100 (agent profile) event. +/// Set the channel addition policy on the signing identity's kind:10100 +/// (agent profile) event. +/// +/// Kind:10100 is replaceable, so this is a read-modify-write through +/// [`agent_profile::apply_profile_update`] rather than a bare publish. Writing +/// a document containing only `channel_add_policy` would replace the author's +/// whole profile and drop their display name, type, capabilities, and status — +/// see the `agent_profile` module docs for the full failure mode. pub async fn cmd_set_add_policy(client: &BuzzClient, policy: &str) -> Result<(), CliError> { - match policy { - "anyone" | "owner_only" | "nobody" => {} - _ => { - return Err(CliError::Usage(format!( - "--policy must be 'anyone', 'owner_only', or 'nobody' (got: {policy})" - ))) - } - } + agent_profile::validate_add_policy(policy)?; // Check if this policy is allowed by the deployment. // NOTE: This gate covers only the `buzz channels set-add-policy` CLI path. @@ -1032,16 +1033,11 @@ pub async fn cmd_set_add_policy(client: &BuzzClient, policy: &str) -> Result<(), } } - let content = serde_json::json!({ "channel_add_policy": policy }).to_string(); - use nostr::{EventBuilder, Kind}; - let builder = EventBuilder::new( - Kind::Custom(buzz_sdk::kind::KIND_AGENT_PROFILE as u16), - &content, - ) - .tags([]); - let event = client.sign_event(builder)?; - - let resp = client.submit_event(event).await?; + let update = agent_profile::ProfileUpdate { + channel_add_policy: Some(policy.to_string()), + ..Default::default() + }; + let resp = agent_profile::apply_profile_update(client, &update).await?; println!("{}", normalize_write_response(&resp)); Ok(()) } diff --git a/crates/buzz-cli/src/commands/keys.rs b/crates/buzz-cli/src/commands/keys.rs new file mode 100644 index 0000000000..53460313fb --- /dev/null +++ b/crates/buzz-cli/src/commands/keys.rs @@ -0,0 +1,239 @@ +//! `buzz keys` subcommands — local Nostr identity operations. +//! +//! These commands run entirely on the machine that invokes them. They make no +//! relay request and, unlike every other subcommand, do not require +//! `BUZZ_PRIVATE_KEY` to already be set — `keys generate` is how that value +//! comes into existence in the first place. +//! +//! This matters for self-hosted agents. An agent that runs on its own machine +//! should mint its own identity there, so the secret is created where it will +//! be used and never has to be transported from somewhere else. Generating the +//! key on an operator workstation and copying it to the agent host inverts +//! that: the secret exists in two places, and the operator's machine becomes a +//! custodian of an identity it does not run. + +use std::fs::OpenOptions; +use std::io::Write; +use std::path::{Path, PathBuf}; + +use nostr::{Keys, ToBech32}; + +use crate::error::CliError; + +/// Permission bits for a freshly written secret-key file: owner read/write only. +#[cfg(unix)] +const SECRET_FILE_MODE: u32 = 0o600; + +/// Run `buzz keys generate`. +/// +/// Mints a fresh secp256k1 keypair and reports the **public** half on stdout. +/// The secret half is written to `out` and is printed only when `stdout_secret` +/// is set — an explicit opt-in for callers that pipe into their own secret +/// store rather than a file. +/// +/// `force` permits overwriting an existing `out` path. Without it an existing +/// file is an error: re-running a connect flow must not be able to silently +/// destroy the identity a live agent is already using, which would orphan +/// every message that agent has ever signed. +pub fn cmd_generate(out: Option<&str>, stdout_secret: bool, force: bool) -> Result<(), CliError> { + if out.is_none() && !stdout_secret { + return Err(CliError::Usage( + "no destination for the generated secret key: pass --out to write \ + it to a file, or --stdout to print it" + .into(), + )); + } + + let keys = Keys::generate(); + let pubkey = keys.public_key(); + let npub = pubkey + .to_bech32() + .map_err(|e| CliError::Other(format!("failed to encode npub: {e}")))?; + let nsec = keys + .secret_key() + .to_bech32() + .map_err(|e| CliError::Other(format!("failed to encode nsec: {e}")))?; + + let written = match out { + Some(path) => Some(write_secret_file(Path::new(path), &nsec, force)?), + None => None, + }; + + // Ordering is deliberate: the file is on disk before anything is printed, + // so a caller that reads stdout and then reads the path can never observe + // a pubkey whose secret was not persisted. + let mut report = serde_json::json!({ + "pubkey": pubkey.to_hex(), + "npub": npub, + }); + if let Some(path) = &written { + report["secret_key_path"] = serde_json::json!(path.display().to_string()); + } + if stdout_secret { + report["nsec"] = serde_json::json!(nsec); + } + println!("{report}"); + Ok(()) +} + +/// Create `path` and write `nsec` to it with owner-only permissions. +/// +/// The file is created with its restrictive mode from the outset via +/// `OpenOptions::mode` rather than being chmod-ed afterwards — a +/// create-then-chmod sequence leaves a window in which the secret is on disk +/// world-readable. +/// +/// Returns the canonical path that was written. +fn write_secret_file(path: &Path, nsec: &str, force: bool) -> Result { + if let Some(parent) = path.parent() { + if !parent.as_os_str().is_empty() && !parent.exists() { + return Err(CliError::Usage(format!( + "directory does not exist: {}", + parent.display() + ))); + } + } + + let mut options = OpenOptions::new(); + options.write(true); + if force { + options.create(true).truncate(true); + } else { + // create_new fails if the path exists, which is the guard we want — + // and it is atomic, so two concurrent generates cannot both believe + // they created the file. + options.create_new(true); + } + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(SECRET_FILE_MODE); + } + + let mut file = options.open(path).map_err(|e| match e.kind() { + std::io::ErrorKind::AlreadyExists => CliError::Usage(format!( + "refusing to overwrite existing key file: {} (pass --force to replace it, \ + but note that any agent already using this identity will lose it)", + path.display() + )), + _ => CliError::Other(format!("failed to create {}: {e}", path.display())), + })?; + + // `--force` reuses an existing inode, whose mode is whatever it already + // was; `OpenOptions::mode` only applies on creation. Re-assert the mode so + // the overwrite path cannot leave a permissive file behind. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + file.set_permissions(std::fs::Permissions::from_mode(SECRET_FILE_MODE)) + .map_err(|e| { + CliError::Other(format!( + "failed to set permissions on {}: {e}", + path.display() + )) + })?; + } + + writeln!(file, "{nsec}") + .map_err(|e| CliError::Other(format!("failed to write {}: {e}", path.display())))?; + file.sync_all() + .map_err(|e| CliError::Other(format!("failed to flush {}: {e}", path.display())))?; + + Ok(path.canonicalize().unwrap_or_else(|_| path.to_path_buf())) +} + +pub fn dispatch(cmd: crate::KeysCmd) -> Result<(), CliError> { + use crate::KeysCmd; + match cmd { + KeysCmd::Generate { out, stdout, force } => cmd_generate(out.as_deref(), stdout, force), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn requires_a_destination() { + // Neither --out nor --stdout: the secret would be generated and + // immediately discarded, which is never what the caller meant. + let err = cmd_generate(None, false, false).expect_err("expected usage error"); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn writes_secret_file_with_owner_only_mode() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + let written = write_secret_file(&path, "nsec1test", false).unwrap(); + + let contents = std::fs::read_to_string(&written).unwrap(); + assert_eq!(contents.trim(), "nsec1test"); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&written).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, SECRET_FILE_MODE); + } + } + + #[test] + fn refuses_to_overwrite_without_force() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + write_secret_file(&path, "nsec1original", false).unwrap(); + + let err = write_secret_file(&path, "nsec1replacement", false) + .expect_err("expected overwrite refusal"); + assert!(matches!(err, CliError::Usage(_))); + + // The original identity survives the refused write. + let contents = std::fs::read_to_string(&path).unwrap(); + assert_eq!(contents.trim(), "nsec1original"); + } + + #[test] + fn force_overwrites_and_keeps_owner_only_mode() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + write_secret_file(&path, "nsec1original", false).unwrap(); + + // Loosen the mode so the re-assert has something to correct. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap(); + } + + write_secret_file(&path, "nsec1replacement", true).unwrap(); + let contents = std::fs::read_to_string(&path).unwrap(); + assert_eq!(contents.trim(), "nsec1replacement"); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, SECRET_FILE_MODE); + } + } + + #[test] + fn rejects_missing_parent_directory() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("no-such-dir").join("agent.nsec"); + let err = write_secret_file(&path, "nsec1test", false).expect_err("expected usage error"); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn generated_secret_round_trips_to_the_reported_pubkey() { + // The whole point of the command is that the caller can later load the + // written secret and arrive at the pubkey that was printed. Prove the + // encode/parse pair agrees rather than trusting it. + let keys = Keys::generate(); + let nsec = keys.secret_key().to_bech32().unwrap(); + let reloaded = Keys::parse(&nsec).unwrap(); + assert_eq!(reloaded.public_key(), keys.public_key()); + } +} diff --git a/crates/buzz-cli/src/commands/listen.rs b/crates/buzz-cli/src/commands/listen.rs new file mode 100644 index 0000000000..1e23180e01 --- /dev/null +++ b/crates/buzz-cli/src/commands/listen.rs @@ -0,0 +1,795 @@ +//! Persistent external-agent event stream. +//! +//! `buzz listen` prints one newline-delimited JSON record per matching relay +//! event. With `--envelope v1`, lifecycle records use the same stdout stream. +//! Human diagnostics remain on stderr. + +use std::collections::HashSet; +use std::io::Write; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use serde_json::json; +use uuid::Uuid; + +use crate::client::{extract_d_tag, normalize_events, BuzzClient}; +use crate::error::CliError; +use crate::validate::parse_uuid; + +/// Default kinds for channel traffic. Matches `messages get`. +const DEFAULT_KINDS: &[u32] = &[9, 40002, 40008, 45001, 45003]; +const MAX_LISTEN_CHANNELS: usize = 1024; + +/// How often `--dms` re-checks for newly opened DM conversations. DM-created +/// discovery events do not reach live global subscriptions (channel-scoped +/// fan-out), so the listener periodically polls for them. +const DM_POLL_INTERVAL: Duration = Duration::from_secs(30); + +/// Filter that finds the caller's DM conversations: relay-emitted +/// `KIND_DM_CREATED` metadata p-tags every participant and carries the stable +/// DM channel UUID as its `d` tag. Same query `buzz dms list` uses. +pub(crate) fn dm_discovery_filter(my_pubkey_hex: &str) -> serde_json::Value { + json!({ + "kinds": [buzz_core::kind::KIND_DM_CREATED], + "#p": [my_pubkey_hex], + "limit": 200, + }) +} + +async fn resolve_dm_channels( + client: &BuzzClient, + my_pubkey_hex: &str, +) -> Result, CliError> { + let events = client.query_all(dm_discovery_filter(my_pubkey_hex)).await?; + Ok(channel_ids_from_metadata(&events)) +} + +pub(crate) fn parse_kinds(raw: Option<&str>) -> Result, CliError> { + match raw { + None => Ok(DEFAULT_KINDS.to_vec()), + Some(s) if s.trim().is_empty() => Ok(DEFAULT_KINDS.to_vec()), + Some(s) => { + let mut kinds = Vec::new(); + for part in s.split(',') { + let part = part.trim(); + if part.is_empty() { + continue; + } + let kind = part + .parse::() + .map_err(|_| CliError::Usage(format!("invalid kind in --kinds: {part}")))?; + kinds.push(kind); + } + if kinds.is_empty() { + return Err(CliError::Usage( + "--kinds must list at least one kind".into(), + )); + } + Ok(kinds) + } + } +} + +/// Build REQ filters for `buzz listen`. +/// +/// Each channel gets its own filter and relay subscription. Buzz deliberately +/// excludes logically global subscriptions from live channel fan-out, so a +/// multi-value `#h` filter cannot implement live multi-channel listening. +pub(crate) fn build_listen_filters( + channels: &[String], + mentions_of_me: bool, + my_pubkey_hex: &str, + kinds: &[u32], + since: Option, +) -> Result, CliError> { + if channels.is_empty() { + return Err(CliError::Usage( + "buzz listen requires at least one resolved channel".into(), + )); + } + let mut unique_channels = Vec::new(); + let mut seen = HashSet::new(); + for channel in channels { + let channel = parse_uuid(channel)?.to_string(); + if seen.insert(channel.clone()) { + unique_channels.push(channel); + } + } + if unique_channels.len() > MAX_LISTEN_CHANNELS { + return Err(CliError::Usage(format!( + "buzz listen supports at most {MAX_LISTEN_CHANNELS} channels" + ))); + } + + Ok(unique_channels + .into_iter() + .map(|channel| { + let mut filter = json!({ + "kinds": kinds, + "#h": [channel], + }); + if mentions_of_me { + filter["#p"] = json!([my_pubkey_hex]); + } + if let Some(since) = since { + filter["since"] = json!(since); + } + filter + }) + .collect()) +} + +fn channel_ids_from_metadata(events: &[serde_json::Value]) -> Vec { + let mut channels: Vec = events + .iter() + .filter_map(|event| parse_uuid(&extract_d_tag(event)).ok()) + .map(|channel| channel.to_string()) + .collect(); + channels.sort_unstable(); + channels.dedup(); + channels +} + +async fn resolve_listen_channels( + client: &BuzzClient, + channels: Vec, + mentions_of_me: bool, +) -> Result, CliError> { + if !channels.is_empty() { + return Ok(channels); + } + if !mentions_of_me { + return Err(CliError::Usage( + "buzz listen requires --channel , --mentions-of-me, and/or --dms".into(), + )); + } + + let metadata = client.query_all(json!({"kinds": [39000]})).await?; + let channels = channel_ids_from_metadata(&metadata); + if channels.is_empty() { + return Err(CliError::NotFound( + "no visible channels available for --mentions-of-me".into(), + )); + } + Ok(channels) +} + +fn http_to_ws(http_url: &str) -> String { + http_url + .replace("https://", "wss://") + .replace("http://", "ws://") +} + +fn lifecycle_record(state: &str, message: Option<&str>) -> serde_json::Value { + let mut record = json!({ + "schema_version": 1, + "type": "lifecycle", + "state": state, + }); + if let Some(message) = message { + record["message"] = json!(message); + } + record +} + +pub(crate) fn event_record( + event: serde_json::Value, + envelope: crate::ListenEnvelope, +) -> serde_json::Value { + match envelope { + crate::ListenEnvelope::Flat => event, + crate::ListenEnvelope::V1 => json!({ + "schema_version": 1, + "type": "event", + "event": event, + }), + } +} + +fn write_stdout_record(record: &serde_json::Value) -> Result<(), CliError> { + let line = serde_json::to_string(record) + .map_err(|e| CliError::Other(format!("serialize listen record: {e}")))?; + let mut stdout = std::io::stdout().lock(); + writeln!(stdout, "{line}").map_err(|e| CliError::Other(format!("stdout write: {e}")))?; + stdout + .flush() + .map_err(|e| CliError::Other(format!("stdout flush: {e}")))?; + Ok(()) +} + +fn write_lifecycle( + envelope: crate::ListenEnvelope, + state: &str, + message: Option<&str>, +) -> Result<(), CliError> { + if matches!(envelope, crate::ListenEnvelope::V1) { + write_stdout_record(&lifecycle_record(state, message))?; + } + Ok(()) +} + +fn spawn_shutdown_watcher(running: Arc) { + tokio::spawn(async move { + wait_for_shutdown_signal().await; + running.store(false, Ordering::SeqCst); + }); +} + +#[cfg(unix)] +async fn wait_for_shutdown_signal() { + use tokio::signal::unix::{signal, SignalKind}; + + let interrupt = signal(SignalKind::interrupt()); + let terminate = signal(SignalKind::terminate()); + + match (interrupt, terminate) { + (Ok(mut interrupt), Ok(mut terminate)) => { + tokio::select! { + _ = interrupt.recv() => {} + _ = terminate.recv() => {} + } + } + _ => { + let _ = tokio::signal::ctrl_c().await; + } + } +} + +#[cfg(not(unix))] +async fn wait_for_shutdown_signal() { + let _ = tokio::signal::ctrl_c().await; +} + +fn map_ws_error(context: &str, error: buzz_ws_client::WsClientError) -> CliError { + use buzz_ws_client::WsClientError; + + let detail = error.to_string(); + match error { + WsClientError::WebSocket(_) | WsClientError::Timeout | WsClientError::ConnectionClosed => { + CliError::Transport(format!("{context}: {detail}")) + } + WsClientError::AuthFailed(_) | WsClientError::NoAuthChallenge => { + CliError::Auth(format!("{context}: {detail}")) + } + WsClientError::Url(_) => CliError::Usage(format!("{context}: {detail}")), + WsClientError::EventBuilder(_) => CliError::Key(format!("{context}: {detail}")), + WsClientError::Json(_) + | WsClientError::UnexpectedMessage(_) + | WsClientError::EventRejected(_) => { + CliError::Other(format!("{context}: protocol error: {detail}")) + } + } +} + +fn relay_closed_error(message: &str) -> CliError { + if message.starts_with("auth-required:") || message.starts_with("restricted:") { + CliError::Auth(format!("subscription closed: {message}")) + } else { + CliError::Other(format!("subscription closed: {message}")) + } +} + +async fn sleep_with_shutdown(duration: Duration, running: &AtomicBool) { + let deadline = tokio::time::Instant::now() + duration; + while running.load(Ordering::SeqCst) { + let Some(remaining) = deadline.checked_duration_since(tokio::time::Instant::now()) else { + break; + }; + tokio::time::sleep(remaining.min(Duration::from_millis(100))).await; + } +} + +/// Per-session context for `--dms`: which DM conversations are already +/// subscribed, and everything needed to subscribe a newly discovered one. +struct DmWatch { + my_pubkey: String, + kinds: Vec, + mentions_of_me: bool, + since: Option, + known: HashSet, +} + +/// Resolve DM channels for this session and build the final filter set. +/// +/// Runs per session (not once at startup) so a reconnect keeps DM +/// conversations that were discovered mid-session. +async fn build_session_filters( + client: &BuzzClient, + channels: &[String], + dms: bool, + mentions_of_me: bool, + my_pubkey: &str, + kinds: &[u32], + since: Option, +) -> Result<(Vec, Option), CliError> { + let mut session_channels = channels.to_vec(); + let mut dm_watch = None; + if dms { + let dm_channels = resolve_dm_channels(client, my_pubkey).await?; + session_channels.extend(dm_channels); + dm_watch = Some(DmWatch { + my_pubkey: my_pubkey.to_string(), + kinds: kinds.to_vec(), + mentions_of_me, + since, + known: session_channels + .iter() + .filter_map(|c| parse_uuid(c).ok()) + .map(|c| c.to_string()) + .collect(), + }); + } + // With --dms an empty start is valid: the agent may simply have no DM + // conversations yet; the poll subscribes them as they are opened. + let filters = if session_channels.is_empty() && dms { + Vec::new() + } else { + build_listen_filters(&session_channels, mentions_of_me, my_pubkey, kinds, since)? + }; + Ok((filters, dm_watch)) +} + +/// Run the listen loop until shutdown or fatal error. +#[allow(clippy::too_many_arguments)] +pub async fn cmd_listen( + client: &BuzzClient, + channels: Vec, + mentions_of_me: bool, + dms: bool, + kinds_raw: Option, + since: Option, + envelope: crate::ListenEnvelope, + reconnect: bool, +) -> Result<(), CliError> { + let kinds = parse_kinds(kinds_raw.as_deref())?; + let my_pubkey = client.keys().public_key().to_hex(); + let channels = if channels.is_empty() && dms && !mentions_of_me { + // --dms alone is a complete subscription request. + Vec::new() + } else { + resolve_listen_channels(client, channels, mentions_of_me).await? + }; + let ws_url = http_to_ws(client.relay_url()); + let running = Arc::new(AtomicBool::new(true)); + spawn_shutdown_watcher(running.clone()); + + let mut backoff_ms = 500_u64; + const MAX_BACKOFF_MS: u64 = 30_000; + + while running.load(Ordering::SeqCst) { + let session = async { + let (filters, dm_watch) = build_session_filters( + client, + &channels, + dms, + mentions_of_me, + &my_pubkey, + &kinds, + since, + ) + .await?; + listen_session( + client, + &ws_url, + &filters, + envelope, + running.clone(), + dm_watch, + ) + .await + }; + match session.await { + Ok(()) => { + if !running.load(Ordering::SeqCst) || !reconnect { + break; + } + } + Err(error) => { + if !running.load(Ordering::SeqCst) { + break; + } + if !reconnect || !crate::error::is_retryable_error(&error) { + let _ = write_lifecycle(envelope, "fatal", Some(&error.to_string())); + return Err(error); + } + eprintln!( + "{}", + json!({ + "error": "listen_reconnect", + "message": error.to_string(), + "backoff_ms": backoff_ms, + }) + ); + } + } + + if !reconnect || !running.load(Ordering::SeqCst) { + break; + } + sleep_with_shutdown(Duration::from_millis(backoff_ms), running.as_ref()).await; + backoff_ms = backoff_ms.saturating_mul(2).min(MAX_BACKOFF_MS); + } + + Ok(()) +} + +async fn listen_session( + client: &BuzzClient, + ws_url: &str, + filters: &[serde_json::Value], + envelope: crate::ListenEnvelope, + running: Arc, + mut dm_watch: Option, +) -> Result<(), CliError> { + use buzz_ws_client::{NostrWsConnection, RelayMessage}; + + let mut conn = + NostrWsConnection::connect_authenticated(ws_url, client.keys(), client.auth_tag()) + .await + .map_err(|error| map_ws_error("websocket connect", error))?; + + write_lifecycle(envelope, "connected", None)?; + + let subscriptions: Vec<(String, serde_json::Value)> = filters + .iter() + .map(|filter| { + ( + format!("buzz-listen-{}", &Uuid::new_v4().to_string()[..8]), + filter.clone(), + ) + }) + .collect(); + let mut subscription_ids: HashSet = subscriptions + .iter() + .map(|(sub_id, _)| sub_id.clone()) + .collect(); + let mut awaiting_eose = subscription_ids.clone(); + let mut eose_emitted = false; + + for (sub_id, filter) in &subscriptions { + conn.send_raw(&json!(["REQ", sub_id, filter])) + .await + .map_err(|error| map_ws_error("websocket subscribe", error))?; + } + + // A --dms-only session can legitimately start with zero subscriptions + // (no DM conversations yet). Nothing will send an EOSE, so declare + // readiness here rather than leaving an envelope consumer waiting. + if subscriptions.is_empty() && !eose_emitted { + write_lifecycle(envelope, "eose", None)?; + eose_emitted = true; + } + + let mut next_dm_poll = tokio::time::Instant::now() + DM_POLL_INTERVAL; + + while running.load(Ordering::SeqCst) { + if let Some(watch) = dm_watch.as_mut() { + if tokio::time::Instant::now() >= next_dm_poll { + next_dm_poll = tokio::time::Instant::now() + DM_POLL_INTERVAL; + match resolve_dm_channels(client, &watch.my_pubkey).await { + Ok(discovered) => { + for channel in discovered { + if !watch.known.insert(channel.clone()) { + continue; + } + let filter = build_listen_filters( + std::slice::from_ref(&channel), + watch.mentions_of_me, + &watch.my_pubkey, + &watch.kinds, + watch.since, + )? + .remove(0); + let sub_id = + format!("buzz-listen-{}", &Uuid::new_v4().to_string()[..8]); + conn.send_raw(&json!(["REQ", sub_id, filter])) + .await + .map_err(|error| map_ws_error("websocket subscribe", error))?; + subscription_ids.insert(sub_id); + write_lifecycle(envelope, "dm_channel_added", Some(&channel))?; + } + } + Err(error) => { + // The websocket is still healthy; a failed discovery + // poll is diagnosable but not fatal. + eprintln!( + "{}", + json!({"error": "dm_poll_failed", "message": error.to_string()}) + ); + } + } + } + } + + let msg = match conn.next_event(Duration::from_millis(500)).await { + Ok(msg) => msg, + Err(buzz_ws_client::WsClientError::Timeout) => continue, + Err(error) => return Err(map_ws_error("websocket receive", error)), + }; + + match msg { + RelayMessage::Event { + subscription_id, + event, + } => { + if !subscription_ids.contains(&subscription_id) { + return Err(CliError::Other(format!( + "websocket receive: protocol error: unknown subscription {subscription_id}" + ))); + } + let raw = serde_json::to_value(event.as_ref()) + .map_err(|e| CliError::Other(format!("event serialize: {e}")))?; + let normalized = normalize_events(std::slice::from_ref(&raw)); + let events = serde_json::from_str::>(&normalized) + .unwrap_or_else(|_| vec![raw]); + let event = events.into_iter().next().unwrap_or_else(|| json!({})); + write_stdout_record(&event_record(event, envelope))?; + } + RelayMessage::Eose { subscription_id } => { + if awaiting_eose.remove(&subscription_id) + && awaiting_eose.is_empty() + && !eose_emitted + { + write_lifecycle(envelope, "eose", None)?; + eose_emitted = true; + } + } + RelayMessage::Closed { + subscription_id, + message, + } => { + if !subscription_ids.contains(&subscription_id) { + return Err(CliError::Other(format!( + "websocket receive: protocol error: unknown subscription {subscription_id}" + ))); + } + write_lifecycle(envelope, "closed", Some(&message))?; + return Err(relay_closed_error(&message)); + } + RelayMessage::Notice { message } => { + eprintln!("{}", json!({"notice": message})); + } + RelayMessage::Ok(_) | RelayMessage::Auth { .. } | RelayMessage::Count { .. } => {} + } + } + + for sub_id in subscription_ids { + let _ = conn.send_raw(&json!(["CLOSE", sub_id])).await; + } + let _ = conn.disconnect().await; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn requires_channel_or_mentions() { + let err = + build_listen_filters(&[], false, &"a".repeat(64), DEFAULT_KINDS, None).unwrap_err(); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn channel_filter_uses_h_tag() { + let channel = "11111111-1111-1111-1111-111111111111".to_string(); + let filters = build_listen_filters( + std::slice::from_ref(&channel), + false, + &"a".repeat(64), + DEFAULT_KINDS, + None, + ) + .unwrap(); + + assert_eq!(filters.len(), 1); + assert_eq!(filters[0]["#h"][0], channel); + assert!(filters[0].get("#p").is_none()); + } + + #[test] + fn mentions_filter_uses_p_tag() { + let pubkey = "b".repeat(64); + let channel = "11111111-1111-1111-1111-111111111111".to_string(); + let filters = build_listen_filters( + std::slice::from_ref(&channel), + true, + &pubkey, + DEFAULT_KINDS, + None, + ) + .unwrap(); + + assert_eq!(filters.len(), 1); + assert_eq!(filters[0]["#p"][0], pubkey); + assert_eq!(filters[0]["#h"][0], channel); + } + + #[test] + fn channel_and_mentions_are_single_and_filter() { + let channel = "11111111-1111-1111-1111-111111111111".to_string(); + let pubkey = "c".repeat(64); + let filters = build_listen_filters( + std::slice::from_ref(&channel), + true, + &pubkey, + DEFAULT_KINDS, + None, + ) + .unwrap(); + + assert_eq!(filters.len(), 1); + assert_eq!(filters[0]["#h"][0], channel); + assert_eq!(filters[0]["#p"][0], pubkey); + } + + #[test] + fn multiple_channels_get_independent_filters() { + let channels = vec![ + "11111111-1111-1111-1111-111111111111".to_string(), + "22222222-2222-2222-2222-222222222222".to_string(), + ]; + let filters = + build_listen_filters(&channels, true, &"d".repeat(64), DEFAULT_KINDS, None).unwrap(); + + assert_eq!(filters.len(), 2); + assert_eq!(filters[0]["#h"], json!([channels[0]])); + assert_eq!(filters[1]["#h"], json!([channels[1]])); + assert_eq!(filters[0]["#p"], json!(["d".repeat(64)])); + assert_eq!(filters[1]["#p"], json!(["d".repeat(64)])); + } + + #[test] + fn duplicate_channels_are_canonicalized_and_deduplicated() { + let filters = build_listen_filters( + &[ + "11111111-1111-1111-1111-111111111111".to_string(), + "11111111111111111111111111111111".to_string(), + ], + false, + &"a".repeat(64), + DEFAULT_KINDS, + None, + ) + .unwrap(); + + assert_eq!(filters.len(), 1); + assert_eq!( + filters[0]["#h"], + json!(["11111111-1111-1111-1111-111111111111"]) + ); + } + + #[test] + fn metadata_channel_ids_are_valid_sorted_and_unique() { + let events = json!([ + {"tags": [["d", "22222222-2222-2222-2222-222222222222"]]}, + {"tags": [["d", "not-a-channel"]]}, + {"tags": [["d", "11111111111111111111111111111111"]]}, + {"tags": [["d", "22222222-2222-2222-2222-222222222222"]]} + ]); + + assert_eq!( + channel_ids_from_metadata(events.as_array().unwrap()), + vec![ + "11111111-1111-1111-1111-111111111111", + "22222222-2222-2222-2222-222222222222", + ] + ); + } + + #[test] + fn websocket_disconnect_is_retryable_network_error() { + let error = map_ws_error( + "websocket receive", + buzz_ws_client::WsClientError::ConnectionClosed, + ); + + assert!(matches!(error, CliError::Transport(_))); + assert!(crate::error::is_retryable_error(&error)); + assert_eq!(crate::error::exit_code(&error), 2); + } + + #[test] + fn restricted_subscription_close_is_auth_error() { + let error = relay_closed_error("restricted: not a channel member"); + + assert!(matches!(error, CliError::Auth(_))); + assert_eq!(crate::error::exit_code(&error), 3); + } + + #[test] + fn since_is_added_to_filter() { + let filters = build_listen_filters( + &["11111111-1111-1111-1111-111111111111".to_string()], + false, + &"a".repeat(64), + DEFAULT_KINDS, + Some(1785100000), + ) + .unwrap(); + + assert_eq!(filters[0]["since"], 1785100000_u64); + } + + #[test] + fn parse_kinds_defaults_and_parses_csv() { + assert_eq!(parse_kinds(None).unwrap(), DEFAULT_KINDS); + assert_eq!(parse_kinds(Some("9, 40002")).unwrap(), vec![9, 40002]); + } + + #[test] + fn parse_kinds_rejects_invalid_value() { + let err = parse_kinds(Some("9,nope")).unwrap_err(); + assert!(matches!(err, CliError::Usage(_))); + } + + #[test] + fn v1_event_record_wraps_flat_event() { + let event = json!({ + "id": "event", + "pubkey": "author", + "kind": 40002, + "content": "hello", + "created_at": 1785100000_u64, + "tags": [["h", "channel"]], + }); + + let record = event_record(event.clone(), crate::ListenEnvelope::V1); + + assert_eq!(record["schema_version"], 1); + assert_eq!(record["type"], "event"); + assert_eq!(record["event"], event); + } + + #[test] + fn flat_event_record_is_unchanged() { + let event = json!({"id": "event"}); + assert_eq!( + event_record(event.clone(), crate::ListenEnvelope::Flat), + event + ); + } + + #[test] + fn dm_discovery_filter_targets_dm_created_p_tagged_to_me() { + let me = "c".repeat(64); + let filter = dm_discovery_filter(&me); + + assert_eq!( + filter["kinds"][0], + buzz_core::kind::KIND_DM_CREATED, + "must query the relay-emitted DM-created metadata kind" + ); + assert_eq!(filter["#p"][0], me); + } + + #[test] + fn dm_channel_ids_come_from_d_tags_and_dedupe() { + let dm = "22222222-2222-2222-2222-222222222222"; + let events = vec![ + json!({"tags": [["d", dm], ["p", "aa"]]}), + json!({"tags": [["d", dm]]}), + json!({"tags": [["p", "bb"]]}), + json!({"tags": [["d", "not-a-uuid"]]}), + ]; + + assert_eq!(channel_ids_from_metadata(&events), vec![dm.to_string()]); + } + + #[test] + fn lifecycle_record_uses_v1_shape() { + let record = lifecycle_record("eose", None); + + assert_eq!( + record, + json!({ + "schema_version": 1, + "type": "lifecycle", + "state": "eose", + }) + ); + } +} diff --git a/crates/buzz-cli/src/commands/messages.rs b/crates/buzz-cli/src/commands/messages.rs index 40a9ae80b5..7adac80c3f 100644 --- a/crates/buzz-cli/src/commands/messages.rs +++ b/crates/buzz-cli/src/commands/messages.rs @@ -339,8 +339,11 @@ fn format_events(normalized: &str, format: &crate::OutputFormat) -> String { .map(|e| { serde_json::json!({ "id": e.get("id").cloned().unwrap_or_default(), + "pubkey": e.get("pubkey").cloned().unwrap_or_default(), + "kind": e.get("kind").cloned().unwrap_or_default(), "content": e.get("content").cloned().unwrap_or_default(), "created_at": e.get("created_at").cloned().unwrap_or_default(), + "tags": e.get("tags").cloned().unwrap_or_default(), }) }) .collect(); @@ -993,8 +996,8 @@ pub async fn dispatch( #[cfg(test)] mod tests { use super::{ - event_mention_pubkeys, find_root_from_tags, match_profiles_by_name, merge_message_mentions, - missing_members, normalize_explicit_mentions, parse_member_pubkeys, + event_mention_pubkeys, find_root_from_tags, format_events, match_profiles_by_name, + merge_message_mentions, missing_members, normalize_explicit_mentions, parse_member_pubkeys, resolve_names_to_pubkeys, }; use buzz_sdk::mentions::{ @@ -1012,6 +1015,42 @@ mod tests { const PK_VALID_B: &str = "c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05"; const PK_VALID_C: &str = "f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68"; + #[test] + fn compact_message_output_keeps_adapter_contract_fields() { + let normalized = serde_json::json!([ + { + "id": ID_A, + "pubkey": PUBKEY, + "kind": 40002, + "content": "hello @agent", + "created_at": 1785100000_u64, + "tags": [ + ["h", "00000000-0000-0000-0000-000000000000"], + ["p", PK_VALID_A] + ], + "sig": "not-in-compact" + } + ]) + .to_string(); + + let output = format_events(&normalized, &crate::OutputFormat::Compact); + let compact: serde_json::Value = serde_json::from_str(&output).unwrap(); + + assert_eq!(compact[0]["id"], ID_A); + assert_eq!(compact[0]["pubkey"], PUBKEY); + assert_eq!(compact[0]["kind"], 40002); + assert_eq!(compact[0]["content"], "hello @agent"); + assert_eq!(compact[0]["created_at"], 1785100000_u64); + assert_eq!( + compact[0]["tags"], + json!([ + ["h", "00000000-0000-0000-0000-000000000000"], + ["p", PK_VALID_A] + ]) + ); + assert!(compact[0].get("sig").is_none()); + } + #[test] fn root_marker_wins_over_reply_marker() { let tags = json!([ diff --git a/crates/buzz-cli/src/commands/mod.rs b/crates/buzz-cli/src/commands/mod.rs index 8691590636..f53438f99c 100644 --- a/crates/buzz-cli/src/commands/mod.rs +++ b/crates/buzz-cli/src/commands/mod.rs @@ -1,3 +1,4 @@ +pub mod agent_profile; pub mod agents; pub mod channel_templates; pub mod channels; @@ -5,6 +6,8 @@ pub mod dms; pub mod emoji; pub mod feed; pub mod issues; +pub mod keys; +pub mod listen; pub mod mem; pub mod messages; pub mod moderation; diff --git a/crates/buzz-cli/src/commands/users.rs b/crates/buzz-cli/src/commands/users.rs index 7c15d285a0..5f321b8757 100644 --- a/crates/buzz-cli/src/commands/users.rs +++ b/crates/buzz-cli/src/commands/users.rs @@ -4,9 +4,35 @@ use nostr::PublicKey; use crate::client::{extract_d_tag, normalize_write_response, BuzzClient}; use crate::error::CliError; use crate::validate::validate_hex64; +use nostr::ToBech32; // TODO(phase-4): Replace raw nostr::EventBuilder usage in cmd_set_presence with buzz-sdk builder +/// Build identity JSON for `buzz users me`. +/// +/// Kept pure so the no-relay identity contract stays easy to test. +pub(crate) fn me_identity_json(pubkey_hex: &str, npub_bech32: &str) -> serde_json::Value { + serde_json::json!({ + "pubkey": pubkey_hex, + "npub": npub_bech32, + }) +} + +/// Print the active CLI identity from the loaded private key. +/// +/// This performs no relay I/O, so resident adapters can validate their local +/// Buzz identity before opening subscriptions or sending messages. +pub async fn cmd_me(client: &BuzzClient, _format: &crate::OutputFormat) -> Result<(), CliError> { + let public_key = client.keys().public_key(); + let pubkey = public_key.to_hex(); + let npub = public_key + .to_bech32() + .map_err(|e| CliError::Other(format!("npub encode failed: {e}")))?; + + println!("{}", me_identity_json(&pubkey, &npub)); + Ok(()) +} + /// Get user profiles (kind:0 metadata events). /// /// - 0 pubkeys, no name → query our own profile @@ -536,6 +562,7 @@ pub async fn dispatch( ) -> Result<(), CliError> { use crate::UsersCmd; match cmd { + UsersCmd::Me => cmd_me(client, format).await, UsersCmd::Get { pubkeys, name, @@ -574,8 +601,8 @@ pub async fn dispatch( #[cfg(test)] mod tests { use super::{ - owned_agent_pubkeys_from_events, owner_scoped_profiles, owner_verification, - presence_subject, + me_identity_json, owned_agent_pubkeys_from_events, owner_scoped_profiles, + owner_verification, presence_subject, }; use nostr::Keys; use serde_json::json; @@ -745,6 +772,24 @@ mod tests { assert!(profiles[2].get("owner_pubkey").is_none()); } + #[test] + fn me_identity_json_contains_public_identity_only() { + let value = me_identity_json( + "35c18ae273fccfaf80d629e20e7f8721b90499379addff533054acc2504c12b4", + "npub1example", + ); + + assert_eq!( + value, + json!({ + "pubkey": "35c18ae273fccfaf80d629e20e7f8721b90499379addff533054acc2504c12b4", + "npub": "npub1example", + }) + ); + assert!(value.get("private_key").is_none()); + assert!(value.get("nsec").is_none()); + } + #[test] fn presence_subject_uses_p_tag() { let event = json!({"pubkey": "relay", "tags": [["p", "user"]]}); diff --git a/crates/buzz-cli/src/error.rs b/crates/buzz-cli/src/error.rs index 2edcd6aa9d..9a89b6772b 100644 --- a/crates/buzz-cli/src/error.rs +++ b/crates/buzz-cli/src/error.rs @@ -14,6 +14,10 @@ pub enum CliError { #[error("network error: {}", fmt_reqwest_error(.0))] Network(#[from] reqwest::Error), + /// Non-HTTP transport failure, such as a WebSocket disconnect or timeout. + #[error("network error: {0}")] + Transport(String), + /// Auth missing or rejected (401/403) #[error("auth error: {0}")] Auth(String), @@ -78,6 +82,7 @@ pub fn is_retryable_error(e: &CliError) -> bool { || net_err.is_body() || net_err.is_decode() } + CliError::Transport(_) => true, CliError::Relay { status, .. } => matches!(status, 429 | 502 | 503 | 504), CliError::DeliveryUnknown(_) => false, _ => false, @@ -98,6 +103,7 @@ pub fn exit_code(e: &CliError) -> i32 { } } CliError::Network(_) => 2, + CliError::Transport(_) => 2, CliError::Auth(_) => 3, CliError::Key(_) => 3, CliError::Conflict(_) => 5, @@ -120,6 +126,7 @@ pub fn print_error(e: &CliError) { } } CliError::Network(_) => "network_error", + CliError::Transport(_) => "network_error", CliError::Auth(_) => "auth_error", CliError::Key(_) => "key_error", CliError::Conflict(_) => "conflict", @@ -181,6 +188,9 @@ mod tests { #[test] fn other_errors_are_not_retryable() { assert!(!is_retryable_error(&CliError::Usage("bad flag".into()))); + assert!(is_retryable_error(&CliError::Transport( + "connection closed".into() + ))); assert!(!is_retryable_error(&CliError::Auth("missing key".into()))); assert!(!is_retryable_error(&CliError::Key("bad key".into()))); assert!(!is_retryable_error(&CliError::Conflict( diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 0726406d29..0cfcd0d978 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -171,8 +171,44 @@ pub enum OutputFormat { Compact, } +#[derive(Clone, Copy, clap::ValueEnum)] +pub enum ListenEnvelope { + /// Existing flat event objects, one per stdout line + #[value(name = "flat")] + Flat, + /// Versioned v1 event and lifecycle envelopes + #[value(name = "v1")] + V1, +} + #[derive(Subcommand)] enum Cmd { + /// Stream matching relay events as newline-delimited JSON + Listen { + /// Channel UUID to subscribe to. Repeat for multiple channels. + #[arg(long = "channel")] + channels: Vec, + /// Only receive events that p-tag this CLI identity + #[arg(long, default_value_t = false)] + mentions_of_me: bool, + /// Also stream your direct-message conversations, picking up newly + /// opened ones without a restart. Each DM event carries its stable + /// conversation UUID in the `h` tag. + #[arg(long, default_value_t = false)] + dms: bool, + /// Comma-separated event kinds. Defaults to Buzz message kinds. + #[arg(long)] + kinds: Option, + /// Unix timestamp lower bound for replay + #[arg(long)] + since: Option, + /// Output envelope schema + #[arg(long, value_enum, default_value_t = ListenEnvelope::Flat)] + envelope: ListenEnvelope, + /// Disable automatic reconnect with exponential backoff + #[arg(long, default_value_t = false)] + no_reconnect: bool, + }, /// Draft owner-reviewed agent creation and updates #[command(subcommand)] Agents(AgentsCmd), @@ -233,6 +269,9 @@ enum Cmd { /// Persona pack operations (local, no relay connection needed) #[command(subcommand)] Pack(PackCmd), + /// Nostr identity operations (local, no relay connection needed) + #[command(subcommand)] + Keys(KeysCmd), /// Community moderation — reports queue, bans, timeouts, audit trail #[command(subcommand)] Moderation(ModerationCmd), @@ -258,6 +297,9 @@ impl RespondToArg { #[derive(Subcommand)] pub enum AgentsCmd { + /// Read or write this identity's agent profile (kind:10100) + #[command(subcommand)] + Profile(AgentProfileCmd), /// Open a prefilled create-agent form in the owner's Buzz Desktop DraftCreate { /// Current channel UUID; the new agent is added here after save @@ -344,6 +386,48 @@ buzz agents archived" Archived, } +/// Agent profile (kind:10100) commands. +/// +/// The profile is the agent-authored directory record Buzz Desktop discovers +/// agents through. A self-hosted agent publishes its own, which is how it +/// becomes visible and mentionable in a workspace without the Desktop owning +/// or supervising the process. +#[derive(Subcommand)] +pub enum AgentProfileCmd { + /// Print this identity's current agent profile + Get, + /// Update this identity's agent profile, preserving unspecified fields + #[command( + after_help = "kind:10100 is a REPLACEABLE event: the relay keeps only the newest one \ +per author. This command therefore reads the current profile and layers your changes onto \ +it, so a partial update cannot drop the fields you did not mention.\n\n\ +`channel_add_policy` is required — it is inherited from the existing profile when present, \ +otherwise pass --policy. A profile published without it replaces the visible record while \ +leaving the relay's stored policy untouched.\n\n\ +Examples:\n \ +buzz agents profile set --display-name Scout --agent-type researcher --policy owner_only\n \ +buzz agents profile set --capabilities search,summarize\n \ +buzz agents profile set --status online" + )] + Set { + /// Display name shown in the Buzz agent directory + #[arg(long)] + display_name: Option, + /// Free-form agent type label (e.g. researcher, reviewer) + #[arg(long)] + agent_type: Option, + /// Comma-separated capability list; replaces the existing list + #[arg(long, value_delimiter = ',')] + capabilities: Option>, + /// Presence shown in the directory: online, away, or offline + #[arg(long)] + status: Option, + /// Who may add this agent to channels: anyone, owner_only, or nobody + #[arg(long)] + policy: Option, + }, +} + #[derive(Subcommand)] pub enum MessagesCmd { /// Send a message to a channel @@ -803,6 +887,8 @@ pub enum DmsCmd { #[derive(Subcommand)] pub enum UsersCmd { + /// Print the active CLI identity without contacting the relay + Me, /// Look up user profiles by pubkey or name Get { /// User pubkey(s) to look up (64-char hex). Omit for your own profile @@ -1672,6 +1758,40 @@ pub enum PackCmd { }, } +/// Local Nostr identity commands. +/// +/// These run without a relay connection and without a pre-existing +/// `BUZZ_PRIVATE_KEY`. They exist so that a self-hosted agent can mint its own +/// identity on the machine it runs on, instead of an operator generating the +/// secret elsewhere and copying it over. +#[derive(Subcommand)] +pub enum KeysCmd { + /// Generate a new Nostr keypair for a self-hosted agent identity + #[command( + after_help = "The secret key is written to --out with mode 0600 and is NOT printed \ +unless --stdout is passed. stdout always carries the public half (pubkey, npub) so the \ +caller can register the identity without ever handling the secret.\n\n\ +Run this on the machine that will use the identity — that is the whole point: the secret \ +is created where it is used and never transported.\n\n\ +Examples:\n \ +buzz keys generate --out ~/.config/buzz/agent.nsec\n \ +buzz keys generate --stdout | my-secret-store write buzz/agent\n\n\ +Verify afterwards with:\n \ +BUZZ_PRIVATE_KEY=$(cat ~/.config/buzz/agent.nsec) buzz users me" + )] + Generate { + /// Path to write the secret key to, created with mode 0600 + #[arg(long)] + out: Option, + /// Also print the secret key on stdout (for piping into a secret store) + #[arg(long)] + stdout: bool, + /// Overwrite an existing --out file; any agent using that identity loses it + #[arg(long)] + force: bool, + }, +} + /// Community moderation commands. /// /// The community (tenant) is selected by the relay host in `--relay` / @@ -1779,6 +1899,13 @@ async fn run(cli: Cli) -> Result<(), CliError> { }; } + // Keys commands are local-only AND must run before the BUZZ_PRIVATE_KEY + // requirement below: `keys generate` is how that key comes to exist, so + // demanding one first would make the command unreachable. + if let Cmd::Keys(sub) = cli.command { + return commands::keys::dispatch(sub); + } + // Auth: private key is required for all relay operations. // The keypair IS the identity — no tokens, no other auth. let private_key_str = cli.private_key.ok_or_else(|| { @@ -1806,6 +1933,27 @@ async fn run(cli: Cli) -> Result<(), CliError> { let client = BuzzClient::new(relay_url, keys, auth_tag, auth_tag_json)?; match cli.command { + Cmd::Listen { + channels, + mentions_of_me, + dms, + kinds, + since, + envelope, + no_reconnect, + } => { + commands::listen::cmd_listen( + &client, + channels, + mentions_of_me, + dms, + kinds, + since, + envelope, + !no_reconnect, + ) + .await + } Cmd::Agents(sub) => commands::agents::dispatch(sub, &client).await, Cmd::Messages(sub) => commands::messages::dispatch(sub, &client, &cli.format).await, Cmd::Channels(sub) => commands::channels::dispatch(sub, &client, &cli.format).await, @@ -1826,7 +1974,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { Cmd::Upload(sub) => commands::upload::dispatch(sub, &client).await, Cmd::Mem(sub) => commands::mem::dispatch(sub, &client).await, Cmd::Moderation(sub) => commands::moderation::dispatch(sub, &client, &cli.format).await, - Cmd::Pack(_) => unreachable!("handled above"), + Cmd::Pack(_) | Cmd::Keys(_) => unreachable!("handled above"), } } @@ -1875,6 +2023,8 @@ mod tests { "emoji", "feed", "issues", + "keys", + "listen", "media", "mem", "messages", @@ -1937,6 +2087,7 @@ mod tests { "archived", "draft-create", "draft-update", + "profile", "unarchive" ] ); @@ -1988,6 +2139,7 @@ mod tests { names(&cmd, "users"), vec![ "get", + "me", "presence", "set-presence", "set-profile", @@ -2045,6 +2197,7 @@ mod tests { assert_eq!(names(&cmd, "media"), vec!["get"]); assert_eq!(names(&cmd, "upload"), vec!["file"]); assert_eq!(names(&cmd, "pack"), vec!["inspect", "validate"]); + assert_eq!(names(&cmd, "keys"), vec!["generate"]); assert_eq!( names(&cmd, "moderation"), vec![ @@ -2063,7 +2216,7 @@ mod tests { #[test] fn subcommand_counts_are_stable() { let expected: Vec<(&str, usize)> = vec![ - ("agents", 5), + ("agents", 6), ("canvas", 2), ("channels", 16), ("dms", 4), @@ -2079,7 +2232,7 @@ mod tests { ("repos", 5), ("social", 7), ("upload", 1), - ("users", 5), + ("users", 6), ("workflows", 8), ]; diff --git a/crates/buzz-cli/tests/external_agent_cli.rs b/crates/buzz-cli/tests/external_agent_cli.rs new file mode 100644 index 0000000000..2f52a563fb --- /dev/null +++ b/crates/buzz-cli/tests/external_agent_cli.rs @@ -0,0 +1,37 @@ +use std::process::Command; + +const TEST_PRIVATE_KEY: &str = "1111111111111111111111111111111111111111111111111111111111111111"; +const TEST_PUBLIC_KEY: &str = "4f355bdcb7cc0af728ef3cceb9615d90684bb5b2ca5f859ab0f0b704075871aa"; +const TEST_NPUB: &str = "npub1fu64hh9hes90w2808n8tjc2ajp5yhddjef0ctx4s7zmsgp6cwx4qgy4eg9"; + +#[test] +fn users_me_is_local_and_never_prints_private_key() { + let output = Command::new(env!("CARGO_BIN_EXE_buzz")) + .args(["users", "me"]) + .env("BUZZ_PRIVATE_KEY", TEST_PRIVATE_KEY) + .env("BUZZ_RELAY_URL", "http://127.0.0.1:1") + .env_remove("BUZZ_AUTH_TAG") + .output() + .expect("buzz users me should start"); + + assert!( + output.status.success(), + "users me should not contact the unreachable relay: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert!( + output.stderr.is_empty(), + "users me should not write stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8 JSON"); + assert!(!stdout.contains(TEST_PRIVATE_KEY)); + assert_eq!( + serde_json::from_str::(stdout.trim()).expect("valid identity JSON"), + serde_json::json!({ + "pubkey": TEST_PUBLIC_KEY, + "npub": TEST_NPUB, + }) + ); +} diff --git a/crates/buzz-cli/tests/external_agent_fixtures.rs b/crates/buzz-cli/tests/external_agent_fixtures.rs new file mode 100644 index 0000000000..249b028a91 --- /dev/null +++ b/crates/buzz-cli/tests/external_agent_fixtures.rs @@ -0,0 +1,285 @@ +use std::collections::{HashMap, HashSet}; +use std::fs; +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +fn fixtures_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/external_agent_v1") +} + +#[derive(Debug, Deserialize)] +struct Contract { + contract: String, + agent_pubkey: String, + owner_pubkey: String, + allowlisted_pubkeys: Vec, + records: Vec, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +struct Fact { + fixture: String, + event_id: Option, + author_pubkey: Option, + channel_id: Option, + immediate_parent_id: Option, + thread_root_id: Option, + explicitly_mentions_agent: bool, + is_self_authored: bool, + author_policy_result: String, + activation_result: String, + conversation_lane: Option, +} + +fn null_fact(fixture: &str, author_policy: &str, activation: &str) -> Fact { + Fact { + fixture: fixture.to_string(), + event_id: None, + author_pubkey: None, + channel_id: None, + immediate_parent_id: None, + thread_root_id: None, + explicitly_mentions_agent: false, + is_self_authored: false, + author_policy_result: author_policy.to_string(), + activation_result: activation.to_string(), + conversation_lane: None, + } +} + +fn is_hex64(value: &str) -> bool { + value.len() == 64 && value.chars().all(|character| character.is_ascii_hexdigit()) +} + +fn tag_values<'a>(event: &'a Value, tag_name: &str) -> Vec<&'a str> { + event["tags"] + .as_array() + .into_iter() + .flatten() + .filter_map(Value::as_array) + .filter(|tag| tag.first().and_then(Value::as_str) == Some(tag_name)) + .filter_map(|tag| tag.get(1).and_then(Value::as_str)) + .collect() +} + +fn channel_fact(event: &Value) -> Result { + let channels: HashSet<&str> = tag_values(event, "h").into_iter().collect(); + match channels.len() { + 0 => Err("invalid_missing_channel"), + 1 => Ok(channels.into_iter().next().unwrap().to_string()), + _ => Err("invalid_conflicting_channel"), + } +} + +fn thread_facts(event: &Value) -> Result<(Option, Option), ()> { + let mut root = None; + let mut reply = None; + for tag in event["tags"] + .as_array() + .into_iter() + .flatten() + .filter_map(Value::as_array) + { + if tag.first().and_then(Value::as_str) != Some("e") { + continue; + } + let Some(marker) = tag.get(3).and_then(Value::as_str) else { + continue; + }; + if marker != "root" && marker != "reply" { + continue; + } + let Some(event_id) = tag.get(1).and_then(Value::as_str) else { + return Err(()); + }; + if !is_hex64(event_id) { + return Err(()); + } + let destination = if marker == "root" { + &mut root + } else { + &mut reply + }; + if destination + .as_ref() + .is_some_and(|existing| existing != event_id) + { + return Err(()); + } + *destination = Some(event_id.to_string()); + } + Ok((reply.clone(), root.or(reply))) +} + +fn normalize_event(fixture: &str, envelope: &Value, contract: &Contract) -> Fact { + let event = &envelope["event"]; + let event_id = event["id"].as_str().map(str::to_string); + let author = event["pubkey"].as_str().map(str::to_string); + let is_self_authored = author.as_deref() == Some(contract.agent_pubkey.as_str()); + let explicitly_mentions_agent = tag_values(event, "p") + .into_iter() + .any(|pubkey| pubkey == contract.agent_pubkey); + + if envelope["schema_version"].as_u64() != Some(1) { + return Fact { + fixture: fixture.to_string(), + event_id, + author_pubkey: author, + channel_id: channel_fact(event).ok(), + immediate_parent_id: None, + thread_root_id: None, + explicitly_mentions_agent, + is_self_authored, + author_policy_result: "unknown".to_string(), + activation_result: "unsupported_schema_version".to_string(), + conversation_lane: None, + }; + } + + let author_policy = if is_self_authored { + "self" + } else if author.as_deref() == Some(contract.owner_pubkey.as_str()) { + "owner" + } else if author + .as_ref() + .is_some_and(|pubkey| contract.allowlisted_pubkeys.contains(pubkey)) + { + "allowlisted" + } else { + "denied" + }; + + let channel = channel_fact(event); + let thread = thread_facts(event); + let activation = if !matches!(event["kind"].as_u64(), Some(9 | 40002)) { + "ignored_unsupported_kind" + } else if let Err(reason) = channel { + reason + } else if thread.is_err() { + "invalid_malformed_thread" + } else if is_self_authored { + "ignored_self_authored" + } else if author_policy == "denied" { + "ignored_non_owner" + } else if !explicitly_mentions_agent { + "ignored_missing_mention" + } else { + "accepted" + }; + let channel_id = channel.ok(); + let (immediate_parent_id, thread_root_id) = thread.unwrap_or_default(); + let conversation_lane = if activation == "accepted" { + channel_id + .as_ref() + .zip(thread_root_id.as_ref().or(event_id.as_ref())) + .map(|(channel, root)| format!("{channel}:{root}")) + } else { + None + }; + + Fact { + fixture: fixture.to_string(), + event_id, + author_pubkey: author, + channel_id, + immediate_parent_id, + thread_root_id, + explicitly_mentions_agent, + is_self_authored, + author_policy_result: author_policy.to_string(), + activation_result: activation.to_string(), + conversation_lane, + } +} + +fn normalize_fixture(fixture: &str, raw: &str, contract: &Contract) -> Vec { + if fixture == "malformed_json_line.ndjson" { + assert!( + raw.lines() + .all(|line| serde_json::from_str::(line).is_err()), + "malformed JSON fixture must fail parsing" + ); + return vec![null_fact(fixture, "not_applicable", "invalid_json")]; + } + + let envelopes: Vec = raw + .lines() + .map(|line| serde_json::from_str(line).expect("fixture line should parse")) + .collect(); + if fixture == "lifecycle_sequence.ndjson" { + let states: Vec<&str> = envelopes + .iter() + .map(|envelope| { + assert_eq!(envelope["schema_version"], 1); + assert_eq!(envelope["type"], "lifecycle"); + envelope["state"].as_str().expect("lifecycle state") + }) + .collect(); + assert_eq!(states, ["connected", "eose", "closed"]); + return vec![null_fact(fixture, "not_applicable", "lifecycle_only")]; + } + + let mut facts: Vec = envelopes + .iter() + .map(|envelope| { + assert_eq!(envelope["type"], "event"); + normalize_event(fixture, envelope, contract) + }) + .collect(); + if fixture == "duplicate_event_id.ndjson" { + assert_eq!( + envelopes.len(), + 2, + "duplicate fixture must contain two deliveries" + ); + assert_eq!( + envelopes[0], envelopes[1], + "a verified Nostr replay duplicate must be byte-identical" + ); + assert_eq!(facts[0], facts[1]); + facts.truncate(1); + facts[0].activation_result = "accepted_then_duplicate_suppressed".to_string(); + } + facts +} + +#[test] +fn external_agent_v1_fixtures_match_expected_normalized_facts() { + let dir = fixtures_dir(); + let expected_raw = fs::read_to_string(dir.join("expected_facts.json")).unwrap(); + let contract: Contract = serde_json::from_str(&expected_raw).unwrap(); + assert_eq!(contract.contract, "external_agent_v1"); + + let mut fixture_names = HashSet::new(); + for entry in fs::read_dir(&dir).unwrap() { + let path = entry.unwrap().path(); + if path.extension().and_then(|extension| extension.to_str()) == Some("ndjson") { + fixture_names.insert(path.file_name().unwrap().to_string_lossy().to_string()); + } + } + let expected_names: HashSet = contract + .records + .iter() + .map(|record| record.fixture.clone()) + .collect(); + assert_eq!( + fixture_names, expected_names, + "fixtures and expected_facts must reference the same file set" + ); + + let mut actual_by_fixture: HashMap> = HashMap::new(); + for fixture in &fixture_names { + let raw = fs::read_to_string(dir.join(fixture)).unwrap(); + actual_by_fixture.insert(fixture.clone(), normalize_fixture(fixture, &raw, &contract)); + } + let mut expected_by_fixture: HashMap> = HashMap::new(); + for record in &contract.records { + expected_by_fixture + .entry(record.fixture.clone()) + .or_default() + .push(record.clone()); + } + assert_eq!(actual_by_fixture, expected_by_fixture); +} diff --git a/crates/buzz-cli/tests/external_agent_listen.rs b/crates/buzz-cli/tests/external_agent_listen.rs new file mode 100644 index 0000000000..0078d6faed --- /dev/null +++ b/crates/buzz-cli/tests/external_agent_listen.rs @@ -0,0 +1,493 @@ +use std::net::TcpListener as StdTcpListener; +use std::process::Command; +#[cfg(unix)] +use std::process::Stdio; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::extract::State; +use axum::response::{Json, Response}; +use axum::routing::{get, post}; +use axum::Router; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use tokio::net::TcpListener; +use tokio::sync::oneshot; + +const TEST_PRIVATE_KEY: &str = "1111111111111111111111111111111111111111111111111111111111111111"; +const TEST_PUBLIC_KEY: &str = "4f355bdcb7cc0af728ef3cceb9615d90684bb5b2ca5f859ab0f0b704075871aa"; +const SENDER_PRIVATE_KEY: &str = "2222222222222222222222222222222222222222222222222222222222222222"; +const CHANNEL_A: &str = "11111111-1111-1111-1111-111111111111"; +const CHANNEL_B: &str = "22222222-2222-2222-2222-222222222222"; + +#[derive(Clone)] +struct FakeRelayState { + report_tx: Arc>>>>, + shutdown_tx: Arc>>>, +} + +#[cfg(unix)] +#[derive(Clone)] +struct SignalRelayState { + ready_tx: Arc>>>, + close_tx: Arc>>>, + shutdown_tx: Arc>>>, +} + +async fn fake_relay_upgrade(State(state): State, ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(move |socket| fake_relay_session(socket, state)) +} + +async fn fake_channel_query() -> Json { + Json(json!([ + { + "id": "3000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "kind": 39000, + "content": "", + "created_at": 1785100000, + "tags": [["d", CHANNEL_B]] + }, + { + "id": "3000000000000000000000000000000000000000000000000000000000000002", + "pubkey": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "kind": 39000, + "content": "", + "created_at": 1785100001, + "tags": [["d", CHANNEL_A]] + } + ])) +} + +#[cfg(unix)] +async fn signal_relay_upgrade( + State(state): State, + ws: WebSocketUpgrade, +) -> Response { + ws.on_upgrade(move |socket| signal_relay_session(socket, state)) +} + +async fn recv_json(socket: &mut WebSocket) -> Value { + match socket + .recv() + .await + .expect("client should send a WebSocket frame") + .expect("client WebSocket frame should be valid") + { + Message::Text(text) => serde_json::from_str(text.as_str()).expect("valid client JSON"), + other => panic!("expected text frame, got {other:?}"), + } +} + +fn signed_message(channel: &str, index: usize) -> Value { + let keys = Keys::parse(SENDER_PRIVATE_KEY).expect("valid sender key"); + let event = EventBuilder::new(Kind::Custom(40002), format!("fixture message {index}")) + .tags([ + Tag::parse(["h", channel]).expect("valid h tag"), + Tag::parse(["p", TEST_PUBLIC_KEY]).expect("valid p tag"), + ]) + .sign_with_keys(&keys) + .expect("signed fixture event"); + serde_json::to_value(event).expect("event JSON") +} + +async fn fake_relay_session(mut socket: WebSocket, state: FakeRelayState) { + socket + .send(Message::Text( + json!(["AUTH", "external-agent-test-challenge"]) + .to_string() + .into(), + )) + .await + .expect("send AUTH challenge"); + + let auth = recv_json(&mut socket).await; + assert_eq!(auth[0], "AUTH"); + let auth_event_id = auth[1]["id"].as_str().expect("AUTH event id"); + socket + .send(Message::Text( + json!(["OK", auth_event_id, true, ""]).to_string().into(), + )) + .await + .expect("accept AUTH event"); + + let mut requests = Vec::new(); + for index in 0..2 { + let request = recv_json(&mut socket).await; + let sub_id = request[1].as_str().expect("subscription id").to_string(); + let channel = request[2]["#h"][0] + .as_str() + .expect("single channel filter") + .to_string(); + requests.push(request); + + socket + .send(Message::Text( + json!(["EVENT", sub_id, signed_message(&channel, index)]) + .to_string() + .into(), + )) + .await + .expect("send event"); + socket + .send(Message::Text(json!(["EOSE", sub_id]).to_string().into())) + .await + .expect("send EOSE"); + } + + state + .report_tx + .lock() + .expect("report lock") + .take() + .expect("report sender") + .send(requests) + .ok(); + let _ = socket.send(Message::Close(None)).await; + state + .shutdown_tx + .lock() + .expect("shutdown lock") + .take() + .expect("shutdown sender") + .send(()) + .ok(); +} + +#[cfg(unix)] +async fn signal_relay_session(mut socket: WebSocket, state: SignalRelayState) { + socket + .send(Message::Text( + json!(["AUTH", "external-agent-signal-test"]) + .to_string() + .into(), + )) + .await + .expect("send AUTH challenge"); + let auth = recv_json(&mut socket).await; + let auth_event_id = auth[1]["id"].as_str().expect("AUTH event id"); + socket + .send(Message::Text( + json!(["OK", auth_event_id, true, ""]).to_string().into(), + )) + .await + .expect("accept AUTH event"); + + let request = recv_json(&mut socket).await; + let sub_id = request[1].as_str().expect("subscription id").to_string(); + socket + .send(Message::Text(json!(["EOSE", sub_id]).to_string().into())) + .await + .expect("send EOSE"); + state + .ready_tx + .lock() + .expect("ready lock") + .take() + .expect("ready sender") + .send(()) + .ok(); + + let mut received_close = false; + while let Some(frame) = socket.recv().await { + match frame.expect("valid client frame") { + Message::Text(text) => { + let message: Value = + serde_json::from_str(text.as_str()).expect("valid client JSON"); + if message[0] == "CLOSE" && message[1] == sub_id { + received_close = true; + break; + } + } + Message::Close(_) => break, + _ => {} + } + } + + state + .close_tx + .lock() + .expect("close report lock") + .take() + .expect("close report sender") + .send(received_close) + .ok(); + state + .shutdown_tx + .lock() + .expect("shutdown lock") + .take() + .expect("shutdown sender") + .send(()) + .ok(); +} + +fn listen_command(relay_url: &str, channels: &[&str]) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_buzz")); + command.arg("listen"); + for channel in channels { + command.args(["--channel", channel]); + } + command + .args(["--mentions-of-me", "--envelope", "v1", "--no-reconnect"]) + .env("BUZZ_PRIVATE_KEY", TEST_PRIVATE_KEY) + .env("BUZZ_RELAY_URL", relay_url) + .env_remove("BUZZ_AUTH_TAG"); + command +} + +async fn start_fake_relay() -> ( + std::net::SocketAddr, + oneshot::Receiver>, + tokio::task::JoinHandle<()>, +) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fake relay"); + let address = listener.local_addr().expect("fake relay address"); + let (report_tx, report_rx) = oneshot::channel(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let state = FakeRelayState { + report_tx: Arc::new(Mutex::new(Some(report_tx))), + shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), + }; + let app = Router::new() + .route("/", get(fake_relay_upgrade)) + .route("/query", post(fake_channel_query)) + .with_state(state); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + }) + .await + .expect("fake relay server"); + }); + (address, report_rx, server) +} + +#[cfg(unix)] +async fn start_signal_relay() -> ( + std::net::SocketAddr, + oneshot::Receiver<()>, + oneshot::Receiver, + tokio::task::JoinHandle<()>, +) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind signal relay"); + let address = listener.local_addr().expect("signal relay address"); + let (ready_tx, ready_rx) = oneshot::channel(); + let (close_tx, close_rx) = oneshot::channel(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let state = SignalRelayState { + ready_tx: Arc::new(Mutex::new(Some(ready_tx))), + close_tx: Arc::new(Mutex::new(Some(close_tx))), + shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), + }; + let app = Router::new() + .route("/", get(signal_relay_upgrade)) + .with_state(state); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + }) + .await + .expect("signal relay server"); + }); + (address, ready_rx, close_rx, server) +} + +fn assert_scoped_requests(requests: &[Value]) { + assert_eq!(requests.len(), 2); + let subscription_ids: Vec<&str> = requests + .iter() + .map(|request| request[1].as_str().expect("subscription id")) + .collect(); + assert_ne!(subscription_ids[0], subscription_ids[1]); + let mut channels: Vec<&str> = requests + .iter() + .map(|request| { + assert_eq!(request.as_array().expect("REQ array").len(), 3); + assert_eq!(request[0], "REQ"); + assert_eq!(request[2]["#h"].as_array().expect("#h array").len(), 1); + assert_eq!(request[2]["#p"], json!([TEST_PUBLIC_KEY])); + request[2]["#h"][0].as_str().expect("channel id") + }) + .collect(); + channels.sort_unstable(); + assert_eq!(channels, vec![CHANNEL_A, CHANNEL_B]); +} + +fn assert_event_stream(output: &std::process::Output) { + assert_eq!(output.status.code(), Some(2)); + let stdout = String::from_utf8(output.stdout.clone()).expect("stdout UTF-8"); + let records: Vec = stdout + .lines() + .map(|line| serde_json::from_str(line).expect("stdout NDJSON")) + .collect(); + assert_eq!( + records + .iter() + .filter(|record| record["type"] == "event") + .count(), + 2 + ); + assert_eq!( + records + .iter() + .filter(|record| record["state"] == "eose") + .count(), + 1 + ); + assert_eq!( + records.first().expect("connected record")["state"], + "connected" + ); + assert_eq!(records.last().expect("fatal record")["state"], "fatal"); +} + +#[cfg(unix)] +async fn assert_graceful_signal(signal: &str) { + let (address, ready_rx, close_rx, server) = start_signal_relay().await; + let mut command = listen_command(&format!("http://{address}"), &[CHANNEL_A]); + command.stdout(Stdio::piped()).stderr(Stdio::piped()); + let child = command.spawn().expect("spawn buzz listen"); + let pid = child.id().to_string(); + + tokio::time::timeout(Duration::from_secs(10), ready_rx) + .await + .expect("listener should reach EOSE") + .expect("ready report"); + let signal_status = Command::new("kill") + .args([signal, &pid]) + .status() + .expect("send process signal"); + assert!(signal_status.success()); + + let output = tokio::time::timeout( + Duration::from_secs(10), + tokio::task::spawn_blocking(move || { + child.wait_with_output().expect("listen process output") + }), + ) + .await + .expect("listen should stop after signal") + .expect("listen wait task"); + assert!( + close_rx.await.expect("CLOSE report"), + "listener should send Nostr CLOSE before disconnect" + ); + server.await.expect("signal relay task"); + + assert!(output.status.success()); + assert!( + output.stderr.is_empty(), + "graceful signal should not emit stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let records: Vec = String::from_utf8(output.stdout) + .expect("stdout UTF-8") + .lines() + .map(|line| serde_json::from_str(line).expect("stdout NDJSON")) + .collect(); + assert_eq!( + records + .iter() + .filter(|record| record["state"] == "connected") + .count(), + 1 + ); + assert_eq!( + records + .iter() + .filter(|record| record["state"] == "eose") + .count(), + 1 + ); + assert!( + records.iter().all(|record| record["state"] != "fatal"), + "graceful signal must not emit fatal" + ); +} + +#[test] +fn listen_network_failure_is_machine_classifiable() { + let listener = StdTcpListener::bind("127.0.0.1:0").expect("bind unused port"); + let address = listener.local_addr().expect("unused port address"); + drop(listener); + + let output = listen_command(&format!("http://{address}"), &[CHANNEL_A, CHANNEL_B]) + .output() + .expect("buzz listen should start"); + + assert_eq!(output.status.code(), Some(2)); + let stdout = String::from_utf8(output.stdout).expect("stdout UTF-8"); + let records: Vec = stdout + .lines() + .map(|line| serde_json::from_str(line).expect("stdout NDJSON")) + .collect(); + assert_eq!(records.len(), 1); + assert_eq!(records[0]["type"], "lifecycle"); + assert_eq!(records[0]["state"], "fatal"); + + let stderr = String::from_utf8(output.stderr).expect("stderr UTF-8"); + let error: Value = serde_json::from_str(stderr.trim()).expect("stderr error JSON"); + assert_eq!(error["error"], "network_error"); + assert_eq!(error["retryable"], true); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn multi_channel_listen_uses_scoped_subscriptions_and_one_eose() { + let (address, report_rx, server) = start_fake_relay().await; + + let relay_url = format!("http://{address}"); + let output = tokio::time::timeout( + Duration::from_secs(10), + tokio::task::spawn_blocking(move || { + listen_command(&relay_url, &[CHANNEL_A, CHANNEL_B]) + .output() + .expect("buzz listen should run") + }), + ) + .await + .expect("buzz listen should exit") + .expect("listen process task"); + let requests = report_rx.await.expect("fake relay request report"); + server.await.expect("fake relay task"); + + assert_scoped_requests(&requests); + assert_event_stream(&output); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mentions_only_discovers_visible_channels_before_listening() { + let (address, report_rx, server) = start_fake_relay().await; + + let relay_url = format!("http://{address}"); + let output = tokio::time::timeout( + Duration::from_secs(10), + tokio::task::spawn_blocking(move || { + listen_command(&relay_url, &[]) + .output() + .expect("buzz listen should run") + }), + ) + .await + .expect("buzz listen should exit") + .expect("listen process task"); + let requests = report_rx.await.expect("fake relay request report"); + server.await.expect("fake relay task"); + + assert_scoped_requests(&requests); + assert_event_stream(&output); +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sigint_and_sigterm_close_subscriptions_cleanly() { + assert_graceful_signal("-INT").await; + assert_graceful_signal("-TERM").await; +} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/README.md b/crates/buzz-cli/tests/fixtures/external_agent_v1/README.md new file mode 100644 index 0000000000..0f3908799f --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/README.md @@ -0,0 +1,23 @@ +# External Agent v1 Fixtures + +These fixtures define the Buzz-owned external-agent ingress contract used by +resident adapters. Each `*.ndjson` file is consumed line by line. Most files +contain versioned `buzz listen --envelope v1` records; malformed and future +schema cases intentionally exercise fail-closed parser behavior. + +Adapters should normalize each event into the facts recorded in +`expected_facts.json` and apply policy locally. The fixture data is not a shared +runtime package and does not carry private keys. + +The Rust contract test contains the reference normalization order. Schema, +channel, and thread validation fail closed before activation; self-authored and +unauthorized events are ignored before a conversation lane is created. Replay +duplicates are byte-identical and use the event ID as their idempotency key. + +Common identities: + +- `agent_pubkey`: `aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa` +- `owner_pubkey`: `bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb` +- `allowlisted_pubkey`: `cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc` +- `non_owner_pubkey`: `dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd` +- `channel_id`: `11111111-1111-1111-1111-111111111111` diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/allowlisted_sender.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/allowlisted_sender.ndjson new file mode 100644 index 0000000000..e71dad37cc --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/allowlisted_sender.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000005","pubkey":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","kind":40002,"content":"@agent run the allowed diagnostic","created_at":1785100005,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/conflicting_h_tags.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/conflicting_h_tags.ndjson new file mode 100644 index 0000000000..56ef3098c5 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/conflicting_h_tags.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000009","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent ambiguous channel","created_at":1785100009,"tags":[["h","11111111-1111-1111-1111-111111111111"],["h","22222222-2222-2222-2222-222222222222"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/duplicate_event_id.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/duplicate_event_id.ndjson new file mode 100644 index 0000000000..3dcfc2692f --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/duplicate_event_id.ndjson @@ -0,0 +1,2 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000011","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent duplicate one","created_at":1785100011,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000011","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent duplicate one","created_at":1785100011,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/expected_facts.json b/crates/buzz-cli/tests/fixtures/external_agent_v1/expected_facts.json new file mode 100644 index 0000000000..f6a395c9bc --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/expected_facts.json @@ -0,0 +1,26 @@ +{ + "contract": "external_agent_v1", + "agent_pubkey": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "owner_pubkey": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "allowlisted_pubkeys": [ + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + ], + "records": [ + {"fixture":"owner_top_level_mention.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000001","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:1000000000000000000000000000000000000000000000000000000000000001"}, + {"fixture":"owner_direct_reply.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000002","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":"2000000000000000000000000000000000000000000000000000000000000001","thread_root_id":"2000000000000000000000000000000000000000000000000000000000000001","explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:2000000000000000000000000000000000000000000000000000000000000001"}, + {"fixture":"owner_nested_thread_mention.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000003","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":"2000000000000000000000000000000000000000000000000000000000000002","thread_root_id":"2000000000000000000000000000000000000000000000000000000000000001","explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:2000000000000000000000000000000000000000000000000000000000000001"}, + {"fixture":"non_owner_mention.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000004","author_pubkey":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"denied","activation_result":"ignored_non_owner","conversation_lane":null}, + {"fixture":"allowlisted_sender.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000005","author_pubkey":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"allowlisted","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:1000000000000000000000000000000000000000000000000000000000000005"}, + {"fixture":"self_authored.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000006","author_pubkey":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":false,"is_self_authored":true,"author_policy_result":"self","activation_result":"ignored_self_authored","conversation_lane":null}, + {"fixture":"unsupported_kind.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000007","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"ignored_unsupported_kind","conversation_lane":null}, + {"fixture":"missing_h_tag.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000008","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":null,"immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"invalid_missing_channel","conversation_lane":null}, + {"fixture":"conflicting_h_tags.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000009","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":null,"immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"invalid_conflicting_channel","conversation_lane":null}, + {"fixture":"malformed_e_marker.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000010","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"invalid_malformed_thread","conversation_lane":null}, + {"fixture":"duplicate_event_id.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000011","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted_then_duplicate_suppressed","conversation_lane":"11111111-1111-1111-1111-111111111111:1000000000000000000000000000000000000000000000000000000000000011"}, + {"fixture":"same_second_distinct_events.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000012","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:1000000000000000000000000000000000000000000000000000000000000012"}, + {"fixture":"same_second_distinct_events.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000013","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"owner","activation_result":"accepted","conversation_lane":"11111111-1111-1111-1111-111111111111:1000000000000000000000000000000000000000000000000000000000000013"}, + {"fixture":"lifecycle_sequence.ndjson","event_id":null,"author_pubkey":null,"channel_id":null,"immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":false,"is_self_authored":false,"author_policy_result":"not_applicable","activation_result":"lifecycle_only","conversation_lane":null}, + {"fixture":"malformed_json_line.ndjson","event_id":null,"author_pubkey":null,"channel_id":null,"immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":false,"is_self_authored":false,"author_policy_result":"not_applicable","activation_result":"invalid_json","conversation_lane":null}, + {"fixture":"future_schema_version.ndjson","event_id":"1000000000000000000000000000000000000000000000000000000000000015","author_pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","channel_id":"11111111-1111-1111-1111-111111111111","immediate_parent_id":null,"thread_root_id":null,"explicitly_mentions_agent":true,"is_self_authored":false,"author_policy_result":"unknown","activation_result":"unsupported_schema_version","conversation_lane":null} + ] +} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/future_schema_version.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/future_schema_version.ndjson new file mode 100644 index 0000000000..c4ffd50be1 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/future_schema_version.ndjson @@ -0,0 +1 @@ +{"schema_version":99,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000015","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent future schema","created_at":1785100015,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/lifecycle_sequence.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/lifecycle_sequence.ndjson new file mode 100644 index 0000000000..e86b886f24 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/lifecycle_sequence.ndjson @@ -0,0 +1,3 @@ +{"schema_version":1,"type":"lifecycle","state":"connected"} +{"schema_version":1,"type":"lifecycle","state":"eose"} +{"schema_version":1,"type":"lifecycle","state":"closed","message":"subscription closed: relay shutdown"} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_e_marker.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_e_marker.ndjson new file mode 100644 index 0000000000..e36543b0dd --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_e_marker.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000010","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent malformed thread marker","created_at":1785100010,"tags":[["h","11111111-1111-1111-1111-111111111111"],["e","not-a-valid-event-id","","root"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_json_line.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_json_line.ndjson new file mode 100644 index 0000000000..7142fa8548 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/malformed_json_line.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000014" diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/missing_h_tag.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/missing_h_tag.ndjson new file mode 100644 index 0000000000..1d00937d34 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/missing_h_tag.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000008","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent missing channel","created_at":1785100008,"tags":[["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/non_owner_mention.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/non_owner_mention.ndjson new file mode 100644 index 0000000000..e0d73ec843 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/non_owner_mention.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000004","pubkey":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd","kind":40002,"content":"@agent are you there?","created_at":1785100004,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_direct_reply.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_direct_reply.ndjson new file mode 100644 index 0000000000..2ab1e5fc9c --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_direct_reply.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000002","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"Can you expand on that?","created_at":1785100002,"tags":[["h","11111111-1111-1111-1111-111111111111"],["e","2000000000000000000000000000000000000000000000000000000000000001","","reply"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_nested_thread_mention.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_nested_thread_mention.ndjson new file mode 100644 index 0000000000..bcb7cb44af --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_nested_thread_mention.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000003","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent check this nested detail","created_at":1785100003,"tags":[["h","11111111-1111-1111-1111-111111111111"],["e","2000000000000000000000000000000000000000000000000000000000000001","","root"],["e","2000000000000000000000000000000000000000000000000000000000000002","","reply"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_top_level_mention.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_top_level_mention.ndjson new file mode 100644 index 0000000000..810a6cb3e8 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/owner_top_level_mention.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000001","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent please summarize the build failure","created_at":1785100001,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/same_second_distinct_events.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/same_second_distinct_events.ndjson new file mode 100644 index 0000000000..783a0eb349 --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/same_second_distinct_events.ndjson @@ -0,0 +1,2 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000012","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent same second first","created_at":1785100012,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000013","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":40002,"content":"@agent same second second","created_at":1785100012,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/self_authored.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/self_authored.ndjson new file mode 100644 index 0000000000..c2cf7b2ece --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/self_authored.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000006","pubkey":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","kind":40002,"content":"Agent reply echo from relay","created_at":1785100006,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"]]}} diff --git a/crates/buzz-cli/tests/fixtures/external_agent_v1/unsupported_kind.ndjson b/crates/buzz-cli/tests/fixtures/external_agent_v1/unsupported_kind.ndjson new file mode 100644 index 0000000000..ff1ed2d78e --- /dev/null +++ b/crates/buzz-cli/tests/fixtures/external_agent_v1/unsupported_kind.ndjson @@ -0,0 +1 @@ +{"schema_version":1,"type":"event","event":{"id":"1000000000000000000000000000000000000000000000000000000000000007","pubkey":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","kind":7,"content":"+","created_at":1785100007,"tags":[["h","11111111-1111-1111-1111-111111111111"],["p","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]]}} diff --git a/crates/buzz-cli/tests/keys_generate.rs b/crates/buzz-cli/tests/keys_generate.rs new file mode 100644 index 0000000000..7141ff2da1 --- /dev/null +++ b/crates/buzz-cli/tests/keys_generate.rs @@ -0,0 +1,168 @@ +//! Integration tests for `buzz keys generate`. +//! +//! These spawn the compiled binary as a subprocess because the two contracts +//! that matter most cannot be observed from a unit test: +//! +//! 1. **Local-only.** The command must run with no `BUZZ_PRIVATE_KEY`, no +//! `BUZZ_RELAY_URL`, and no network. Unit tests call `cmd_generate` directly +//! and so bypass `run()`, which is exactly where the "private key is +//! required" gate lives — the gate this command has to be dispatched before. +//! 2. **The secret stays off stdout.** Only a real invocation proves what the +//! process actually wrote to its stdout stream. + +use std::process::Command; + +/// Spawn `buzz` with a scrubbed environment: no identity, no relay, no auth +/// tag. Any of those leaking in from the developer's shell would mask a +/// regression where the command started depending on them. +fn run_keys(args: &[&str]) -> std::process::Output { + let bin = env!("CARGO_BIN_EXE_buzz"); + Command::new(bin) + .args(args) + .current_dir(std::env::temp_dir()) + .env_remove("BUZZ_PRIVATE_KEY") + .env_remove("BUZZ_RELAY_URL") + .env_remove("BUZZ_AUTH_TAG") + .output() + .expect("failed to spawn buzz") +} + +fn parse_stdout(out: &std::process::Output) -> serde_json::Value { + let stdout = String::from_utf8_lossy(&out.stdout); + serde_json::from_str(stdout.trim()) + .unwrap_or_else(|e| panic!("stdout was not JSON ({e}): {stdout}")) +} + +#[test] +fn generates_without_a_private_key_or_relay() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + + let out = run_keys(&["keys", "generate", "--out", path.to_str().unwrap()]); + assert!( + out.status.success(), + "expected success, got {:?}; stderr: {}", + out.status.code(), + String::from_utf8_lossy(&out.stderr) + ); + + let report = parse_stdout(&out); + let pubkey = report["pubkey"].as_str().expect("pubkey in report"); + assert_eq!(pubkey.len(), 64, "pubkey should be 64-char hex: {pubkey}"); + assert!(pubkey.chars().all(|c| c.is_ascii_hexdigit())); + assert!(report["npub"] + .as_str() + .expect("npub in report") + .starts_with("npub1")); + + // The written secret is a real nsec, and it is the one the reported pubkey + // belongs to — the round trip a caller depends on. + let nsec = std::fs::read_to_string(&path).unwrap(); + let nsec = nsec.trim(); + assert!(nsec.starts_with("nsec1"), "expected an nsec, got: {nsec}"); + let reloaded = nostr::Keys::parse(nsec).expect("generated nsec must parse"); + assert_eq!(reloaded.public_key().to_hex(), pubkey); +} + +#[test] +fn does_not_print_the_secret_by_default() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + + let out = run_keys(&["keys", "generate", "--out", path.to_str().unwrap()]); + assert!(out.status.success()); + + let nsec = std::fs::read_to_string(&path).unwrap(); + let nsec = nsec.trim(); + + // Neither the literal secret nor the bech32 prefix may appear on either + // stream. Checking both streams matters: an accidental `eprintln!` of the + // key is just as much of a leak as a `println!`. + let stdout = String::from_utf8_lossy(&out.stdout); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(!stdout.contains(nsec), "secret key leaked to stdout"); + assert!(!stderr.contains(nsec), "secret key leaked to stderr"); + assert!(!stdout.contains("nsec1"), "nsec prefix appeared on stdout"); + assert!(!stderr.contains("nsec1"), "nsec prefix appeared on stderr"); + + let report = parse_stdout(&out); + assert!( + report.get("nsec").is_none(), + "report must not carry the secret without --stdout" + ); +} + +#[test] +fn stdout_flag_is_an_explicit_opt_in() { + let out = run_keys(&["keys", "generate", "--stdout"]); + assert!(out.status.success()); + + let report = parse_stdout(&out); + let nsec = report["nsec"] + .as_str() + .expect("nsec in report with --stdout"); + assert!(nsec.starts_with("nsec1")); + + // With no --out there is no file, so no path is reported. + assert!(report.get("secret_key_path").is_none()); + + let reloaded = nostr::Keys::parse(nsec).expect("piped nsec must parse"); + assert_eq!( + reloaded.public_key().to_hex(), + report["pubkey"].as_str().unwrap() + ); +} + +#[test] +fn refuses_a_destinationless_invocation() { + let out = run_keys(&["keys", "generate"]); + assert_eq!( + out.status.code(), + Some(1), + "expected input-error exit code 1; stderr: {}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(out.stdout.is_empty(), "no key should be reported"); +} + +#[test] +fn refuses_to_clobber_an_existing_identity() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + let arg = path.to_str().unwrap(); + + let first = run_keys(&["keys", "generate", "--out", arg]); + assert!(first.status.success()); + let original = std::fs::read_to_string(&path).unwrap(); + + let second = run_keys(&["keys", "generate", "--out", arg]); + assert_eq!(second.status.code(), Some(1), "expected refusal"); + + // The live identity is intact — this is the property that protects an + // already-connected agent from a re-run of the connect flow. + assert_eq!(std::fs::read_to_string(&path).unwrap(), original); + + let forced = run_keys(&["keys", "generate", "--out", arg, "--force"]); + assert!(forced.status.success(), "--force should replace the file"); + assert_ne!(std::fs::read_to_string(&path).unwrap(), original); +} + +#[cfg(unix)] +#[test] +fn written_secret_is_owner_only() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.nsec"); + + let out = run_keys(&["keys", "generate", "--out", path.to_str().unwrap()]); + assert!(out.status.success()); + + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!( + mode & 0o777, + 0o600, + "secret key file must be owner read/write only, got {:o}", + mode & 0o777 + ); +} diff --git a/docs/cli-external-agents.md b/docs/cli-external-agents.md new file mode 100644 index 0000000000..2f538e3b48 --- /dev/null +++ b/docs/cli-external-agents.md @@ -0,0 +1,118 @@ +# Buzz CLI External Agent Contract + +This document records the stable CLI prerequisites used by resident external +agent adapters. These commands are harness-neutral: they do not assume any +specific runtime, vendor, or deployment layout. + +## Identity + +`buzz users me` prints the identity derived from the configured +`BUZZ_PRIVATE_KEY` or `--private-key` value: + +```json +{"pubkey":"<64-char hex pubkey>","npub":"npub1..."} +``` + +The command performs no relay request and never prints private key material. +Adapters use it during startup to prove that local key custody matches the +configured resident agent identity before they subscribe or send. + +## Compact Message Reads + +`buzz --format compact messages get`, `messages thread`, and `messages search` +return sig-stripped message objects with the fields adapters need for policy, +activation, self-suppression, and threading: + +```json +[ + { + "id": "", + "pubkey": "", + "kind": 40002, + "content": "hello", + "created_at": 1785100000, + "tags": [["h", ""], ["p", ""]] + } +] +``` + +Existing compact consumers can continue reading `id`, `content`, and +`created_at`; the new fields are additive. + +Human-readable errors remain on stderr as JSON through the existing CLI error +contract. Successful read commands print JSON on stdout only. + +## Realtime Listen + +`buzz listen` streams matching relay events as newline-delimited JSON. The +resident adapter owns durable cursor advancement and may disable CLI reconnects: + +```bash +buzz listen \ + --channel "$CHANNEL_UUID" \ + --mentions-of-me \ + --since "$REPLAY_SINCE" \ + --envelope v1 \ + --no-reconnect +``` + +Filter semantics are conjunctive inside one relay filter: + +- `--channel` only: events in any configured channel; +- `--mentions-of-me` only: discover currently visible channels over the + authenticated query bridge, then receive events that p-tag this CLI identity; +- both: events that match one configured channel and p-tag this CLI identity. + +### Direct messages (`--dms`) + +`--dms` additionally subscribes to every direct-message conversation this +identity participates in. Buzz assigns each DM conversation a stable channel +UUID, and DM events carry it in their `h` tag exactly like channel traffic, so +an adapter can map that UUID to a native `direct` peer without any second DM +architecture. Conversations are discovered from the relay-emitted DM-created +metadata (`kind:41001`, p-tagged with every participant) — the same query +`buzz dms list` uses. + +Newly opened conversations are picked up without a restart: the listener +re-checks every 30 seconds and adds a channel-scoped subscription for each new +conversation. Under `--envelope v1` each addition emits a lifecycle record: + +```json +{"schema_version":1,"type":"lifecycle","state":"dm_channel_added","message":""} +``` + +`--dms` alone is a complete subscription request: with no DM conversations yet +the session connects, emits `eose` immediately, and subscribes conversations as +they are opened. A failed discovery re-check is reported on stderr as +`{"error":"dm_poll_failed"}` and retried at the next interval; the live +subscriptions are unaffected. + +Each configured channel uses an independent channel-scoped relay subscription. +This preserves live delivery across multiple channels without weakening the +relay's channel/global fan-out boundary. Duplicate channel arguments are +deduplicated. Mention-only discovery is repeated when the adapter restarts the +process; restart after membership changes to refresh that channel set. + +The v1 envelope prints event records as: + +```json +{"schema_version":1,"type":"event","event":{"id":"","pubkey":"","kind":40002,"content":"hello","created_at":1785100000,"tags":[["h",""],["p",""]]}} +``` + +Lifecycle records use the same stdout stream: + +```json +{"schema_version":1,"type":"lifecycle","state":"connected"} +{"schema_version":1,"type":"lifecycle","state":"eose"} +``` + +Allowed v1 lifecycle states are `connected`, `eose`, `dm_channel_added`, +`closed`, and `fatal`. +For multi-channel listeners, one `eose` record is emitted after every +channel-scoped subscription reaches EOSE. Diagnostics and reconnect notices are +emitted on stderr as JSON. + +With automatic reconnect enabled, the CLI reuses the original `--since` value. +Consumers must deduplicate replayed events by event ID. Resident adapters should +use `--no-reconnect`, persist their durable cursor, and start a new process with +the documented overlap.