jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-08 10:21:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

keyless: run buzz-acp through the broker

Browse Source 
Promote the HTTP broker transport into a shared crate and add an explicit runtime transport seam to buzz-acp. Broker mode rejects local keys, derives its identity through storage.address, polls configured channels through channel.read, validates returned events, and provisions agent subprocesses with broker-only credentials.

Relay-only housekeeping and enrichment stay disabled until the frozen contract grows an agreed host-owned path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Signed-off-by: Joel Robotham <jrobotham@squareup.com>
This commit is contained in:
Joel Robotham
2026-08-27 13:09:21 +10:00
parent 4b8f21845d
commit 01240c16eb
16 changed files with 908 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
Cargo.lock generated
View File

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

2
Cargo.toml
View File

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

1
crates/buzz-acp/Cargo.toml
View File

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

44
crates/buzz-acp/README.md
View File

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

14
crates/buzz-acp/src/acp.rs
View File

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

254
crates/buzz-acp/src/config.rs
View File

@@ -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", &"<redacted>")
.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<String>,
#[arg(long, env = "BUZZ_BROKER_CREDENTIAL", hide_env_values = true)]
pub broker_credential: Option<String>,
/// 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<String>,
@@ -517,6 +564,8 @@ pub struct ChannelFilter {
#[derive(Debug)]
pub struct Config {
pub keys: Keys,
pub agent_mode: AgentMode,
pub broker: Option<BrokerConfig>,
pub relay_url: String,
pub agent_command: String,
pub agent_args: Vec<String>,
@@ -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<Self, ConfigError> {
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,

181
crates/buzz-acp/src/lib.rs
View File

@@ -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<nostr::Tag> = std::env::var("BUZZ_AUTH_TAG")
.ok()
let relay_auth_tag: Option<nostr::Tag> = (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<String> = resolve_agent_owner(&config);
let startup_owner: Option<String> = 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<McpServer> {
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

69
crates/buzz-acp/src/pool.rs
View File

@@ -561,7 +561,7 @@ const PROJECT_INFO_CACHE_TTL: std::time::Duration = std::time::Duration::from_se
pub struct ChannelInfoResolver {
cache: std::sync::Arc<std::sync::RwLock<std::collections::HashMap<Uuid, PromptChannelInfo>>>,
projects: std::sync::Arc<std::sync::RwLock<std::collections::HashMap<Uuid, CachedProjectInfo>>>,
rest_client: RestClient,
rest_client: Option<RestClient>,
}
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<Uuid, ChannelInfo>) -> 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<PromptChannelInfo>,
ctx: &PromptContext,
) -> Option<ConversationContext> {
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(),

8
crates/buzz-acp/src/relay.rs
View File

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

356
crates/buzz-acp/src/runtime_transport.rs Normal file
View File

@@ -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<Uuid>,
filters: HashMap<Uuid, ChannelFilter>,
cursors: HashMap<Uuid, String>,
pending: VecDeque<BuzzEvent>,
seen_current: HashSet<String>,
seen_previous: HashSet<String>,
poll: tokio::time::Interval,
placeholder_keys: Keys,
}
impl BrokerRuntime {
pub async fn connect(
base_url: String,
credential: String,
channel_ids: Vec<Uuid>,
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<BuzzEvent> {
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<ActionOutcome, RelayError> {
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<Uuid>,
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<tokio::sync::mpsc::Receiver<Event>> {
match self {
Self::Local(relay) => relay.take_observer_control_rx(),
Self::Broker(_) => None,
}
}
pub async fn discover_channels(&self) -> Result<HashMap<Uuid, ChannelInfo>, 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<u64>,
) -> 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<BuzzEvent> {
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<Event, RelayError> {
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;
}
}
}

17
crates/buzz-broker-client/Cargo.toml Normal file
View File

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

18
crates/buzz-cli/src/broker_client.rs → crates/buzz-broker-client/src/lib.rs
View File

@@ -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<String>, credential: impl Into<String>) -> 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]

1
crates/buzz-cli/Cargo.toml
View File

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

7
crates/buzz-cli/KEYLESS.md
View File

@@ -16,6 +16,13 @@ path), `messages send` / reply, `reactions add`, `users set-profile`, and
`mem address <slug>` 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

2
crates/buzz-cli/src/backend.rs
View File

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

1
crates/buzz-cli/src/lib.rs
View File

@@ -1,6 +1,5 @@
pub mod agent_management;
pub mod backend;
pub mod broker_client;
mod client;
mod commands;
mod error;