From 688f4b5c5fa77468bd944cf12dcccdb1fdbcacd3 Mon Sep 17 00:00:00 2001 From: Joel Robotham Date: Fri, 28 Aug 2026 16:17:11 +1000 Subject: [PATCH] feat(keyless): route broker capabilities Signed-off-by: Joel Robotham --- Cargo.lock | 1 + crates/buzz-acp/Cargo.toml | 1 + crates/buzz-acp/README.md | 24 +- crates/buzz-acp/src/config.rs | 74 +++-- crates/buzz-acp/src/engram_fetch.rs | 16 +- crates/buzz-acp/src/lib.rs | 161 ++++++++--- crates/buzz-acp/src/pool.rs | 108 +++++-- crates/buzz-acp/src/relay.rs | 68 ++--- crates/buzz-acp/src/runtime_transport.rs | 341 +++++++++++++++++++++-- crates/buzz-broker-client/src/lib.rs | 133 +++++++-- crates/buzz-cli/KEYLESS.md | 31 ++- crates/buzz-cli/examples/mock_broker.rs | 36 +++ crates/buzz-cli/src/backend.rs | 115 +++++++- crates/buzz-cli/src/commands/mem.rs | 96 ++++++- crates/buzz-cli/src/commands/users.rs | 21 +- crates/buzz-cli/src/lib.rs | 7 +- 16 files changed, 997 insertions(+), 236 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 252eb9b1be..dbdca88cc1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -830,6 +830,7 @@ name = "buzz-acp" version = "0.1.0" dependencies = [ "anyhow", + "axum", "base64 0.22.1", "buzz-broker-client", "buzz-core", diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index 8a3dd9748c..f10b001ab7 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -78,5 +78,6 @@ evalexpr = { workspace = true } nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] +axum = { workspace = true } tokio = { workspace = true, features = ["test-util"] } httparse = "1" diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 3d7656c5f4..d42247b05d 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -87,20 +87,20 @@ 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 +Broker mode requires explicit channel UUIDs and `respond-to=owner-only` because +the contract does not expose channel discovery or metadata. Core-memory reads, +presence, typing, observer telemetry, and turn liveness use broker actions. +Relay-only conversation enrichment, setup nudges, sibling-profile lookup, and +reaction-based turn status remain disabled. The `buzz` CLI operations available +to the spawned agent are documented in [`../buzz-cli/KEYLESS.md`](../buzz-cli/KEYLESS.md). -Broker actions time out after 30 seconds, and an `unauthenticated` polling -verdict terminates the harness so a revoked credential cannot leave a -warn-spamming process that appears healthy. Poll cursors and event deduplication -remain in memory only; restart-window semantics are a required real-host -integration check. +Broker actions time out after 30 seconds and require TLS except on loopback. An +`unauthenticated` polling verdict terminates the harness so a revoked credential +cannot leave a warn-spamming process that appears healthy. Poll cursors and +event deduplication remain in memory only. After a pagination chain is drained, +the runtime returns to cursorless polling and relies on bounded event-ID deduplication so later +messages remain visible. ## Running with Codex diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 8aa052ed07..62ac20565e 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -8,7 +8,7 @@ use std::path::PathBuf; use clap::Parser; use clap::ValueEnum; -use nostr::Keys; +use nostr::{Keys, PublicKey}; use thiserror::Error; use url::Url; use uuid::Uuid; @@ -965,11 +965,17 @@ impl Config { .into(), )); } - if args.agent_owner.is_none() { - return Err(ConfigError::ConfigFile( + let owner = args.agent_owner.as_deref().ok_or_else(|| { + ConfigError::ConfigFile( "--agent-owner / BUZZ_ACP_AGENT_OWNER is required in broker mode".into(), - )); - } + ) + })?; + let owner = PublicKey::from_hex(owner.trim()).map_err(|error| { + ConfigError::ConfigFile(format!( + "broker-mode --agent-owner must be a 64-hex public key: {error}" + )) + })?; + args.agent_owner = Some(owner.to_hex()); Some(BrokerConfig { base_url, credential, @@ -1188,8 +1194,8 @@ 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([ + match broker.as_ref() { + Some(broker) => 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()), @@ -1198,7 +1204,14 @@ impl Config { ("BUZZ_RELAY_URL".into(), String::new()), ("BUZZ_PRIVATE_KEY".into(), String::new()), ("BUZZ_AUTH_TAG".into(), String::new()), - ]); + ]), + None => persona_env_vars.extend([ + ("BUZZ_AGENT_MODE".into(), "local".into()), + ("BUZZ_BROKER_URL".into(), String::new()), + ("BUZZ_BROKER_CREDENTIAL".into(), String::new()), + ("BUZZ_RELAY_URL".into(), args.relay_url.clone()), + ("BUZZ_PRIVATE_KEY".into(), keys.secret_key().to_secret_hex()), + ]), } let model = args.model; @@ -1251,9 +1264,9 @@ impl Config { args.context_message_limit }, max_turns_per_session: args.max_turns_per_session, - 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, + presence_enabled: !args.no_presence, + typing_enabled: !args.no_typing, + memory_enabled: args.memory && !args.no_memory, model, effort_level: args.effort_level, session_title: args @@ -1266,7 +1279,7 @@ impl Config { allowed_respond_to, persona_env_vars, has_generated_codex_config, - relay_observer: args.agent_mode == AgentMode::Local && args.relay_observer, + relay_observer: args.relay_observer, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1687,14 +1700,14 @@ mod tests { } #[test] - fn broker_mode_is_keyless_and_disables_relay_only_features() { + fn broker_mode_is_keyless_and_enables_broker_capabilities() { 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.presence_enabled); + assert!(config.typing_enabled); + assert!(config.memory_enabled); assert!(!config.relay_observer); assert_eq!(config.context_message_limit, 0); assert!(config @@ -1707,6 +1720,35 @@ mod tests { .any(|(name, value)| name == "BUZZ_PRIVATE_KEY" && value.is_empty())); } + #[test] + fn broker_mode_rejects_an_invalid_owner_pubkey() { + let mut args = broker_args(&[]); + args.agent_owner = Some("not-a-pubkey".into()); + assert!(Config::from_args(args) + .expect_err("invalid owner must fail at startup") + .to_string() + .contains("64-hex")); + } + + #[test] + fn explicit_local_mode_tombstones_inherited_broker_provisioning() { + let key = "1".repeat(64); + let args = + CliArgs::parse_from(["buzz-acp", "--agent-mode", "local", "--private-key", &key]); + let config = Config::from_args(args).expect("valid local config"); + let value = |name: &str| { + config + .persona_env_vars + .iter() + .find(|(key, _)| key == name) + .map(|(_, value)| value.as_str()) + }; + assert_eq!(value("BUZZ_AGENT_MODE"), Some("local")); + assert_eq!(value("BUZZ_BROKER_URL"), Some("")); + assert_eq!(value("BUZZ_BROKER_CREDENTIAL"), Some("")); + assert!(value("BUZZ_PRIVATE_KEY").is_some_and(|value| !value.is_empty())); + } + #[test] fn broker_mode_rejects_an_agent_private_key() { let key = "1".repeat(64); diff --git a/crates/buzz-acp/src/engram_fetch.rs b/crates/buzz-acp/src/engram_fetch.rs index d5ae6df076..3863d41922 100644 --- a/crates/buzz-acp/src/engram_fetch.rs +++ b/crates/buzz-acp/src/engram_fetch.rs @@ -25,6 +25,13 @@ pub const ONBOARDING_NUDGE: &str = "No core memory found. \ Use `buzz mem set core \"…\"` to create one (it will hold your identity, \ rules, and goals across sessions). Ask your user about yourself."; +pub fn render_core_section(profile: Option) -> Option { + Some(crate::prompt_framing::semantic_section( + "core-memory", + profile.as_deref().unwrap_or(ONBOARDING_NUDGE), + )) +} + /// Build the rendered prompt section for the agent's core. /// /// Returns: @@ -39,14 +46,7 @@ pub async fn build_core_section( owner: &PublicKey, ) -> Option { match fetch_core_body(rest, agent_keys, owner).await { - Ok(Some(profile)) => Some(crate::prompt_framing::semantic_section( - "core-memory", - &profile, - )), - Ok(None) => Some(crate::prompt_framing::semantic_section( - "core-memory", - ONBOARDING_NUDGE, - )), + Ok(profile) => render_core_section(profile), Err(reason) => { tracing::warn!( target: "engram::core", diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 87aaac7cbe..31da7c4bf3 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -86,30 +86,6 @@ fn current_working_directory() -> Result { Ok(cwd.to_string_lossy().into_owned()) } -/// Publish a kind:20001 presence update event via the WebSocket connection. -/// -/// Ephemeral kinds (20000-29999) are rejected by the HTTP bridge, so presence -/// updates must be routed through the WS path. -/// -/// Content is a bare status string (`"online"`, `"away"`, `"offline"`) matching -/// the desktop client's format. The relay stores this in Redis and synthesizes -/// it back on presence queries. -async fn publish_presence( - publisher: &relay::RelayEventPublisher, - keys: &nostr::Keys, - status: &str, -) -> Result<(), relay::RelayError> { - use buzz_core::kind::KIND_PRESENCE_UPDATE; - use nostr::{EventBuilder, Kind}; - - let event = EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), status) - .tags([]) - .sign_with_keys(keys) - .map_err(|e| relay::RelayError::Http(format!("presence sign error: {e}")))?; - publisher.publish_event(event).await?; - Ok(()) -} - fn emit_runtime_lifecycle( observer: Option<&observer::ObserverHandle>, start_nonce: &str, @@ -648,6 +624,69 @@ fn spawn_relay_observer_publisher( }) } +fn spawn_broker_observer_publisher( + observer: observer::ObserverHandle, + broker: runtime_transport::BrokerActions, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut rx = observer.subscribe(); + let snapshot = observer.snapshot(); + let mut queue = ObserverPublishQueue::default(); + let max_snapshot_seq = snapshot.iter().map(|event| event.seq).max().unwrap_or(0); + for event in snapshot { + queue.ingest(event); + } + + let mut publish_tick = tokio::time::interval_at( + tokio::time::Instant::now() + OBSERVER_PUBLISH_TICK, + OBSERVER_PUBLISH_TICK, + ); + publish_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut closed = false; + loop { + tokio::select! { + result = rx.recv(), if !closed => match result { + Ok(event) if event.seq > max_snapshot_seq => queue.ingest(event), + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + tracing::warn!(dropped = count, "broker observer publisher lagged"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => closed = true, + }, + _ = publish_tick.tick() => { + if let Some(mut event) = queue.next_frame() { + fit_observer_event_to_budget(&mut event); + match serde_json::to_string(&event) { + Ok(payload) => { + let frame = buzz_sdk::broker::ObserverFrame { + kind: event.kind.clone(), + payload, + }; + match broker.observer_emit(vec![frame]).await { + Ok(receipt) if receipt.accepted == 1 => {} + Ok(receipt) => tracing::warn!( + accepted = receipt.accepted, + "broker did not accept the observer frame" + ), + Err(error) => tracing::warn!( + "broker observer frame dropped: {error}" + ), + } + } + Err(error) => tracing::warn!( + "failed to serialize broker observer frame: {error}" + ), + } + } + if closed && queue.is_empty() { + break; + } + } + } + } + }) +} + async fn run_relay_observer_publisher( snapshot: Vec, mut rx: tokio::sync::broadcast::Receiver, @@ -2071,8 +2110,8 @@ async fn tokio_main() -> Result<()> { ), } - let presence_publisher = relay.event_publisher(); - let presence_keys = config.keys.clone(); + let signal_publisher = relay.signal_publisher(config.keys.clone()); + let broker_actions = relay.broker_actions(); // Priority: BUZZ_AUTH_TAG (NIP-OA attestation) → --agent-owner flag. let startup_owner: Option = match config.agent_mode { @@ -2108,8 +2147,14 @@ async fn tokio_main() -> Result<()> { let mut relay_observer_control_rx = None; let mut relay_observer_publisher_task = None; let mut relay_observer_publisher = None; + let mut broker_observer_publisher = None; if config.relay_observer { - if let (Some(observer), Some(owner_pubkey_hex)) = + if config.agent_mode == AgentMode::Broker { + if let (Some(observer), Some(actions)) = (observer.clone(), broker_actions.clone()) { + broker_observer_publisher = Some((observer, actions)); + tracing::info!("broker observer enabled"); + } + } else if let (Some(observer), Some(owner_pubkey_hex)) = (observer.clone(), owner_cache.pubkey.clone()) { match PublicKey::from_hex(&owner_pubkey_hex) { @@ -2212,6 +2257,9 @@ async fn tokio_main() -> Result<()> { owner, )); } + if let Some((observer, actions)) = broker_observer_publisher.take() { + relay_observer_publisher_task = Some(spawn_broker_observer_publisher(observer, actions)); + } let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default(); let dedup_mode = config.dedup_mode; @@ -2222,7 +2270,10 @@ async fn tokio_main() -> Result<()> { // connected. Publishing after channel subscriptions gives desktop callers // a durable readiness boundary before they send a startup mention. if config.presence_enabled { - match publish_presence(&presence_publisher, &presence_keys, "online").await { + match signal_publisher + .presence_set(buzz_core::presence::PresenceStatus::Online) + .await + { Ok(_) => tracing::info!("presence set to online"), Err(e) => tracing::warn!("failed to set initial presence: {e}"), } @@ -2265,6 +2316,7 @@ async fn tokio_main() -> Result<()> { heartbeat_prompt: config.heartbeat_prompt.clone(), cwd, relay_features_enabled: config.agent_mode == AgentMode::Local, + broker_actions, rest_client: relay.rest_client(), channel_info, context_message_limit: config.context_message_limit, @@ -3163,10 +3215,12 @@ async fn tokio_main() -> Result<()> { if let Some(h) = presence_task.take() { h.abort(); } - let pp = presence_publisher.clone(); - let pk = presence_keys.clone(); + let signals = signal_publisher.clone(); presence_task = Some(tokio::spawn(async move { - if let Err(e) = publish_presence(&pp, &pk, "online").await { + if let Err(e) = signals + .presence_set(buzz_core::presence::PresenceStatus::Online) + .await + { tracing::warn!("presence heartbeat failed: {e}"); } })); @@ -3179,19 +3233,28 @@ async fn tokio_main() -> Result<()> { } } => { let _ = result_rx; - // Use try_publish (non-blocking) for typing indicators — - // they're ephemeral and must not block the main loop during - // relay reconnection (#35). + // Signals are ephemeral. Publish each on a detached task so + // a slow relay or broker never stalls the main event loop. for (&ch, thread_tags) in &typing_channels { - if let Ok(event) = relay.build_typing_event( - ch, - thread_tags.root_event_id.as_deref(), - thread_tags.parent_event_id.as_deref(), - ) { - if let Err(e) = relay.try_publish_event(event) { - tracing::debug!("typing indicator dropped for {ch}: {e}"); + let signals = signal_publisher.clone(); + let root = thread_tags.root_event_id.clone(); + let parent = thread_tags.parent_event_id.clone(); + tokio::spawn(async move { + match tokio::time::timeout( + Duration::from_secs(2), + signals.typing_set(ch, root, parent), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => tracing::debug!( + "typing indicator dropped for {ch}: {e}" + ), + Err(_) => tracing::debug!( + "typing indicator timed out for {ch}" + ), } - } + }); } None } @@ -3572,7 +3635,7 @@ async fn tokio_main() -> Result<()> { if config.presence_enabled { match tokio::time::timeout( Duration::from_secs(2), - publish_presence(&presence_publisher, &presence_keys, "offline"), + signal_publisher.presence_set(buzz_core::presence::PresenceStatus::Offline), ) .await { @@ -5139,6 +5202,18 @@ fn build_mcp_servers(config: &Config) -> Vec { env: { let mut env = match config.agent_mode { AgentMode::Local => vec![ + EnvVar { + name: "BUZZ_AGENT_MODE".into(), + value: "local".into(), + }, + EnvVar { + name: "BUZZ_BROKER_URL".into(), + value: String::new(), + }, + EnvVar { + name: "BUZZ_BROKER_CREDENTIAL".into(), + value: String::new(), + }, EnvVar { name: "BUZZ_RELAY_URL".into(), value: config.relay_url.clone(), diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 718d1f876d..bc2560a0e4 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -42,6 +42,7 @@ use crate::queue::{ PromptProfile, PromptProfileLookup, ThreadTags, }; use crate::relay::{ChannelInfo, RestClient}; +use crate::runtime_transport::BrokerActions; /// Window within which agent activity before a hard-cap death qualifies /// the turn as "recently active" (eligible for requeue instead of dead-letter). @@ -751,6 +752,9 @@ pub struct PromptContext { /// 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, + /// Broker-backed features that replace direct relay operations in keyless + /// mode. `None` for local mode. + pub broker_actions: Option, /// REST client for pre-prompt context fetches (thread/DM history). pub rest_client: RestClient, /// Shared channel metadata for startup-known and dynamically joined channels. @@ -1970,7 +1974,10 @@ pub async fn run_prompt_task( })); let liveness = run_turn_liveness( agent.acp.observer_handle(), + ctx.broker_actions.clone(), agent.acp.observer_agent_index(), + observer_channel_id, + turn_id.clone(), observer::context_for_turn( observer_channel_id, None, @@ -2047,19 +2054,38 @@ pub async fn run_prompt_task( // // Operator opt-out: `--no-memory` / `BUZZ_ACP_NO_MEMORY` skips the fetch. if ctx.memory_enabled { - if let (PromptSource::Channel(cid), Some(owner_pk)) = - (&source, ctx.agent_owner_pubkey.as_ref()) - { + if let PromptSource::Channel(cid) = &source { let is_new_channel_session = !agent.state.sessions.contains_key(cid); if is_new_channel_session && !agent.state.core_sections.contains_key(cid) { // Bounded — we'd rather start the session with no core hint // than block session creation on a stalled relay. const CORE_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3); - let fetch = crate::engram_fetch::build_core_section( - &ctx.rest_client, - &ctx.agent_keys, - owner_pk, - ); + let fetch = async { + if let Some(broker) = ctx.broker_actions.as_ref() { + match broker + .storage_get(buzz_core::engram::CORE_SLUG.to_string()) + .await + { + Ok(record) => crate::engram_fetch::render_core_section(record.value), + Err(error) => { + tracing::warn!( + target: "engram::core", + "broker core fetch failed: {error} — emitting no section" + ); + None + } + } + } else if let Some(owner_pk) = ctx.agent_owner_pubkey.as_ref() { + crate::engram_fetch::build_core_section( + &ctx.rest_client, + &ctx.agent_keys, + owner_pk, + ) + .await + } else { + None + } + }; let section = match tokio::time::timeout(CORE_FETCH_TIMEOUT, fetch).await { Ok(s) => s, Err(_) => { @@ -4430,20 +4456,25 @@ impl Drop for ReactionGuard { // by `run_prompt_task` after session resolution — so pings emitted for the // remainder of the turn carry the real session, matching every other // observer frame for this turn instead of a permanent `None`. +#[allow(clippy::too_many_arguments)] async fn run_turn_liveness( observer: Option, + broker: Option, agent_index: Option, + channel_id: Option, + turn_id: String, mut context: observer::ObserverContext, interval: Duration, state: Arc>, ) { - let Some(observer) = observer else { + if observer.is_none() && broker.is_none() { return std::future::pending::<()>().await; - }; + } if interval.is_zero() { return std::future::pending::<()>().await; } let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // The first tick completes immediately; skip it so the first liveness ping // fires one interval after the turn starts, not at t=0 (turn_started already // marks t=0). @@ -4453,21 +4484,31 @@ async fn run_turn_liveness( // Nothing awaitable between the lock and the emit: `LivenessGuard::drop` // takes this same lock before its `abort()`, so the guard can only ever // observe this tick fully emitted or not yet started — never mid-emit. - let guard = match state.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - }; - if guard.closed { - return; + { + let guard = match state.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + if guard.closed { + return; + } + context.session_id = guard.session_id.clone(); + if broker.is_none() { + if let Some(observer) = observer.as_ref() { + observer.emit( + "turn_liveness", + agent_index, + &context, + serde_json::json!({}), + ); + } + } + } + if let (Some(broker), Some(channel_id)) = (broker.as_ref(), channel_id) { + if let Err(error) = broker.liveness_ping(channel_id, turn_id.clone()).await { + tracing::debug!(%channel_id, "broker liveness ping dropped: {error}"); + } } - context.session_id = guard.session_id.clone(); - observer.emit( - "turn_liveness", - agent_index, - &context, - serde_json::json!({}), - ); - drop(guard); } } @@ -7432,7 +7473,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let _liveness_guard = LivenessGuard::new( tokio::spawn(run_turn_liveness( Some(observer.clone()), + None, Some(0), + None, + "turn".into(), context, Duration::from_secs(10), Arc::clone(&state), @@ -7482,7 +7526,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let guard = LivenessGuard::new( tokio::spawn(run_turn_liveness( Some(observer.clone()), + None, Some(0), + None, + "turn".into(), context, Duration::from_secs(10), Arc::clone(&state), @@ -7534,7 +7581,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let guard = LivenessGuard::new( tokio::spawn(run_turn_liveness( Some(observer.clone()), + None, Some(0), + None, + "turn".into(), context, Duration::from_secs(10), Arc::clone(&state), @@ -7576,7 +7626,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let context = observer::context_for(None, None, Some("t-1".into())); let liveness = run_turn_liveness( Some(observer.clone()), + None, Some(0), + None, + "turn".into(), context, Duration::ZERO, open_liveness_state(), @@ -7600,6 +7653,9 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let liveness = run_turn_liveness( None, None, + None, + None, + "turn".into(), context, Duration::from_secs(10), open_liveness_state(), @@ -7638,7 +7694,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" })); let liveness = run_turn_liveness( Some(observer.clone()), + None, Some(0), + None, + "turn".into(), context, Duration::from_secs(10), state, @@ -8262,6 +8321,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" base_prompt: None, cwd: ".".to_string(), relay_features_enabled: true, + broker_actions: None, 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 1391be78ae..49e5e9491b 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -118,7 +118,6 @@ use std::time::Instant; use buzz_core::kind::{ KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, - KIND_TYPING_INDICATOR, }; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, Kind, RelayUrl, Tag}; @@ -658,8 +657,18 @@ 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). + /// Best-effort publish for ephemeral signals. Drops immediately when the + /// relay command queue is unavailable or full. + pub fn try_publish_event(&self, event: Event) -> Result<(), RelayError> { + self.cmd_tx + .try_send(RelayCommand::PublishEvent { + event: Box::new(event), + }) + .map_err(|_| RelayError::ConnectionClosed) + } + + /// A publisher with no relay behind it. Retained as a fail-closed sentinel + /// for any relay-only path accidentally reached in broker mode. pub(crate) fn disabled() -> Self { let (cmd_tx, cmd_rx) = mpsc::channel(1); drop(cmd_rx); @@ -924,48 +933,6 @@ impl HarnessRelay { .map_err(|_| RelayError::ConnectionClosed) } - /// Fire-and-forget publish — uses `try_send` so it never blocks the caller. - /// - /// Suitable for ephemeral commands like typing indicators where dropping - /// the event on a full command channel is acceptable. - pub fn try_publish_event(&self, event: Event) -> Result<(), RelayError> { - self.cmd_tx - .try_send(RelayCommand::PublishEvent { - event: Box::new(event), - }) - .map_err(|_| RelayError::ConnectionClosed) - } - - /// Build a typing indicator event (kind:20002) for a channel. - pub fn build_typing_event( - &self, - channel_id: Uuid, - root_event_id: Option<&str>, - parent_event_id: Option<&str>, - ) -> Result { - let h_tag = Tag::parse(["h", &channel_id.to_string()]) - .map_err(|e| RelayError::AuthFailed(e.to_string()))?; - let mut tags = vec![h_tag]; - if let Some(parent) = parent_event_id { - if let Some(root) = root_event_id { - if root != parent { - tags.push( - Tag::parse(["e", root, "", "root"]) - .map_err(|e| RelayError::AuthFailed(e.to_string()))?, - ); - } - } - tags.push( - Tag::parse(["e", parent, "", "reply"]) - .map_err(|e| RelayError::AuthFailed(e.to_string()))?, - ); - } - let event = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") - .tags(tags) - .sign_with_keys(&self.keys)?; - Ok(event) - } - /// Pins the floor `since` for membership notification replay. /// /// Call once after `connect()` with the Unix timestamp captured just before @@ -5970,10 +5937,13 @@ mod tests { ); // Typing indicator while gated: still dropped, not parked. - let typing = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") - .tags([Tag::parse(["h", &Uuid::new_v4().to_string()]).unwrap()]) - .sign_with_keys(&keys) - .expect("sign typing indicator"); + let typing = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_TYPING_INDICATOR as u16), + "", + ) + .tags([Tag::parse(["h", &Uuid::new_v4().to_string()]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign typing indicator"); let ok = execute_connected_command( &mut client, &mut state, diff --git a/crates/buzz-acp/src/runtime_transport.rs b/crates/buzz-acp/src/runtime_transport.rs index 95cc0fcc64..d89585b156 100644 --- a/crates/buzz-acp/src/runtime_transport.rs +++ b/crates/buzz-acp/src/runtime_transport.rs @@ -1,8 +1,8 @@ //! 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. +//! mode polls `channel.read` and routes keyless storage and live signals back +//! through the broker. Capabilities with no broker action remain relay-only. use std::collections::{HashMap, HashSet, VecDeque}; use std::time::Duration; @@ -10,7 +10,9 @@ use std::time::Duration; use buzz_broker_client::HttpBrokerClient; use buzz_sdk::broker::{ ActionArgs, ActionOutcome, BrokerClientExt, BrokerErrorCode, BrokerRequest, BrokerResult, - ChannelReadArgs, StorageAddressArgs, + ChannelReadArgs, EventPublished, LivenessPingArgs, ObserverEmitArgs, ObserverFrame, + ObserverReceipt, PresenceSetArgs, StorageAddressArgs, StorageGetArgs, StorageRecord, + TypingSetArgs, }; use nostr::{Event, Keys}; use tokio::time::MissedTickBehavior; @@ -26,6 +28,77 @@ pub enum RuntimeTransport { Broker(BrokerRuntime), } +#[derive(Clone)] +pub enum RuntimeSignalPublisher { + Local { + publisher: RelayEventPublisher, + keys: Keys, + }, + Broker(BrokerActions), +} + +impl RuntimeSignalPublisher { + pub async fn presence_set( + &self, + status: buzz_core::presence::PresenceStatus, + ) -> Result<(), RelayError> { + match self { + Self::Local { publisher, keys } => { + use buzz_core::kind::KIND_PRESENCE_UPDATE; + use nostr::{EventBuilder, Kind}; + + let event = EventBuilder::new( + Kind::Custom(KIND_PRESENCE_UPDATE as u16), + status.to_string(), + ) + .tags([]) + .sign_with_keys(keys) + .map_err(|error| RelayError::Http(format!("presence sign error: {error}")))?; + publisher.publish_event(event).await + } + Self::Broker(actions) => actions.presence_set(status).await.map(|_| ()), + } + } + + pub async fn typing_set( + &self, + channel_id: Uuid, + root_event_id: Option, + parent_event_id: Option, + ) -> Result<(), RelayError> { + match self { + Self::Local { publisher, keys } => { + use buzz_core::kind::KIND_TYPING_INDICATOR; + use nostr::{EventBuilder, Kind, Tag}; + + let h_tag = Tag::parse(["h", &channel_id.to_string()]) + .map_err(|error| RelayError::Http(error.to_string()))?; + let mut tags = vec![h_tag]; + if let Some(parent) = parent_event_id.as_deref() { + if let Some(root) = root_event_id.as_deref() { + if root != parent { + tags.push( + Tag::parse(["e", root, "", "root"]) + .map_err(|error| RelayError::Http(error.to_string()))?, + ); + } + } + tags.push( + Tag::parse(["e", parent, "", "reply"]) + .map_err(|error| RelayError::Http(error.to_string()))?, + ); + } + let event = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") + .tags(tags) + .sign_with_keys(keys) + .map_err(|error| RelayError::Http(format!("typing sign error: {error}")))?; + publisher.try_publish_event(event) + } + Self::Broker(actions) => actions.typing_set(channel_id).await.map(|_| ()), + } + } +} + pub struct BrokerRuntime { client: HttpBrokerClient, channel_ids: Vec, @@ -39,6 +112,102 @@ pub struct BrokerRuntime { terminal_error: Option, } +/// Cloneable handle for broker-backed capabilities used outside the polling +/// loop (memory, presence, typing, observer telemetry, and turn liveness). +#[derive(Clone)] +pub struct BrokerActions { + client: HttpBrokerClient, +} + +impl BrokerActions { + async fn run(&self, args: ActionArgs) -> Result { + execute(&self.client, args) + .await + .map_err(BrokerActionError::relay_error) + } + + pub async fn storage_get(&self, slug: String) -> Result { + match self + .run(ActionArgs::StorageGet(StorageGetArgs { slug })) + .await? + { + ActionOutcome::StorageGet(record) => Ok(record), + _ => Err(wrong_outcome("storage.get")), + } + } + + pub async fn presence_set( + &self, + status: buzz_core::presence::PresenceStatus, + ) -> Result { + match self + .run(ActionArgs::PresenceSet(PresenceSetArgs { status })) + .await? + { + ActionOutcome::PresenceSet(published) => Ok(published), + _ => Err(wrong_outcome("presence.set")), + } + } + + pub async fn typing_set(&self, channel_id: Uuid) -> Result { + match self + .run(ActionArgs::TypingSet(TypingSetArgs { + channel_id: channel_id.to_string(), + })) + .await? + { + ActionOutcome::TypingSet(published) => Ok(published), + _ => Err(wrong_outcome("typing.set")), + } + } + + pub async fn observer_emit( + &self, + frames: Vec, + ) -> Result { + match self + .run(ActionArgs::ObserverEmit(ObserverEmitArgs { frames })) + .await? + { + ActionOutcome::ObserverEmit(receipt) => Ok(receipt), + _ => Err(wrong_outcome("observer.emit")), + } + } + + pub async fn liveness_ping( + &self, + channel_id: Uuid, + turn_id: String, + ) -> Result { + match self + .run(ActionArgs::LivenessPing(LivenessPingArgs { + channel_id: channel_id.to_string(), + turn_id, + })) + .await? + { + ActionOutcome::LivenessPing(published) => Ok(published), + _ => Err(wrong_outcome("liveness.ping")), + } + } +} + +fn wrong_outcome(action: &str) -> RelayError { + RelayError::Http(format!("broker returned the wrong outcome for {action}")) +} + +fn advance_cursor( + cursors: &mut HashMap, + channel_id: Uuid, + next_cursor: Option, +) { + if let Some(cursor) = next_cursor { + cursors.insert(channel_id, cursor); + } else { + cursors.remove(&channel_id); + } +} + struct BrokerActionError { detail: String, code: Option, @@ -64,7 +233,8 @@ impl BrokerRuntime { poll_interval: Duration, placeholder_keys: Keys, ) -> Result<(Self, String), RelayError> { - let client = HttpBrokerClient::new(base_url, credential); + let client = HttpBrokerClient::new(base_url, credential) + .map_err(|error| RelayError::Http(format!("broker config: {error}")))?; let outcome = execute( &client, ActionArgs::StorageAddress(StorageAddressArgs { @@ -152,9 +322,11 @@ impl BrokerRuntime { continue; } }; - if let Some(cursor) = page.next_cursor { - self.cursors.insert(channel_id, cursor); - } + // `nextCursor` is a pagination continuation, not a durable + // tail watermark. Return to the host's current default window + // after draining so newly-arrived events remain visible; the + // rotating dedup sets absorb the replay. + advance_cursor(&mut self.cursors, channel_id, page.next_cursor); for message in page.messages { if let Err(error) = message.verify() { tracing::warn!(%channel_id, "broker returned an unverifiable event: {error}"); @@ -353,6 +525,27 @@ impl RuntimeTransport { } } + pub fn broker_actions(&self) -> Option { + match self { + Self::Local(_) => None, + Self::Broker(runtime) => Some(BrokerActions { + client: runtime.client.clone(), + }), + } + } + + pub fn signal_publisher(&self, keys: Keys) -> RuntimeSignalPublisher { + match self { + Self::Local(relay) => RuntimeSignalPublisher::Local { + publisher: relay.event_publisher(), + keys, + }, + Self::Broker(runtime) => RuntimeSignalPublisher::Broker(BrokerActions { + client: runtime.client.clone(), + }), + } + } + pub fn rest_client(&self) -> RestClient { match self { Self::Local(relay) => relay.rest_client(), @@ -365,32 +558,120 @@ impl RuntimeTransport { } } - 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; } } } + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::{Body, Bytes}; + use axum::http::StatusCode; + use axum::response::Response; + use axum::routing::post; + use axum::Router; + use std::sync::{Arc, Mutex}; + use tokio::net::TcpListener; + + #[test] + fn terminal_page_returns_polling_to_the_default_window() { + let channel_id = Uuid::new_v4(); + let mut cursors = HashMap::new(); + advance_cursor(&mut cursors, channel_id, Some("continuation".into())); + assert_eq!( + cursors.get(&channel_id).map(String::as_str), + Some("continuation") + ); + + advance_cursor(&mut cursors, channel_id, None); + assert!(!cursors.contains_key(&channel_id)); + } + + #[tokio::test] + async fn broker_actions_route_storage_and_live_signals() { + let seen = Arc::new(Mutex::new(Vec::::new())); + let app = Router::new().route( + "/v1/action", + post({ + let seen = Arc::clone(&seen); + move |body: Bytes| { + let seen = Arc::clone(&seen); + async move { + let request: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let request_id = request["requestId"].as_str().unwrap(); + let action = request["action"].as_str().unwrap(); + seen.lock().unwrap().push(action.to_string()); + let outcome = match action { + "storage.get" => serde_json::json!({ "value": "core" }), + "observer.emit" => serde_json::json!({ "accepted": 1 }), + _ => serde_json::json!({ + "eventId": "cacf5f811cc8ef3f4af3f92cc222f92a86cdf6a26728a144c8e63b74ab6db359", + "kind": 20001, + "createdAt": 1_700_000_000u64, + }), + }; + let response = serde_json::json!({ + "type": "broker_result", + "protocolVersion": 1, + "requestId": request_id, + "status": "succeeded", + "action": action, + "outcome": outcome, + }); + Response::builder() + .status(StatusCode::OK) + .body(Body::from(response.to_string())) + .unwrap() + } + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let actions = BrokerActions { + client: HttpBrokerClient::new(format!("http://{address}"), "credential").unwrap(), + }; + let channel_id = Uuid::new_v4(); + + assert_eq!( + actions + .storage_get("core".into()) + .await + .unwrap() + .value + .as_deref(), + Some("core") + ); + actions + .presence_set(buzz_core::presence::PresenceStatus::Online) + .await + .unwrap(); + actions.typing_set(channel_id).await.unwrap(); + actions + .observer_emit(vec![ObserverFrame { + kind: "turn_started".into(), + payload: "{}".into(), + }]) + .await + .unwrap(); + actions + .liveness_ping(channel_id, "turn-1".into()) + .await + .unwrap(); + + assert_eq!( + *seen.lock().unwrap(), + [ + "storage.get", + "presence.set", + "typing.set", + "observer.emit", + "liveness.ping", + ] + ); + } +} diff --git a/crates/buzz-broker-client/src/lib.rs b/crates/buzz-broker-client/src/lib.rs index f0ca5c36a9..bf346d4a1c 100644 --- a/crates/buzz-broker-client/src/lib.rs +++ b/crates/buzz-broker-client/src/lib.rs @@ -11,11 +11,26 @@ use buzz_sdk::broker::{ }; const DEFAULT_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024; + +/// Invalid broker provisioning rejected before a bearer credential can leave +/// the process. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BrokerClientConfigError(String); + +impl std::fmt::Display for BrokerClientConfigError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for BrokerClientConfigError {} /// A broker host endpoint plus the agent's bearer credential. /// /// Holds no key and knows nothing of the relay: its whole authority is the /// opaque credential it replays on every request. +#[derive(Clone)] pub struct HttpBrokerClient { base_url: String, credential: String, @@ -25,8 +40,17 @@ pub struct HttpBrokerClient { impl HttpBrokerClient { /// 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()) + pub fn new( + base_url: impl Into, + credential: impl Into, + ) -> Result { + let http = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|error| { + BrokerClientConfigError(format!("failed to build broker HTTP client: {error}")) + })?; + Self::with_client(base_url, credential, http) } /// As [`Self::new`], reusing an existing reqwest client and its pool. @@ -34,7 +58,7 @@ impl HttpBrokerClient { base_url: impl Into, credential: impl Into, http: reqwest::Client, - ) -> Self { + ) -> Result { Self::with_client_timeout(base_url, credential, http, DEFAULT_REQUEST_TIMEOUT) } @@ -43,13 +67,47 @@ impl HttpBrokerClient { credential: impl Into, http: reqwest::Client, request_timeout: std::time::Duration, - ) -> Self { - Self { - base_url: base_url.into(), - credential: credential.into(), + ) -> Result { + let base_url = base_url.into(); + let parsed = reqwest::Url::parse(&base_url) + .map_err(|error| BrokerClientConfigError(format!("invalid broker URL: {error}")))?; + let loopback = parsed.host_str().is_some_and(|host| { + host.eq_ignore_ascii_case("localhost") + || host + .parse::() + .is_ok_and(|address| address.is_loopback()) + }); + if parsed.scheme() != "https" && !(parsed.scheme() == "http" && loopback) { + return Err(BrokerClientConfigError( + "broker URL must use HTTPS (plain HTTP is allowed only for loopback)".into(), + )); + } + if parsed.host_str().is_none() + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.query().is_some() + || parsed.fragment().is_some() + { + return Err(BrokerClientConfigError( + "broker URL must be an origin/path without credentials, query, or fragment".into(), + )); + } + + let credential = credential.into(); + if credential.trim().is_empty() { + return Err(BrokerClientConfigError( + "broker credential must not be empty".into(), + )); + } + reqwest::header::HeaderValue::from_str(&format!("Bearer {credential}")) + .map_err(|_| BrokerClientConfigError("broker credential is not header-safe".into()))?; + + Ok(Self { + base_url, + credential, http, request_timeout, - } + }) } } @@ -61,7 +119,7 @@ impl BrokerClient for HttpBrokerClient { self.base_url.trim_end_matches('/') ); let (status, body) = tokio::time::timeout(self.request_timeout, async { - let response = self + let mut response = self .http .post(url) .header(reqwest::header::CONTENT_TYPE, "application/json") @@ -75,10 +133,29 @@ impl BrokerClient for HttpBrokerClient { .map_err(|e| BrokerTransportError::Unreachable(e.to_string()))?; let status = response.status().as_u16(); - let body = response - .bytes() + if response + .content_length() + .is_some_and(|length| length > MAX_RESPONSE_BYTES as u64) + { + return Err(BrokerTransportError::NoEnvelope { + status, + detail: format!("response exceeds {MAX_RESPONSE_BYTES} bytes"), + }); + } + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() .await - .map_err(|e| BrokerTransportError::Unreachable(e.to_string()))?; + .map_err(|e| BrokerTransportError::Unreachable(e.to_string()))? + { + if body.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { + return Err(BrokerTransportError::NoEnvelope { + status, + detail: format!("response exceeds {MAX_RESPONSE_BYTES} bytes"), + }); + } + body.extend_from_slice(&chunk); + } Ok::<_, BrokerTransportError>((status, body)) }) .await @@ -175,7 +252,7 @@ mod tests { ); let (base, seen_auth) = spawn(StatusCode::OK, body).await; - let client = HttpBrokerClient::new(base, CRED); + let client = HttpBrokerClient::new(base, CRED).unwrap(); let validated = client.execute(&req).await.expect("a verdict"); assert!(matches!(validated.result(), BrokerResult::Succeeded { .. })); @@ -194,7 +271,7 @@ mod tests { ); let (base, _) = spawn(StatusCode::OK, body).await; - let client = HttpBrokerClient::new(base, CRED); + let client = HttpBrokerClient::new(base, CRED).unwrap(); let validated = client.execute(&req).await.expect("a verdict"); assert!(matches!(validated.result(), BrokerResult::Failed { .. })); @@ -205,7 +282,7 @@ mod tests { let req = post_request(); let (base, _) = spawn(StatusCode::BAD_GATEWAY, "upstream boom".to_string()).await; - let client = HttpBrokerClient::new(base, CRED); + let client = HttpBrokerClient::new(base, CRED).unwrap(); let err = client.execute(&req).await.expect_err("no envelope"); assert!(matches!( @@ -217,7 +294,7 @@ mod tests { #[tokio::test] async fn unreachable_host_is_a_transport_error() { let req = post_request(); - let client = HttpBrokerClient::new("http://127.0.0.1:1", CRED); + let client = HttpBrokerClient::new("http://127.0.0.1:1", CRED).unwrap(); let err = client.execute(&req).await.expect_err("unreachable"); assert!(matches!(err, BrokerTransportError::Unreachable(_))); @@ -238,11 +315,33 @@ mod tests { CRED, reqwest::Client::new(), std::time::Duration::from_millis(20), - ); + ) + .unwrap(); let err = client.execute(&post_request()).await.expect_err("timeout"); assert!( matches!(err, BrokerTransportError::Unreachable(message) if message.contains("timed out")) ); } + + #[test] + fn remote_plaintext_and_empty_credentials_are_rejected() { + assert!(HttpBrokerClient::new("http://broker.example", CRED).is_err()); + assert!(HttpBrokerClient::new("https://broker.example", " ").is_err()); + assert!(HttpBrokerClient::new("http://localhost:8787", CRED).is_ok()); + } + + #[tokio::test] + async fn oversized_response_is_rejected_before_parsing() { + let (base, _) = spawn(StatusCode::OK, "x".repeat(MAX_RESPONSE_BYTES + 1)).await; + let error = HttpBrokerClient::new(base, CRED) + .unwrap() + .execute(&post_request()) + .await + .expect_err("oversized response"); + assert!(matches!( + error, + BrokerTransportError::NoEnvelope { detail, .. } if detail.contains("exceeds") + )); + } } diff --git a/crates/buzz-cli/KEYLESS.md b/crates/buzz-cli/KEYLESS.md index 176c1f9962..0eca14ad09 100644 --- a/crates/buzz-cli/KEYLESS.md +++ b/crates/buzz-cli/KEYLESS.md @@ -12,14 +12,15 @@ The backend is selected by `--agent-mode` (env `BUZZ_AGENT_MODE`), default - `broker` — keyless; route every operation through a broker host. Broker mode currently covers: `messages get` (with `--mentions-only`, the wake -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. +path), `messages send` / reply, `reactions add`, `users set-profile/set-presence`, and +`mem address/get/set/hash`. The ACP runtime also routes core-memory reads, +presence, typing, observer telemetry, and turn liveness through the broker. +Commands without a matching broker capability still need 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 +configured channels through `channel.read`, uses broker-backed memory and live +signals, 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. @@ -57,7 +58,10 @@ buzz messages send --channel "$CH" --reply-to --content "on it" buzz messages get --channel "$CH" --mentions-only --limit 10 buzz reactions add --channel "$CH" --event --emoji "👍" buzz users set-profile --name "Ada" --about "a keyless agent" +buzz users set-presence online buzz mem address core +buzz mem get core +buzz mem set core "I am Ada." ``` Terminal 1 logs each action, the bearer credential, and the args it received. @@ -85,17 +89,14 @@ is a minimal reference for the wire shape. - Broker HTTP actions have a 30-second end-to-end transport timeout. A runtime credential rejected as `unauthenticated` is terminal instead of being polled forever. -- Runtime cursors and event deduplication are currently in memory only. The - first cursorless read after restart follows the host's default-window - semantics, which must be pinned at the real-host integration checkpoint to - prevent missed or replayed mentions. +- Runtime cursors and event deduplication are currently in memory only. Once a + pagination chain ends, polling returns to the host's default window and + bounded event-ID deduplication absorbs replayed messages. - This CLI slice exposes only the first `messages get --limit` window; broker cursor pagination and thread reads are not yet command-line options. The ACP runtime therefore prompts from the triggering event without relay-fetched thread history. -- `buzz mem address ` prints the broker's `{authorPubkey, kind, dTag}` - outcome as JSON. This temporary bridge proves secret-dependent address - derivation without giving the client a key or relay route. It does not yet - make `mem get/set/patch/rm` keyless; fetching, decrypting, encrypting, and - publishing memory records belong to the later end-to-end runtime storage - work. +- `buzz mem address ` prints `{authorPubkey, kind, dTag}`. `mem get/set` + use broker-side decryption/encryption and `mem hash` composes a read locally. + Listing, patching, and deletion remain unavailable because listing and + tombstone semantics are not represented by the current contract. diff --git a/crates/buzz-cli/examples/mock_broker.rs b/crates/buzz-cli/examples/mock_broker.rs index d0a79a2959..7e19cc0d71 100644 --- a/crates/buzz-cli/examples/mock_broker.rs +++ b/crates/buzz-cli/examples/mock_broker.rs @@ -73,6 +73,42 @@ async fn action(headers: axum::http::HeaderMap, body: Bytes) -> Response { "dTag": FAKE_EVENT_ID, }), ), + "storage.get" => succeeded( + request_id, + action, + serde_json::json!({ "value": "I am a keyless agent using broker-backed memory." }), + ), + "storage.put" => succeeded( + request_id, + action, + serde_json::json!({ "eventId": FAKE_EVENT_ID, "kind": 30174, "createdAt": 1_700_000_000u64 }), + ), + "presence.set" => succeeded( + request_id, + action, + serde_json::json!({ "eventId": FAKE_EVENT_ID, "kind": 20001, "createdAt": 1_700_000_000u64 }), + ), + "typing.set" => succeeded( + request_id, + action, + serde_json::json!({ "eventId": FAKE_EVENT_ID, "kind": 20002, "createdAt": 1_700_000_000u64 }), + ), + "observer.emit" => succeeded( + request_id, + action, + serde_json::json!({ + "accepted": request + .get("args") + .and_then(|args| args.get("frames")) + .and_then(|frames| frames.as_array()) + .map_or(0, Vec::len), + }), + ), + "liveness.ping" => succeeded( + request_id, + action, + serde_json::json!({ "eventId": FAKE_EVENT_ID, "kind": 24200, "createdAt": 1_700_000_000u64 }), + ), "channel.read" => succeeded(request_id, action, serde_json::json!({ "messages": [] })), other => serde_json::json!({ "type": "broker_result", diff --git a/crates/buzz-cli/src/backend.rs b/crates/buzz-cli/src/backend.rs index a975848c06..f395a71fc5 100644 --- a/crates/buzz-cli/src/backend.rs +++ b/crates/buzz-cli/src/backend.rs @@ -14,7 +14,8 @@ use uuid::Uuid; use buzz_sdk::broker::{ ActionArgs, ActionOutcome, BrokerClientExt, BrokerError, BrokerRequest, BrokerResult, ChannelReadArgs, EventPublished, MessagePage, MessagePostArgs, MessageReplyArgs, - ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, StorageAddressArgs, + PresenceSetArgs, ProfileSetArgs, PubkeyHex, ReactionAddArgs, StorageAddress, + StorageAddressArgs, StorageGetArgs, StoragePutArgs, StorageRecord, }; use buzz_sdk::ThreadRef; @@ -34,7 +35,10 @@ pub trait AgentBackend { async fn message_reply(&self, args: MessageReplyArgs) -> Result; async fn reaction_add(&self, args: ReactionAddArgs) -> Result; async fn profile_set(&self, args: ProfileSetArgs) -> Result; + async fn presence_set(&self, args: PresenceSetArgs) -> Result; async fn storage_address(&self, args: StorageAddressArgs) -> Result; + async fn storage_get(&self, args: StorageGetArgs) -> Result; + async fn storage_put(&self, args: StoragePutArgs) -> Result; } /// Keyless backend: no key, no relay route. Every operation is a broker request. @@ -102,12 +106,33 @@ impl AgentBackend for BrokerBackend { } } + async fn presence_set(&self, args: PresenceSetArgs) -> Result { + match self.run(ActionArgs::PresenceSet(args)).await? { + ActionOutcome::PresenceSet(published) => Ok(published), + _ => Err(unexpected_outcome("presence.set")), + } + } + async fn storage_address(&self, args: StorageAddressArgs) -> Result { match self.run(ActionArgs::StorageAddress(args)).await? { ActionOutcome::StorageAddress(address) => Ok(address), _ => Err(unexpected_outcome("storage.address")), } } + + async fn storage_get(&self, args: StorageGetArgs) -> Result { + match self.run(ActionArgs::StorageGet(args)).await? { + ActionOutcome::StorageGet(record) => Ok(record), + _ => Err(unexpected_outcome("storage.get")), + } + } + + async fn storage_put(&self, args: StoragePutArgs) -> Result { + match self.run(ActionArgs::StoragePut(args)).await? { + ActionOutcome::StoragePut(published) => Ok(published), + _ => Err(unexpected_outcome("storage.put")), + } + } } /// Local backend: holds the key and talks to the relay directly. Preserves @@ -235,6 +260,12 @@ impl AgentBackend for LocalBackend { self.publish(event).await } + async fn presence_set(&self, _args: PresenceSetArgs) -> Result { + Err(CliError::Other( + "local presence updates use the existing signed command path".into(), + )) + } + async fn storage_address(&self, args: StorageAddressArgs) -> Result { let owner = crate::commands::mem::resolve_owner(&self.client, None)?; let conversation_key = @@ -246,6 +277,18 @@ impl AgentBackend for LocalBackend { d_tag: buzz_core::engram::d_tag(&conversation_key, &args.slug), }) } + + async fn storage_get(&self, _args: StorageGetArgs) -> Result { + Err(CliError::Other( + "local storage reads use the existing encrypted-memory command path".into(), + )) + } + + async fn storage_put(&self, _args: StoragePutArgs) -> Result { + Err(CliError::Other( + "local storage writes use the existing encrypted-memory command path".into(), + )) + } } /// A runtime-selected backend. Implements [`AgentBackend`] by dispatch, so @@ -257,11 +300,13 @@ pub enum Backend { impl Backend { /// Keyless: talk to a broker `base_url` with `credential`. - #[must_use] - pub fn broker(base_url: impl Into, credential: impl Into) -> Self { - Self::Broker(BrokerBackend::new(HttpBrokerClient::new( - base_url, credential, - ))) + pub fn broker( + base_url: impl Into, + credential: impl Into, + ) -> Result { + let client = HttpBrokerClient::new(base_url, credential) + .map_err(|error| CliError::Usage(format!("invalid broker configuration: {error}")))?; + Ok(Self::Broker(BrokerBackend::new(client))) } /// Local: hold the key and talk to the relay. @@ -307,12 +352,33 @@ impl AgentBackend for Backend { } } + async fn presence_set(&self, args: PresenceSetArgs) -> Result { + match self { + Self::Local(b) => b.presence_set(args).await, + Self::Broker(b) => b.presence_set(args).await, + } + } + async fn storage_address(&self, args: StorageAddressArgs) -> Result { match self { Self::Local(b) => b.storage_address(args).await, Self::Broker(b) => b.storage_address(args).await, } } + + async fn storage_get(&self, args: StorageGetArgs) -> Result { + match self { + Self::Local(b) => b.storage_get(args).await, + Self::Broker(b) => b.storage_get(args).await, + } + } + + async fn storage_put(&self, args: StoragePutArgs) -> Result { + match self { + Self::Local(b) => b.storage_put(args).await, + Self::Broker(b) => b.storage_put(args).await, + } + } } fn broker_verdict(status: &str, error: &BrokerError) -> CliError { @@ -377,7 +443,7 @@ mod tests { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr: SocketAddr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); - BrokerBackend::new(HttpBrokerClient::new(format!("http://{addr}"), "cred")) + BrokerBackend::new(HttpBrokerClient::new(format!("http://{addr}"), "cred").unwrap()) } fn succeeded(rid: &str, action: &str, outcome: serde_json::Value) -> (StatusCode, String) { @@ -515,6 +581,41 @@ mod tests { assert_eq!(address.d_tag, EVENT_ID); } + #[tokio::test] + async fn broker_storage_get_returns_plaintext_or_absence() { + let backend = spawn_host(|rid, action| { + succeeded(rid, action, serde_json::json!({ "value": "remember me" })) + }) + .await; + let record = backend + .storage_get(StorageGetArgs { + slug: "core".into(), + }) + .await + .expect("record"); + assert_eq!(record.value.as_deref(), Some("remember me")); + } + + #[tokio::test] + async fn broker_storage_put_returns_the_published_event() { + let backend = spawn_host(|rid, action| { + succeeded( + rid, + action, + serde_json::json!({ "eventId": EVENT_ID, "kind": 30174, "createdAt": 1_700_000_000u64 }), + ) + }) + .await; + let published = backend + .storage_put(StoragePutArgs { + slug: "core".into(), + value: "remember me".into(), + }) + .await + .expect("published"); + assert_eq!(published.kind, 30174); + } + #[tokio::test] async fn broker_read_returns_a_page_of_signed_events() { // A real signed event, so the strict event reader accepts it. diff --git a/crates/buzz-cli/src/commands/mem.rs b/crates/buzz-cli/src/commands/mem.rs index 2e2124d4fa..d0b8852630 100644 --- a/crates/buzz-cli/src/commands/mem.rs +++ b/crates/buzz-cli/src/commands/mem.rs @@ -24,7 +24,7 @@ use buzz_core::engram::{ self, conversation_key, d_tag, normalize_slug, select_head, validate_and_decrypt, Body, Listing, }; use buzz_core::kind::KIND_AGENT_ENGRAM; -use buzz_sdk::broker::StorageAddressArgs; +use buzz_sdk::broker::{StorageAddressArgs, StorageGetArgs, StoragePutArgs}; use nostr::PublicKey; use crate::backend::{AgentBackend, Backend}; @@ -804,11 +804,16 @@ pub async fn dispatch(cmd: crate::MemCmd, client: &BuzzClient) -> Result<(), Cli } } -/// Keyless (broker) dispatch for encrypted-memory addressing only. -/// -/// The returned coordinates are intentionally surfaced as-is. They identify -/// the encrypted record but do not, by themselves, provide read, decrypt, -/// encrypt, or publish semantics; those remain a later runtime slice. +async fn broker_value(backend: &Backend, slug: String) -> Result { + backend + .storage_get(StorageGetArgs { slug: slug.clone() }) + .await? + .value + .ok_or_else(|| CliError::NotFound(format!("not found: {slug}"))) +} + +/// Keyless (broker) dispatch. The host owns encryption, signing, and relay +/// publication; the CLI sees only slug-addressed plaintext records. pub async fn dispatch_broker(cmd: crate::MemCmd, backend: &Backend) -> Result<(), CliError> { use crate::MemCmd; match cmd { @@ -823,9 +828,82 @@ pub async fn dispatch_broker(cmd: crate::MemCmd, backend: &Backend) -> Result<() ); Ok(()) } - _ => Err(CliError::Usage( - "keyless (broker) mode supports only 'mem address' in this group; encrypted-memory \ - reads and writes need the later runtime storage slice" + MemCmd::Get { slug, owner, agent } => { + if owner.is_some() || agent.is_some() { + return Err(CliError::Usage( + "--owner/--agent are unavailable in broker mode; identity is credential-bound" + .into(), + )); + } + let slug = normalize_slug(&slug) + .map_err(|e| CliError::Usage(format!("invalid slug: {e}")))?; + use std::io::Write; + std::io::stdout() + .write_all(broker_value(backend, slug).await?.as_bytes()) + .map_err(|e| CliError::Other(e.to_string())) + } + MemCmd::Hash { slug, owner, agent } => { + if owner.is_some() || agent.is_some() { + return Err(CliError::Usage( + "--owner/--agent are unavailable in broker mode; identity is credential-bound" + .into(), + )); + } + let slug = normalize_slug(&slug) + .map_err(|e| CliError::Usage(format!("invalid slug: {e}")))?; + println!("{}", sha256_hex(&broker_value(backend, slug).await?)); + Ok(()) + } + MemCmd::Set { + slug, + value, + owner, + allow_empty, + } => { + if owner.is_some() { + return Err(CliError::Usage( + "--owner is unavailable in broker mode; identity is credential-bound".into(), + )); + } + let slug = normalize_slug(&slug) + .map_err(|e| CliError::Usage(format!("invalid slug: {e}")))?; + let value = if value == "-" { + let mut input = String::new(); + std::io::stdin() + .take((engram::NIP44_PLAINTEXT_MAX + 1) as u64) + .read_to_string(&mut input) + .map_err(|e| CliError::Other(format!("stdin read failed: {e}")))?; + input + } else { + value + }; + if value.is_empty() { + let detail = if allow_empty { + "the broker storage contract does not support empty records" + } else { + "refusing to write an empty broker record" + }; + return Err(CliError::Usage(detail.into())); + } + let published = backend + .storage_put(StoragePutArgs { slug: slug.clone(), value }) + .await?; + eprintln!( + "wrote {slug} (event {}, created_at {})", + published.event_id, published.created_at + ); + Ok(()) + } + MemCmd::Ls { .. } => Err(CliError::Usage( + "'mem ls' is unavailable in broker mode because the contract has no storage listing action" + .into(), + )), + MemCmd::Patch { .. } => Err(CliError::Usage( + "'mem patch' is not yet wired to broker read-modify-write; use mem get/hash/set" + .into(), + )), + MemCmd::Rm { .. } => Err(CliError::Usage( + "'mem rm' is unavailable because the broker contract has no delete/tombstone action" .into(), )), } diff --git a/crates/buzz-cli/src/commands/users.rs b/crates/buzz-cli/src/commands/users.rs index 59ec6101c0..3484612197 100644 --- a/crates/buzz-cli/src/commands/users.rs +++ b/crates/buzz-cli/src/commands/users.rs @@ -424,7 +424,7 @@ pub async fn cmd_set_profile( Ok(()) } -/// Keyless (broker) dispatch for the users group: `set-profile` only. +/// Keyless (broker) dispatch for profile and presence updates. pub async fn dispatch_broker(cmd: crate::UsersCmd, backend: &Backend) -> Result<(), CliError> { use crate::UsersCmd; match cmd { @@ -460,8 +460,25 @@ pub async fn dispatch_broker(cmd: crate::UsersCmd, backend: &Backend) -> Result< ); Ok(()) } + UsersCmd::SetPresence { status } => { + let status = match status { + crate::PresenceStatus::Online => buzz_core::presence::PresenceStatus::Online, + crate::PresenceStatus::Away => buzz_core::presence::PresenceStatus::Away, + crate::PresenceStatus::Offline => buzz_core::presence::PresenceStatus::Offline, + }; + let published = backend + .presence_set(buzz_sdk::broker::PresenceSetArgs { status }) + .await?; + println!( + "{}", + serde_json::to_string(&published) + .map_err(|e| CliError::Other(format!("serialize outcome: {e}")))? + ); + Ok(()) + } _ => Err(CliError::Usage( - "keyless (broker) mode supports only 'users set-profile' in this group".into(), + "keyless (broker) mode supports only 'users set-profile' and 'users set-presence' in this group" + .into(), )), } } diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 05ca8ee4f2..7b7907558a 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -2171,7 +2171,7 @@ async fn run_broker(cli: Cli) -> Result<(), CliError> { .into(), ) })?; - let backend = backend::Backend::broker(base_url, credential); + let backend = backend::Backend::broker(base_url, credential)?; match cli.command { Cmd::Messages(sub) => commands::messages::dispatch_broker(sub, &backend).await, @@ -2179,9 +2179,8 @@ async fn run_broker(cli: Cli) -> Result<(), CliError> { Cmd::Users(sub) => commands::users::dispatch_broker(sub, &backend).await, Cmd::Mem(sub) => commands::mem::dispatch_broker(sub, &backend).await, _ => Err(CliError::Usage( - "keyless (broker) mode currently supports the wake→reply slice plus reactions and \ - profile, plus encrypted-memory addressing: 'messages get/send', 'reactions add', \ - 'users set-profile', and 'mem address'. Other commands need the local backend — \ + "keyless (broker) mode currently supports messages, reactions, profile/presence updates, and \ + broker-backed memory get/set/address/hash. Other commands need the local backend — \ unset --agent-mode or set it to 'local'" .into(), )),