diff --git a/Cargo.lock b/Cargo.lock index 9544a63b89..252eb9b1be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -831,6 +831,7 @@ version = "0.1.0" dependencies = [ "anyhow", "base64 0.22.1", + "buzz-broker-client", "buzz-core", "buzz-persona", "buzz-sdk", @@ -973,12 +974,24 @@ dependencies = [ "tower", ] +[[package]] +name = "buzz-broker-client" +version = "0.1.0" +dependencies = [ + "axum", + "buzz-sdk", + "reqwest 0.13.4", + "serde_json", + "tokio", +] + [[package]] name = "buzz-cli" version = "0.1.0" dependencies = [ "axum", "base64 0.22.1", + "buzz-broker-client", "buzz-core", "buzz-persona", "buzz-sdk", diff --git a/Cargo.toml b/Cargo.toml index d6ee839f1b..777e32953f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ members = [ "crates/buzz-workflow", "crates/buzz-media", "crates/buzz-cli", + "crates/buzz-broker-client", "crates/buzz-pairing-cli", "crates/buzz-sdk", "crates/buzz-persona", @@ -147,6 +148,7 @@ buzz-audit = { path = "crates/buzz-audit" } buzz-workflow = { path = "crates/buzz-workflow" } buzz-media = { path = "crates/buzz-media" } buzz-sdk = { path = "crates/buzz-sdk" } +buzz-broker-client = { path = "crates/buzz-broker-client" } buzz-ws-client = { path = "crates/buzz-ws-client" } buzz-relay-mesh = { path = "crates/buzz-relay-mesh" } buzz-datastore-tracing = { path = "crates/buzz-datastore-tracing" } diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806..8a3dd9748c 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -19,6 +19,7 @@ path = "src/main.rs" # Internal buzz-core = { workspace = true } buzz-sdk = { workspace = true } +buzz-broker-client = { workspace = true } buzz-persona = { path = "../buzz-persona" } # Nostr diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index e6164b02dd..88f866732b 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -1,9 +1,13 @@ # buzz-acp -ACP harness that connects AI agents to Buzz. The harness listens for @mentions on the relay, prompts your agent, and the agent replies using the Buzz CLI. +ACP harness that connects AI agents to Buzz. In local mode the harness listens +for @mentions on the relay. In keyless broker mode it polls a host that owns the +agent key and relay route. Both modes prompt the same ACP agent, which replies +using the Buzz CLI. ``` -Buzz Relay ──WS──→ buzz-acp ──stdio──→ Your Agent +Buzz Relay ──WS──→ buzz-acp ──stdio──→ Your Agent (local) +Broker Host ─HTTP→ buzz-acp ──stdio──→ Your Agent (keyless) │ Buzz CLI (send_message, etc.) @@ -64,6 +68,34 @@ buzz-acp That's it. The harness spawns `goose acp`, connects to the relay, discovers channels, and starts listening. When someone @mentions the agent, goose receives the message and can reply using the Buzz CLI that the harness configures automatically. +## Keyless broker mode (prototype) + +Broker mode starts the runtime without an agent private key or direct relay +connection. The harness derives the agent public key through `storage.address`, +polls each configured channel through `channel.read`, and provisions the spawned +agent and optional MCP server with broker variables so their `buzz` commands use +the same host. + +```bash +export BUZZ_AGENT_MODE=broker +export BUZZ_BROKER_URL=http://127.0.0.1:8787 +export BUZZ_BROKER_CREDENTIAL=dev-token +export BUZZ_ACP_CHANNELS=5df7dfa8-e919-43df-8efd-f1dcb8af7071 +export BUZZ_ACP_AGENT_OWNER=a02c4e0850e5e612b4ddf95dbe2f5c56467cf27c6552203bc833ff438fb31971 +unset BUZZ_PRIVATE_KEY BUZZ_RELAY_URL BUZZ_AUTH_TAG + +buzz-acp +``` + +This slice is intentionally receive-and-reply focused. Because the frozen +broker contract does not expose channel discovery, channel metadata, profiles, +or runtime housekeeping operations, broker mode currently requires explicit +channel UUIDs and `respond-to=owner-only`. Presence, typing, reactions used as +turn status, observer/liveness events, relay conversation enrichment, setup +nudges, and core-memory injection are disabled. The `buzz` CLI operations +available to the spawned agent are documented in +[`../buzz-cli/KEYLESS.md`](../buzz-cli/KEYLESS.md). + ## Running with Codex [codex-acp](https://github.com/agentclientprotocol/codex-acp) wraps OpenAI Codex in an ACP interface. @@ -106,8 +138,14 @@ All configuration is via environment variables (or CLI flags — every env var h | Variable | Required | Default | Description | |----------|----------|---------|-------------| -| `BUZZ_PRIVATE_KEY` | **yes** | — | Agent's Nostr private key (`nsec1...`). Used for relay auth and agent identity. | +| `BUZZ_AGENT_MODE` | no | `local` | Runtime substrate: `local` or keyless `broker`. | +| `BUZZ_PRIVATE_KEY` | local only | — | Agent's Nostr private key (`nsec1...`). Rejected in broker mode. | | `BUZZ_RELAY_URL` | no | `ws://localhost:3000` | Relay WebSocket URL. | +| `BUZZ_BROKER_URL` | broker only | — | Broker base URL; actions are posted to `/v1/action`. | +| `BUZZ_BROKER_CREDENTIAL` | broker only | — | Bearer credential issued by the broker host. | +| `BUZZ_BROKER_POLL_INTERVAL_MS` | no | `1000` | Broker `channel.read` polling interval; minimum `100`. | +| `BUZZ_ACP_CHANNELS` | broker only | — | Comma-separated channel UUIDs to poll. | +| `BUZZ_ACP_AGENT_OWNER` | broker only | — | Owner pubkey accepted by the broker-mode author gate. | | `BUZZ_ACP_AGENT_COMMAND` | no | `goose` | Agent binary to spawn. | | `BUZZ_ACP_AGENT_ARGS` | no | `acp` | Agent arguments (comma-separated). | | `BUZZ_ACP_MCP_COMMAND` | no | `""` (empty) | Path to an optional MCP server binary to provide to the agent subprocess. | diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 0d87bca028..792221092b 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -508,7 +508,19 @@ impl AcpClient { // Handled by build_codex_config_env; skip here to avoid double-setting. continue; } - if std::env::var_os(key).is_none() { + // Substrate provisioning is a security boundary, not a persona + // preference. Broker mode must override inherited local relay/key + // values, including with empty tombstones. + let force_runtime_provisioning = matches!( + key.as_str(), + "BUZZ_AGENT_MODE" + | "BUZZ_BROKER_URL" + | "BUZZ_BROKER_CREDENTIAL" + | "BUZZ_RELAY_URL" + | "BUZZ_PRIVATE_KEY" + | "BUZZ_AUTH_TAG" + ); + if force_runtime_provisioning || std::env::var_os(key).is_none() { cmd.env(key, value); } } diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 2d7b212832..8aa052ed07 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -54,6 +54,33 @@ pub enum SubscribeMode { Config, } +/// Runtime substrate for inbound work and agent CLI operations. +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum AgentMode { + /// Hold the agent key and connect directly to the relay. + Local, + /// Hold only a broker credential; no agent key or relay route. + Broker, +} + +/// Keyless broker provisioning. The credential is redacted from `Debug`. +#[derive(Clone)] +pub struct BrokerConfig { + pub base_url: String, + pub credential: String, + pub poll_interval: std::time::Duration, +} + +impl std::fmt::Debug for BrokerConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BrokerConfig") + .field("base_url", &self.base_url) + .field("credential", &"") + .field("poll_interval", &self.poll_interval) + .finish() + } +} + #[derive(Debug, Clone, Copy, clap::ValueEnum)] pub enum DedupMode { Drop, @@ -246,9 +273,29 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_RELAY_URL", default_value = "ws://localhost:3000")] pub relay_url: String, - #[arg(long, env = "BUZZ_PRIVATE_KEY", hide_env_values = true)] + /// Runtime substrate. Broker mode rejects a private key and requires the + /// broker URL, credential, and explicit channel list. + #[arg(long, env = "BUZZ_AGENT_MODE", default_value = "local")] + pub agent_mode: AgentMode, + + #[arg( + long, + env = "BUZZ_PRIVATE_KEY", + hide_env_values = true, + default_value = "" + )] pub private_key: String, + #[arg(long, env = "BUZZ_BROKER_URL")] + pub broker_url: Option, + + #[arg(long, env = "BUZZ_BROKER_CREDENTIAL", hide_env_values = true)] + pub broker_credential: Option, + + /// Delay between broker polling sweeps in keyless mode. + #[arg(long, env = "BUZZ_BROKER_POLL_INTERVAL_MS", default_value_t = 1000)] + pub broker_poll_interval_ms: u64, + /// Agent owner pubkey (64-char hex). Used for --respond-to=owner-only gate. #[arg(long, env = "BUZZ_ACP_AGENT_OWNER")] pub agent_owner: Option, @@ -517,6 +564,8 @@ pub struct ChannelFilter { #[derive(Debug)] pub struct Config { pub keys: Keys, + pub agent_mode: AgentMode, + pub broker: Option, pub relay_url: String, pub agent_command: String, pub agent_args: Vec, @@ -871,7 +920,70 @@ impl Config { /// tests can construct `CliArgs` via `CliArgs::try_parse_from` and exercise the full /// validation path without going through process args. pub fn from_args(mut args: CliArgs) -> Result { - let keys = Keys::parse(&args.private_key)?; + let broker = match args.agent_mode { + AgentMode::Local => None, + AgentMode::Broker => { + if !args.private_key.is_empty() { + return Err(ConfigError::ConfigFile( + "broker mode is keyless — unset BUZZ_PRIVATE_KEY / --private-key".into(), + )); + } + let base_url = args.broker_url.take().ok_or_else(|| { + ConfigError::ConfigFile( + "BUZZ_BROKER_URL / --broker-url is required in broker mode".into(), + ) + })?; + let credential = args.broker_credential.take().ok_or_else(|| { + ConfigError::ConfigFile( + "BUZZ_BROKER_CREDENTIAL / --broker-credential is required in broker mode" + .into(), + ) + })?; + if args.broker_poll_interval_ms < 100 { + return Err(ConfigError::ConfigFile( + "broker poll interval must be at least 100ms".into(), + )); + } + let channels = args.channels.as_ref().ok_or_else(|| { + ConfigError::ConfigFile( + "--channels / BUZZ_ACP_CHANNELS is required in broker mode; channel discovery is host-owned and not in the frozen contract" + .into(), + ) + })?; + if channels.is_empty() + || channels + .iter() + .any(|channel| Uuid::parse_str(channel).is_err()) + { + return Err(ConfigError::ConfigFile( + "every broker-mode --channels entry must be a channel UUID".into(), + )); + } + if args.respond_to != RespondTo::OwnerOnly { + return Err(ConfigError::ConfigFile( + "broker mode currently requires --respond-to owner-only because the frozen contract does not expose channel or sibling-profile metadata" + .into(), + )); + } + if args.agent_owner.is_none() { + return Err(ConfigError::ConfigFile( + "--agent-owner / BUZZ_ACP_AGENT_OWNER is required in broker mode".into(), + )); + } + Some(BrokerConfig { + base_url, + credential, + poll_interval: std::time::Duration::from_millis(args.broker_poll_interval_ms), + }) + } + }; + let keys = match args.agent_mode { + AgentMode::Local => Keys::parse(&args.private_key)?, + // Never used as the agent identity. A placeholder keeps the local + // runtime's concrete types intact while the broker path diverges + // before relay setup; it is never exported to subprocesses. + AgentMode::Broker => Keys::generate(), + }; // Best-effort zeroize: overwrite the raw private key string to reduce // exposure via core dumps or heap inspection (#41). Without the `zeroize` // crate we can only clear the String — the allocator may retain copies. @@ -1076,6 +1188,18 @@ impl Config { // Spawned desktop agents now carry a complete instance snapshot. Team // instructions arrive independently so they can be layered at runtime. let mut persona_env_vars = Vec::new(); + if let Some(broker) = broker.as_ref() { + persona_env_vars.extend([ + ("BUZZ_AGENT_MODE".into(), "broker".into()), + ("BUZZ_BROKER_URL".into(), broker.base_url.clone()), + ("BUZZ_BROKER_CREDENTIAL".into(), broker.credential.clone()), + // Explicit tombstones keep inherited local credentials and + // routing out of the spawned agent process. + ("BUZZ_RELAY_URL".into(), String::new()), + ("BUZZ_PRIVATE_KEY".into(), String::new()), + ("BUZZ_AUTH_TAG".into(), String::new()), + ]); + } let model = args.model; // Inject CODEX_CONFIG so the @agentclientprotocol/codex-acp adapter (1.x) @@ -1093,6 +1217,8 @@ impl Config { let config = Config { keys, + agent_mode: args.agent_mode, + broker, relay_url: args.relay_url, agent_command, agent_args, @@ -1119,11 +1245,15 @@ impl Config { channels_override: args.channels, no_mention_filter: args.no_mention_filter, config_path: args.config, - context_message_limit: args.context_message_limit, + context_message_limit: if args.agent_mode == AgentMode::Broker { + 0 + } else { + args.context_message_limit + }, max_turns_per_session: args.max_turns_per_session, - presence_enabled: !args.no_presence, - typing_enabled: !args.no_typing, - memory_enabled: args.memory && !args.no_memory, + presence_enabled: args.agent_mode == AgentMode::Local && !args.no_presence, + typing_enabled: args.agent_mode == AgentMode::Local && !args.no_typing, + memory_enabled: args.agent_mode == AgentMode::Local && args.memory && !args.no_memory, model, effort_level: args.effort_level, session_title: args @@ -1136,7 +1266,7 @@ impl Config { allowed_respond_to, persona_env_vars, has_generated_codex_config, - relay_observer: args.relay_observer, + relay_observer: args.agent_mode == AgentMode::Local && args.relay_observer, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1150,6 +1280,20 @@ impl Config { /// Human-readable summary (no secrets). pub fn summary(&self) -> String { + let transport_detail = match (&self.agent_mode, &self.broker) { + (AgentMode::Local, _) => format!( + "mode=local relay={} pubkey={}", + self.relay_url, + self.keys.public_key().to_hex() + ), + (AgentMode::Broker, Some(broker)) => format!( + "mode=broker broker={} pubkey=(derived-at-connect)", + broker.base_url + ), + (AgentMode::Broker, None) => { + "mode=broker broker=(missing) pubkey=(derived-at-connect)".into() + } + }; let respond_to_detail = match &self.respond_to { RespondTo::Allowlist => { format!("respond_to=allowlist({})", self.respond_to_allowlist.len()) @@ -1164,9 +1308,8 @@ impl Config { format!(" allowed_respond_to=[{}]", modes.join(",")) }; format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", - self.relay_url, - self.keys.public_key().to_hex(), + "{} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", + transport_detail, self.agent_command, self.agent_args.join(" "), self.mcp_command, @@ -1474,6 +1617,8 @@ mod tests { fn test_config(mode: SubscribeMode) -> Config { Config { keys: nostr::Keys::generate(), + agent_mode: AgentMode::Local, + broker: None, relay_url: "ws://localhost:3000".into(), agent_command: "goose".into(), agent_args: vec!["acp".into()], @@ -1519,6 +1664,95 @@ mod tests { } } + fn broker_args(extra: &[&str]) -> CliArgs { + const CHANNEL: &str = "5df7dfa8-e919-43df-8efd-f1dcb8af7071"; + const OWNER: &str = "a02c4e0850e5e612b4ddf95dbe2f5c56467cf27c6552203bc833ff438fb31971"; + let mut args = vec![ + "buzz-acp", + "--agent-mode", + "broker", + "--private-key", + "", + "--broker-url", + "http://127.0.0.1:8787", + "--broker-credential", + "cred", + "--channels", + CHANNEL, + "--agent-owner", + OWNER, + ]; + args.extend_from_slice(extra); + CliArgs::parse_from(args) + } + + #[test] + fn broker_mode_is_keyless_and_disables_relay_only_features() { + let config = Config::from_args(broker_args(&[])).expect("valid broker config"); + + assert_eq!(config.agent_mode, AgentMode::Broker); + assert!(config.broker.is_some()); + assert!(!config.presence_enabled); + assert!(!config.typing_enabled); + assert!(!config.memory_enabled); + assert!(!config.relay_observer); + assert_eq!(config.context_message_limit, 0); + assert!(config + .persona_env_vars + .iter() + .any(|(name, value)| name == "BUZZ_AGENT_MODE" && value == "broker")); + assert!(config + .persona_env_vars + .iter() + .any(|(name, value)| name == "BUZZ_PRIVATE_KEY" && value.is_empty())); + } + + #[test] + fn broker_mode_rejects_an_agent_private_key() { + let key = "1".repeat(64); + let mut args = broker_args(&[]); + args.private_key = key; + let result = Config::from_args(args); + + assert!(result + .expect_err("private key must fail closed") + .to_string() + .contains("keyless")); + } + + #[test] + fn broker_mode_requires_explicit_channels() { + const OWNER: &str = "a02c4e0850e5e612b4ddf95dbe2f5c56467cf27c6552203bc833ff438fb31971"; + let args = CliArgs::parse_from([ + "buzz-acp", + "--agent-mode", + "broker", + "--private-key", + "", + "--broker-url", + "http://127.0.0.1:8787", + "--broker-credential", + "cred", + "--agent-owner", + OWNER, + ]); + + assert!(Config::from_args(args) + .expect_err("channels are required") + .to_string() + .contains("--channels")); + } + + #[test] + fn broker_mode_requires_owner_only_author_gate() { + let result = Config::from_args(broker_args(&["--respond-to", "anyone"])); + + assert!(result + .expect_err("broker mode must fail closed without channel metadata") + .to_string() + .contains("owner-only")); + } + fn make_rule( name: &str, channels: ChannelScope, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 25c6e54905..87aaac7cbe 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -11,6 +11,7 @@ mod prompt_framing; mod prompt_project; mod queue; mod relay; +mod runtime_transport; mod setup_mode; mod usage; @@ -32,7 +33,7 @@ use buzz_core::observer::{ }; use clap::Parser; use config::{ - AuthAgentArgs, AuthMethodsArgs, AuthenticateArgs, Config, DedupMode, ModelsArgs, + AgentMode, AuthAgentArgs, AuthMethodsArgs, AuthenticateArgs, Config, DedupMode, ModelsArgs, MultipleEventHandling, RespondTo, SubscribeMode, }; use filter::SubscriptionRule; @@ -45,6 +46,7 @@ use pool::{ use pool_lifecycle::PoolLifecycle; use queue::{CancelReason, EventQueue, FlushBatch, QueuedEvent, ThreadTags}; use relay::{HarnessRelay, RelayEventPublisher}; +use runtime_transport::RuntimeTransport; use tokio::sync::{mpsc, watch}; use tracing_subscriber::EnvFilter; use uuid::Uuid; @@ -1954,6 +1956,10 @@ async fn tokio_main() -> Result<()> { if let Some(payload) = setup_mode::SetupPayload::from_env() .map_err(|e| anyhow::anyhow!("setup payload error: {e}"))? { + ensure!( + config.agent_mode == AgentMode::Local, + "setup-listener mode is not available in keyless broker mode" + ); tracing::info!("buzz-acp: setup payload present, entering setup-listener mode"); return setup_mode::run_setup_listener(config, payload).await; } @@ -1996,18 +2002,50 @@ async fn tokio_main() -> Result<()> { .unwrap_or_default() .as_secs(); - let pubkey_hex = config.keys.public_key().to_hex(); + let local_pubkey_hex = config.keys.public_key().to_hex(); // Parse BUZZ_AUTH_TAG into a nostr::Tag for NIP-OA relay membership delegation. - let relay_auth_tag: Option = std::env::var("BUZZ_AUTH_TAG") - .ok() + let relay_auth_tag: Option = (config.agent_mode == AgentMode::Local) + .then(|| std::env::var("BUZZ_AUTH_TAG").ok()) + .flatten() .filter(|s| !s.is_empty()) .and_then(|s| buzz_sdk::nip_oa::parse_auth_tag(&s).ok()); - let mut relay = - HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) + let (mut relay, pubkey_hex) = match config.agent_mode { + AgentMode::Local => { + let relay = HarnessRelay::connect( + &config.relay_url, + &config.keys, + &local_pubkey_hex, + relay_auth_tag, + ) .await .map_err(|e| anyhow::anyhow!("relay connect error: {e}"))?; + (RuntimeTransport::local(relay), local_pubkey_hex) + } + AgentMode::Broker => { + let broker = config + .broker + .as_ref() + .expect("broker config validated for broker mode"); + let channel_ids = config + .channels_override + .as_ref() + .expect("channels validated for broker mode") + .iter() + .map(|channel| Uuid::parse_str(channel).expect("channel validated")) + .collect(); + RuntimeTransport::broker( + broker.base_url.clone(), + broker.credential.clone(), + channel_ids, + broker.poll_interval, + config.keys.clone(), + ) + .await + .map_err(|e| anyhow::anyhow!("broker connect error: {e}"))? + } + }; // Tell the relay background task the watermark so it can use // `since = watermark - 5s` on the first REQ instead of `since=now`. @@ -2017,19 +2055,30 @@ async fn tokio_main() -> Result<()> { tracing::warn!("failed to set startup watermark: {e}"); } - tracing::info!("connected to relay at {}", config.relay_url); + match config.agent_mode { + AgentMode::Local => tracing::info!("connected to relay at {}", config.relay_url), + AgentMode::Broker => tracing::info!("connected to broker as {pubkey_hex}"), + } relay .subscribe_membership_notifications() .await .map_err(|e| anyhow::anyhow!("membership notification subscribe error: {e}"))?; - tracing::info!("subscribed to membership notifications"); + match config.agent_mode { + AgentMode::Local => tracing::info!("subscribed to membership notifications"), + AgentMode::Broker => tracing::info!( + "broker contract has no membership notifications; using configured channels" + ), + } let presence_publisher = relay.event_publisher(); let presence_keys = config.keys.clone(); // Priority: BUZZ_AUTH_TAG (NIP-OA attestation) → --agent-owner flag. - let startup_owner: Option = resolve_agent_owner(&config); + let startup_owner: Option = match config.agent_mode { + AgentMode::Local => resolve_agent_owner(&config), + AgentMode::Broker => config.agent_owner.clone(), + }; if let Some(ref owner) = startup_owner { tracing::info!("agent owner: {owner}"); } else { @@ -2192,6 +2241,10 @@ async fn tokio_main() -> Result<()> { let base_prompt_content = config.base_prompt_content.take(); let cwd = current_working_directory()?; + let channel_info = match config.agent_mode { + AgentMode::Local => pool::ChannelInfoResolver::new(channel_info_map, relay.rest_client()), + AgentMode::Broker => pool::ChannelInfoResolver::without_fallback(channel_info_map), + }; let ctx = Arc::new(PromptContext { mcp_servers: build_mcp_servers(&config), initial_message: config.initial_message.clone(), @@ -2211,15 +2264,20 @@ async fn tokio_main() -> Result<()> { }, heartbeat_prompt: config.heartbeat_prompt.clone(), cwd, + relay_features_enabled: config.agent_mode == AgentMode::Local, rest_client: relay.rest_client(), - channel_info: pool::ChannelInfoResolver::new(channel_info_map, relay.rest_client()), + channel_info, context_message_limit: config.context_message_limit, max_turns_per_session: config.max_turns_per_session, permission_mode: config.permission_mode, agent_keys: config.keys.clone(), - agent_owner_pubkey: startup_owner - .as_deref() - .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()), + agent_owner_pubkey: (config.agent_mode == AgentMode::Local) + .then(|| { + startup_owner + .as_deref() + .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()) + }) + .flatten(), memory_enabled: config.memory_enabled, harness_name: crate::config::normalize_agent_command_identity(&config.agent_command), relay_url: config.relay_url.clone(), @@ -2930,7 +2988,7 @@ async fn tokio_main() -> Result<()> { // Fire-and-forget: on rare fast-failure paths the // guard's cleanup may race with this add, leaving a // cosmetic stale 👀. Acceptable — see ReactionGuard docs. - if accepted { + if accepted && config.agent_mode == AgentMode::Local { let rc = ctx.rest_client.clone(); let eid = event_id_hex.clone(); tokio::spawn(async move { @@ -3161,7 +3219,7 @@ async fn tokio_main() -> Result<()> { &respawn_tx, &mut respawn_tasks, observer.clone(), - Some(&ctx.rest_client), + (config.agent_mode == AgentMode::Local).then_some(&ctx.rest_client), ) == LoopAction::Exit { break; @@ -5079,31 +5137,53 @@ fn build_mcp_servers(config: &Config) -> Vec { command: config.mcp_command.clone(), args: vec![], env: { - let mut env = vec![ - EnvVar { - name: "BUZZ_RELAY_URL".into(), - value: config.relay_url.clone(), - }, - EnvVar { - name: "BUZZ_PRIVATE_KEY".into(), - // bech32 encoding of a valid secret key is infallible. - // Panic here is correct: injecting a bogus secret would cause - // delayed, hard-to-diagnose agent failures downstream. - value: config - .keys - .secret_key() - .to_bech32() - .expect("secret key bech32 encoding should never fail"), - }, - ]; + let mut env = match config.agent_mode { + AgentMode::Local => vec![ + EnvVar { + name: "BUZZ_RELAY_URL".into(), + value: config.relay_url.clone(), + }, + EnvVar { + name: "BUZZ_PRIVATE_KEY".into(), + // bech32 encoding of a valid secret key is infallible. + value: config + .keys + .secret_key() + .to_bech32() + .expect("secret key bech32 encoding should never fail"), + }, + ], + AgentMode::Broker => { + let broker = config + .broker + .as_ref() + .expect("broker config validated for broker mode"); + vec![ + EnvVar { + name: "BUZZ_AGENT_MODE".into(), + value: "broker".into(), + }, + EnvVar { + name: "BUZZ_BROKER_URL".into(), + value: broker.base_url.clone(), + }, + EnvVar { + name: "BUZZ_BROKER_CREDENTIAL".into(), + value: broker.credential.clone(), + }, + ] + } + }; // Forward BUZZ_AUTH_TAG (NIP-OA owner attestation credential) // so the MCP server can attach it to every signed event. - if let Ok(auth_tag) = std::env::var("BUZZ_AUTH_TAG") { - if !auth_tag.is_empty() { - env.push(EnvVar { - name: "BUZZ_AUTH_TAG".into(), - value: auth_tag, - }); + if config.agent_mode == AgentMode::Local { + if let Ok(auth_tag) = std::env::var("BUZZ_AUTH_TAG") { + if !auth_tag.is_empty() { + env.push(EnvVar { + name: "BUZZ_AUTH_TAG".into(), + value: auth_tag, + }); + } } } // Forward the agent's display name so dev-mcp can use it as the git @@ -6792,6 +6872,8 @@ mod build_mcp_servers_tests { fn test_config() -> Config { Config { keys: nostr::Keys::generate(), + agent_mode: config::AgentMode::Local, + broker: None, relay_url: "ws://localhost:3000".into(), agent_command: "goose".into(), agent_args: vec!["acp".into()], @@ -6856,6 +6938,27 @@ mod build_mcp_servers_tests { ); } + #[test] + fn broker_mcp_server_receives_only_broker_credentials() { + let mut config = test_config(); + config.agent_mode = config::AgentMode::Broker; + config.broker = Some(config::BrokerConfig { + base_url: "http://127.0.0.1:8787".into(), + credential: "broker-token".into(), + poll_interval: std::time::Duration::from_secs(1), + }); + + let servers = build_mcp_servers(&config); + let server = &servers[0]; + let names: Vec<&str> = server.env.iter().map(|env| env.name.as_str()).collect(); + assert!(names.contains(&"BUZZ_AGENT_MODE")); + assert!(names.contains(&"BUZZ_BROKER_URL")); + assert!(names.contains(&"BUZZ_BROKER_CREDENTIAL")); + assert!(!names.contains(&"BUZZ_RELAY_URL")); + assert!(!names.contains(&"BUZZ_PRIVATE_KEY")); + assert!(!names.contains(&"BUZZ_AUTH_TAG")); + } + #[test] fn session_new_mcp_server_forwards_buzz_auth_tag() { let _guard = ENV_LOCK.lock().unwrap(); @@ -7013,6 +7116,8 @@ mod error_outcome_emission_tests { fn test_config() -> Config { Config { keys: nostr::Keys::generate(), + agent_mode: config::AgentMode::Local, + broker: None, relay_url: "ws://localhost:3000".into(), // `true` exits cleanly, so the async respawn fails fast and // harmlessly off the JoinSet — irrelevant to the synchronous diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index f18f7d6fea..265f8eaff2 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -561,7 +561,7 @@ const PROJECT_INFO_CACHE_TTL: std::time::Duration = std::time::Duration::from_se pub struct ChannelInfoResolver { cache: std::sync::Arc>>, projects: std::sync::Arc>>, - rest_client: RestClient, + rest_client: Option, } impl ChannelInfoResolver { @@ -586,7 +586,33 @@ impl ChannelInfoResolver { Self { cache: std::sync::Arc::new(std::sync::RwLock::new(cache)), projects: std::sync::Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())), - rest_client, + rest_client: Some(rest_client), + } + } + + /// Build a resolver that never falls back to a direct relay metadata read. + /// + /// Broker mode uses this constructor because the client has no relay route. + /// Unknown channel types therefore remain unresolved and callers fail closed. + pub fn without_fallback(startup: std::collections::HashMap) -> Self { + let cache = startup + .into_iter() + .filter_map(|(id, info)| { + (info.channel_type != "unknown").then_some(( + id, + PromptChannelInfo { + name: info.name, + channel_type: info.channel_type, + description: info.description, + project: None, + }, + )) + }) + .collect(); + Self { + cache: std::sync::Arc::new(std::sync::RwLock::new(cache)), + projects: std::sync::Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())), + rest_client: None, } } @@ -599,7 +625,8 @@ impl ChannelInfoResolver { { return Some(info); } - let info = fetch_channel_info(channel_id, &self.rest_client).await?; + let rest_client = self.rest_client.as_ref()?; + let info = fetch_channel_info(channel_id, rest_client).await?; if let Ok(mut cache) = self.cache.write() { cache.insert(channel_id, info.clone()); } @@ -662,7 +689,11 @@ impl ChannelInfoResolver { { return Ok(fresh.value.clone()); } - let fetched = match fetch_project_home_for_channel(channel_id, &self.rest_client).await { + // Broker mode has no relay/REST route: no project context, fail closed. + let Some(rest_client) = self.rest_client.as_ref() else { + return Ok(None); + }; + let fetched = match fetch_project_home_for_channel(channel_id, rest_client).await { Ok(fetched) => fetched, Err(error) => { if let Some(project) = cached.and_then(|stale| stale.value) { @@ -713,6 +744,9 @@ pub struct PromptContext { /// (`include_str!`) is inherently `'static`. pub base_prompt: Option<&'static str>, pub cwd: String, + /// Whether direct relay-only enrichments and housekeeping are available. + /// False in broker mode, where the runtime has no relay route. + pub relay_features_enabled: bool, /// REST client for pre-prompt context fetches (thread/DM history). pub rest_client: RestClient, /// Shared channel metadata for startup-known and dynamically joined channels. @@ -1952,7 +1986,9 @@ pub async fn run_prompt_task( .as_ref() .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect()) .unwrap_or_default(); - let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone()); + let _reaction_guard = ctx + .relay_features_enabled + .then(|| ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone())); // Resolve project authority exactly once, before any ACP session creation or // initial-message delivery. An indeterminate result is a local relay-state @@ -2071,13 +2107,15 @@ pub async fn run_prompt_task( resolve_new_session_channel_context(resolved_channel_info.as_ref()).await; title_channel = resolved_channel; origin_channel_type = resolved_channel_type; - if let Some(owner) = ctx.agent_owner_pubkey.as_ref() { - huddle_instructions = - fetch_huddle_instructions(*cid, owner, &ctx.rest_client).await; + if ctx.relay_features_enabled { + if let Some(owner) = ctx.agent_owner_pubkey.as_ref() { + huddle_instructions = + fetch_huddle_instructions(*cid, owner, &ctx.rest_client).await; + } } // A confirmed DM never receives a canvas section; an undeterminable // channel type fails closed as a DM for the same reason. - if needs_canvas && !is_dm { + if ctx.relay_features_enabled && needs_canvas && !is_dm { if let Some(section) = fetch_canvas_section(*cid, &ctx.rest_client).await { pending_canvas = Some((*cid, section)); } @@ -2495,8 +2533,11 @@ pub async fn run_prompt_task( conversation_context.as_ref(), )); - let profile_lookup = - fetch_prompt_profile_lookup(b, conversation_context.as_ref(), &ctx.rest_client).await; + let profile_lookup = if ctx.relay_features_enabled { + fetch_prompt_profile_lookup(b, conversation_context.as_ref(), &ctx.rest_client).await + } else { + None + }; let known_names: Vec<&str> = profile_lookup .iter() @@ -2549,7 +2590,7 @@ pub async fn run_prompt_task( // 💬 — fire-and-forget so the prompt fires immediately. // The guard's cleanup (spawned on drop) removes 💬 after the turn completes. // A brief race where 💬 appears slightly after the agent starts is acceptable. - if !reaction_ids.is_empty() { + if ctx.relay_features_enabled && !reaction_ids.is_empty() { let rest = ctx.rest_client.clone(); let ids = reaction_ids.clone(); tokio::spawn(async move { @@ -3498,6 +3539,9 @@ async fn fetch_conversation_context( channel_info: &Option, ctx: &PromptContext, ) -> Option { + if !ctx.relay_features_enabled { + return None; + } let limit = ctx.context_message_limit; let is_dm = channel_info .as_ref() @@ -8213,6 +8257,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" heartbeat_prompt: None, base_prompt: None, cwd: ".".to_string(), + relay_features_enabled: true, rest_client: RestClient { http: reqwest::Client::new(), base_url: "http://127.0.0.1:0".to_string(), diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 6188e57a11..1391be78ae 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -658,6 +658,14 @@ impl RelayEventPublisher { .map_err(|_| RelayError::ConnectionClosed) } + /// A publisher with no relay behind it. Used only by broker-mode plumbing + /// for code paths disabled by provisioning (presence, typing, observer). + pub(crate) fn disabled() -> Self { + let (cmd_tx, cmd_rx) = mpsc::channel(1); + drop(cmd_rx); + Self { cmd_tx } + } + /// Test-only publisher pair: published events are forwarded to the /// returned receiver instead of a live relay socket. #[cfg(test)] diff --git a/crates/buzz-acp/src/runtime_transport.rs b/crates/buzz-acp/src/runtime_transport.rs new file mode 100644 index 0000000000..41917f5748 --- /dev/null +++ b/crates/buzz-acp/src/runtime_transport.rs @@ -0,0 +1,356 @@ +//! Runtime event transport selected by provisioning. +//! +//! Local mode delegates to the existing authenticated relay client. Broker +//! mode polls the frozen `channel.read` action and deliberately provides no +//! relay publisher: relay-only housekeeping is disabled by broker-mode config. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::time::Duration; + +use buzz_broker_client::HttpBrokerClient; +use buzz_sdk::broker::{ + ActionArgs, ActionOutcome, BrokerClientExt, BrokerRequest, BrokerResult, ChannelReadArgs, + StorageAddressArgs, +}; +use nostr::{Event, Keys}; +use tokio::time::MissedTickBehavior; +use uuid::Uuid; + +use crate::config::ChannelFilter; +use crate::relay::{ + BuzzEvent, ChannelInfo, HarnessRelay, RelayError, RelayEventPublisher, RestClient, +}; + +pub enum RuntimeTransport { + Local(HarnessRelay), + Broker(BrokerRuntime), +} + +pub struct BrokerRuntime { + client: HttpBrokerClient, + channel_ids: Vec, + filters: HashMap, + cursors: HashMap, + pending: VecDeque, + seen_current: HashSet, + seen_previous: HashSet, + poll: tokio::time::Interval, + placeholder_keys: Keys, +} + +impl BrokerRuntime { + pub async fn connect( + base_url: String, + credential: String, + channel_ids: Vec, + poll_interval: Duration, + placeholder_keys: Keys, + ) -> Result<(Self, String), RelayError> { + let client = HttpBrokerClient::new(base_url, credential); + let outcome = execute( + &client, + ActionArgs::StorageAddress(StorageAddressArgs { + slug: "core".into(), + }), + ) + .await?; + let ActionOutcome::StorageAddress(address) = outcome else { + return Err(RelayError::Http( + "broker returned the wrong outcome for storage.address".into(), + )); + }; + let agent_pubkey = address.author_pubkey.as_str().to_string(); + let mut poll = tokio::time::interval(poll_interval); + poll.set_missed_tick_behavior(MissedTickBehavior::Skip); + Ok(( + Self { + client, + channel_ids, + filters: HashMap::new(), + cursors: HashMap::new(), + pending: VecDeque::new(), + seen_current: HashSet::new(), + seen_previous: HashSet::new(), + poll, + placeholder_keys, + }, + agent_pubkey, + )) + } + + fn remember(&mut self, event_id: String) -> bool { + if self.seen_current.contains(&event_id) || self.seen_previous.contains(&event_id) { + return false; + } + self.seen_current.insert(event_id); + if self.seen_current.len() >= 2_000 { + self.seen_previous = std::mem::take(&mut self.seen_current); + } + true + } + + async fn next_event(&mut self) -> Option { + loop { + if let Some(event) = self.pending.pop_front() { + return Some(event); + } + self.poll.tick().await; + let subscriptions: Vec<(Uuid, bool)> = self + .channel_ids + .iter() + .filter_map(|channel_id| { + self.filters + .get(channel_id) + .map(|filter| (*channel_id, filter.require_mention)) + }) + .collect(); + for (channel_id, mentions_only) in subscriptions { + let result = execute( + &self.client, + ActionArgs::ChannelRead(ChannelReadArgs { + channel_id: channel_id.to_string(), + root_event_id: None, + mentions_only, + cursor: self.cursors.get(&channel_id).cloned(), + limit: Some(100), + }), + ) + .await; + let page = match result { + Ok(ActionOutcome::ChannelRead(page)) => page, + Ok(_) => { + tracing::warn!(%channel_id, "broker returned the wrong channel.read outcome"); + continue; + } + Err(error) => { + tracing::warn!(%channel_id, "broker channel.read failed: {error}"); + continue; + } + }; + if let Some(cursor) = page.next_cursor { + self.cursors.insert(channel_id, cursor); + } + for message in page.messages { + if let Err(error) = message.verify() { + tracing::warn!(%channel_id, "broker returned an unverifiable event: {error}"); + continue; + } + let event = message.0; + let addressed_channel = event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "h") + .and_then(|tag| tag.content()) + .and_then(|value| Uuid::parse_str(value).ok()); + if addressed_channel != Some(channel_id) { + tracing::warn!(%channel_id, "broker returned an event for a different channel"); + continue; + } + if self.remember(event.id.to_hex()) { + self.pending.push_back(BuzzEvent { channel_id, event }); + } + } + } + } + } +} + +async fn execute(client: &HttpBrokerClient, args: ActionArgs) -> Result { + let request = BrokerRequest::new(Uuid::new_v4().to_string(), args) + .and_then(BrokerRequest::prepare) + .map_err(|error| RelayError::Http(format!("broker request: {error}")))?; + let response = client + .execute(&request) + .await + .map_err(|error| RelayError::Http(format!("broker transport: {error}")))?; + match response.into_envelope().result { + BrokerResult::Succeeded { outcome } => Ok(outcome), + BrokerResult::Failed { error } | BrokerResult::Indeterminate { error } => { + Err(RelayError::Http(format!( + "broker verdict: {} [{}]", + error.message, + error.code.as_str() + ))) + } + } +} + +impl RuntimeTransport { + pub fn local(relay: HarnessRelay) -> Self { + Self::Local(relay) + } + + pub async fn broker( + base_url: String, + credential: String, + channel_ids: Vec, + poll_interval: Duration, + placeholder_keys: Keys, + ) -> Result<(Self, String), RelayError> { + let (runtime, pubkey) = BrokerRuntime::connect( + base_url, + credential, + channel_ids, + poll_interval, + placeholder_keys, + ) + .await?; + Ok((Self::Broker(runtime), pubkey)) + } + + pub async fn set_startup_watermark(&self, timestamp: u64) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.set_startup_watermark(timestamp).await, + Self::Broker(_) => Ok(()), + } + } + + pub async fn subscribe_membership_notifications(&mut self) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.subscribe_membership_notifications().await, + Self::Broker(_) => Ok(()), + } + } + + pub async fn subscribe_observer_controls(&mut self) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.subscribe_observer_controls().await, + Self::Broker(_) => Ok(()), + } + } + + pub fn take_observer_control_rx(&mut self) -> Option> { + match self { + Self::Local(relay) => relay.take_observer_control_rx(), + Self::Broker(_) => None, + } + } + + pub async fn discover_channels(&self) -> Result, RelayError> { + match self { + Self::Local(relay) => relay.discover_channels().await, + Self::Broker(runtime) => Ok(runtime + .channel_ids + .iter() + .map(|channel_id| { + ( + *channel_id, + ChannelInfo { + name: channel_id.to_string(), + // The frozen broker contract does not expose channel + // metadata. Keep the type unknown so the runtime's + // author gate treats it as a DM (fail closed). + channel_type: "unknown".into(), + description: None, + }, + ) + }) + .collect()), + } + } + + pub async fn subscribe_channel( + &mut self, + channel_id: Uuid, + filter: ChannelFilter, + ) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.subscribe_channel(channel_id, filter).await, + Self::Broker(runtime) => { + runtime.filters.insert(channel_id, filter); + Ok(()) + } + } + } + + pub async fn subscribe_channel_from( + &mut self, + channel_id: Uuid, + filter: ChannelFilter, + replay_since: Option, + ) -> Result<(), RelayError> { + match self { + Self::Local(relay) => { + relay + .subscribe_channel_from(channel_id, filter, replay_since) + .await + } + Self::Broker(runtime) => { + runtime.filters.insert(channel_id, filter); + Ok(()) + } + } + } + + pub async fn unsubscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.unsubscribe_channel(channel_id).await, + Self::Broker(runtime) => { + runtime.filters.remove(&channel_id); + Ok(()) + } + } + } + + pub async fn next_event(&mut self) -> Option { + match self { + Self::Local(relay) => relay.next_event().await, + Self::Broker(runtime) => runtime.next_event().await, + } + } + + pub async fn reconnect(&mut self) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.reconnect().await, + Self::Broker(_) => Ok(()), + } + } + + pub fn event_publisher(&self) -> RelayEventPublisher { + match self { + Self::Local(relay) => relay.event_publisher(), + Self::Broker(_) => RelayEventPublisher::disabled(), + } + } + + pub fn rest_client(&self) -> RestClient { + match self { + Self::Local(relay) => relay.rest_client(), + Self::Broker(runtime) => RestClient { + http: reqwest::Client::new(), + base_url: "http://127.0.0.1:1".into(), + keys: runtime.placeholder_keys.clone(), + auth_tag_json: None, + }, + } + } + + pub fn build_typing_event( + &self, + channel_id: Uuid, + root_event_id: Option<&str>, + parent_event_id: Option<&str>, + ) -> Result { + match self { + Self::Local(relay) => { + relay.build_typing_event(channel_id, root_event_id, parent_event_id) + } + Self::Broker(_) => Err(RelayError::Http( + "typing indicators are disabled in broker mode".into(), + )), + } + } + + pub fn try_publish_event(&self, event: Event) -> Result<(), RelayError> { + match self { + Self::Local(relay) => relay.try_publish_event(event), + Self::Broker(_) => Err(RelayError::ConnectionClosed), + } + } + + pub async fn shutdown(self) { + if let Self::Local(relay) = self { + relay.shutdown().await; + } + } +} diff --git a/crates/buzz-broker-client/Cargo.toml b/crates/buzz-broker-client/Cargo.toml new file mode 100644 index 0000000000..f95f84985b --- /dev/null +++ b/crates/buzz-broker-client/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "buzz-broker-client" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Shared HTTP transport for the Buzz agent broker" + +[dependencies] +buzz-sdk = { workspace = true } +reqwest = { workspace = true } +serde_json = { workspace = true } + +[dev-dependencies] +axum = { workspace = true } +tokio = { workspace = true } diff --git a/crates/buzz-cli/src/broker_client.rs b/crates/buzz-broker-client/src/lib.rs similarity index 90% rename from crates/buzz-cli/src/broker_client.rs rename to crates/buzz-broker-client/src/lib.rs index 3bf3f53236..d656cdb5b1 100644 --- a/crates/buzz-cli/src/broker_client.rs +++ b/crates/buzz-broker-client/src/lib.rs @@ -1,14 +1,9 @@ -//! HTTP transport for the agent broker — keyless client mode. +//! Shared HTTP transport for the agent broker. //! //! Implements [`buzz_sdk::broker::BrokerClient`], the transport primitive the //! contract crate deliberately omits: frozen request bytes out, one envelope //! back. Callers use [`buzz_sdk::broker::BrokerClientExt::execute`], which adds -//! the correlation checks; this type must never interpret a verdict. -//! -//! The binding is one `POST /v1/action` with the opaque bearer credential in -//! the `Authorization` header. An envelope is parsed regardless of HTTP status, -//! because a host verdict lives in the body and an intermediary may remap the -//! status line. +//! correlation checks; this type never interprets a verdict. use buzz_sdk::broker::{ BrokerClient, BrokerFuture, BrokerResponse, BrokerTransportError, Dispatch, PreparedRequest, @@ -26,8 +21,7 @@ pub struct HttpBrokerClient { } impl HttpBrokerClient { - /// A client posting to `base_url` (scheme + authority, no path) with - /// `credential` as its bearer token. + /// A client posting to `base_url` with `credential` as its bearer token. pub fn new(base_url: impl Into, credential: impl Into) -> Self { Self::with_client(base_url, credential, reqwest::Client::new()) } @@ -175,16 +169,12 @@ mod tests { r#"{{"type":"broker_result","protocolVersion":1,"requestId":"{}","status":"failed","error":{{"code":"unauthenticated","message":"nope"}}}}"#, req.request_id(), ); - // A rejected credential arrives as HTTP 200 with a Failed envelope. let (base, _) = spawn(StatusCode::OK, body).await; let client = HttpBrokerClient::new(base, CRED); let validated = client.execute(&req).await.expect("a verdict"); - match validated.result() { - BrokerResult::Failed { .. } => {} - other => panic!("expected Failed, got {other:?}"), - } + assert!(matches!(validated.result(), BrokerResult::Failed { .. })); } #[tokio::test] diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 59d1bb2cee..6d65a631f6 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -43,6 +43,7 @@ chrono = { workspace = true } # Typed event builders for all write operations buzz-sdk = { workspace = true } +buzz-broker-client = { workspace = true } buzz-core = { workspace = true } # Base64 encoding — NIP-98 event serialization for Authorization header diff --git a/crates/buzz-cli/KEYLESS.md b/crates/buzz-cli/KEYLESS.md index 966e76fa19..0050d1099d 100644 --- a/crates/buzz-cli/KEYLESS.md +++ b/crates/buzz-cli/KEYLESS.md @@ -16,6 +16,13 @@ path), `messages send` / reply, `reactions add`, `users set-profile`, and `mem address ` as an addressing-only bridge for encrypted memory. Every other command still needs the local backend. +The same provisioning variables now select the prototype keyless `buzz-acp` +runtime. It derives its public identity through `storage.address`, polls +configured channels through `channel.read`, and passes broker provisioning to +the CLI used by the spawned agent. See +[`../buzz-acp/README.md`](../buzz-acp/README.md#keyless-broker-mode-prototype) +for the runtime command and its deliberate housekeeping limits. + ## Build ```sh diff --git a/crates/buzz-cli/src/backend.rs b/crates/buzz-cli/src/backend.rs index 971a9ba557..a975848c06 100644 --- a/crates/buzz-cli/src/backend.rs +++ b/crates/buzz-cli/src/backend.rs @@ -18,9 +18,9 @@ use buzz_sdk::broker::{ }; use buzz_sdk::ThreadRef; -use crate::broker_client::HttpBrokerClient; use crate::client::BuzzClient; use crate::error::CliError; +use buzz_broker_client::HttpBrokerClient; /// The operations an agent performs, in the broker's vocabulary. /// diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 0d108cb6b6..8a553dc4a2 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1,6 +1,5 @@ pub mod agent_management; pub mod backend; -pub mod broker_client; mod client; mod commands; mod error;