From 6de8baef0bb24e3dbc4fe58bb0fc4895ad5d4d5f Mon Sep 17 00:00:00 2001 From: Joel Robotham Date: Thu, 27 Aug 2026 09:37:52 +1000 Subject: [PATCH] keyless: add the client half of the broker (transport + backend) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Client-side [C1] of the agent broker (block/buzz#6790): lets the buzz CLI run without an agent nsec, routing reads and writes through a host so relay traffic looks like a normal Buzz client. - broker_client.rs: HttpBrokerClient implements the BrokerClient transport primitive the contract crate omits — POST /v1/action with an opaque bearer credential, parse an envelope regardless of HTTP status, never interpret a verdict. Correlation stays in execute(). - backend.rs: AgentBackend trait spoken in broker vocabulary, with two impls behind a Backend enum — BrokerBackend (keyless) and LocalBackend (nsec + relay, today's path). LocalBackend reuses the shared buzz_sdk::build_message builder, so there is no parallel message path. - lib.rs: register the two modules. Not yet wired into the command surface or provisioning. 8/8 unit tests pass (4 transport, 4 backend); clippy -D warnings clean. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Joel Robotham --- crates/buzz-cli/src/backend.rs | 401 +++++++++++++++++++++++++++ crates/buzz-cli/src/broker_client.rs | 212 ++++++++++++++ crates/buzz-cli/src/lib.rs | 2 + 3 files changed, 615 insertions(+) create mode 100644 crates/buzz-cli/src/backend.rs create mode 100644 crates/buzz-cli/src/broker_client.rs diff --git a/crates/buzz-cli/src/backend.rs b/crates/buzz-cli/src/backend.rs new file mode 100644 index 0000000000..8658eebfa6 --- /dev/null +++ b/crates/buzz-cli/src/backend.rs @@ -0,0 +1,401 @@ +//! The relay-touching operations an agent performs, over either a local key + +//! relay ([`LocalBackend`]) or a keyless broker host ([`BrokerBackend`]). +//! +//! Both implement [`AgentBackend`] and speak the broker's vocabulary, so the +//! command layer depends on the trait and never on which side holds the key. +//! Selection is by provisioning: a broker endpoint + credential picks the +//! keyless path; a local key picks the relay path. This is the "seam that +//! covers both" from the keyless plan, done as one abstraction rather than +//! scattered conditionals. + +use nostr::Event; +use uuid::Uuid; + +use buzz_sdk::broker::{ + ActionArgs, ActionOutcome, BrokerClientExt, BrokerError, BrokerRequest, BrokerResult, + ChannelReadArgs, EventPublished, MessagePage, MessagePostArgs, MessageReplyArgs, PubkeyHex, +}; +use buzz_sdk::ThreadRef; + +use crate::broker_client::HttpBrokerClient; +use crate::client::BuzzClient; +use crate::error::CliError; + +/// The operations an agent performs, in the broker's vocabulary. +/// +/// A closed set today — read a channel, post, reply — mirroring the contract's +/// first slice. Adding one is a change here *and* to the contract, deliberately. +#[allow(async_fn_in_trait)] // dispatched through the `Backend` enum, never `dyn`. +pub trait AgentBackend { + async fn channel_read(&self, args: ChannelReadArgs) -> Result; + async fn message_post(&self, args: MessagePostArgs) -> Result; + async fn message_reply(&self, args: MessageReplyArgs) -> Result; +} + +/// Keyless backend: no key, no relay route. Every operation is a broker request. +pub struct BrokerBackend { + client: HttpBrokerClient, +} + +impl BrokerBackend { + #[must_use] + pub fn new(client: HttpBrokerClient) -> Self { + Self { client } + } + + /// Freeze one action, send it, and unwrap the host's verdict to its outcome. + async fn run(&self, args: ActionArgs) -> Result { + let request = BrokerRequest::new(Uuid::new_v4().to_string(), args) + .and_then(BrokerRequest::prepare) + .map_err(|e| CliError::Other(format!("broker request: {e}")))?; + let validated = self + .client + .execute(&request) + .await + .map_err(|e| CliError::Other(format!("broker transport: {e}")))?; + match validated.into_envelope().result { + BrokerResult::Succeeded { outcome } => Ok(outcome), + BrokerResult::Failed { error } => Err(broker_verdict("failed", &error)), + BrokerResult::Indeterminate { error } => Err(broker_verdict("indeterminate", &error)), + } + } +} + +impl AgentBackend for BrokerBackend { + async fn channel_read(&self, args: ChannelReadArgs) -> Result { + match self.run(ActionArgs::ChannelRead(args)).await? { + ActionOutcome::ChannelRead(page) => Ok(page), + _ => Err(unexpected_outcome("channel.read")), + } + } + + async fn message_post(&self, args: MessagePostArgs) -> Result { + match self.run(ActionArgs::MessagePost(args)).await? { + ActionOutcome::MessagePost(published) => Ok(published), + _ => Err(unexpected_outcome("message.post")), + } + } + + async fn message_reply(&self, args: MessageReplyArgs) -> Result { + match self.run(ActionArgs::MessageReply(args)).await? { + ActionOutcome::MessageReply(published) => Ok(published), + _ => Err(unexpected_outcome("message.reply")), + } + } +} + +/// Local backend: holds the key and talks to the relay directly. Preserves +/// today's behavior by reusing the shared `buzz_sdk` builders the CLI already +/// signs and submits, so there is no parallel message-construction path. +pub struct LocalBackend { + client: BuzzClient, +} + +impl LocalBackend { + #[must_use] + pub fn new(client: BuzzClient) -> Self { + Self { client } + } + + /// Compute the outcome from the locally-signed event, then submit it. + async fn publish(&self, event: Event) -> Result { + let published = EventPublished { + event_id: event.id.to_hex(), + kind: u32::from(event.kind.as_u16()), + created_at: event.created_at.as_secs(), + }; + self.client.submit_event(event).await?; + Ok(published) + } +} + +impl AgentBackend for LocalBackend { + async fn channel_read(&self, args: ChannelReadArgs) -> Result { + let mut filter = serde_json::json!({ + "kinds": [9, 40002, 40008, 45001, 45003], + "#h": [args.channel_id], + "limit": args.effective_limit(), + }); + if args.mentions_only { + filter["#p"] = serde_json::json!([self.client.keys().public_key().to_hex()]); + } + if let Some(root) = &args.root_event_id { + filter["#e"] = serde_json::json!([root]); + } + + let raw = self.client.query(&filter).await?; + let values: Vec = + serde_json::from_str(&raw).map_err(|e| CliError::Other(format!("parse read: {e}")))?; + let messages = values + .into_iter() + .filter_map(|v| serde_json::from_value::(v).ok()) + .map(buzz_sdk::broker::BrokerMessage) + .collect(); + // Local paging differs from a host's; a cursor is a host concept, so a + // local read is a single window with no continuation. Fuller paging is + // deferred until the local path is retrofitted onto the trait. + Ok(MessagePage { + messages, + next_cursor: None, + }) + } + + async fn message_post(&self, args: MessagePostArgs) -> Result { + let channel = Uuid::parse_str(&args.channel_id) + .map_err(|e| CliError::Other(format!("channel id: {e}")))?; + let mentions: Vec<&str> = args.mentions.iter().map(PubkeyHex::as_str).collect(); + let builder = buzz_sdk::build_message(channel, &args.content, None, &mentions, false, &[]) + .map_err(|e| CliError::Other(format!("build_message: {e}")))?; + let event = self.client.sign_event(builder)?; + self.publish(event).await + } + + async fn message_reply(&self, args: MessageReplyArgs) -> Result { + let channel = Uuid::parse_str(&args.channel_id) + .map_err(|e| CliError::Other(format!("channel id: {e}")))?; + let parent = nostr::EventId::from_hex(&args.reply_to_event_id) + .map_err(|e| CliError::Other(format!("reply target: {e}")))?; + // Direct reply: root == parent. Nested-thread root derivation (the + // host's job under the broker) is deferred for the local path. + let thread_ref = ThreadRef { + root_event_id: parent, + parent_event_id: parent, + }; + let mentions: Vec<&str> = args.mentions.iter().map(PubkeyHex::as_str).collect(); + let builder = buzz_sdk::build_message( + channel, + &args.content, + Some(&thread_ref), + &mentions, + false, + &[], + ) + .map_err(|e| CliError::Other(format!("build_message: {e}")))?; + let event = self.client.sign_event(builder)?; + self.publish(event).await + } +} + +/// A runtime-selected backend. Implements [`AgentBackend`] by dispatch, so +/// commands hold one value and never branch on custody. +pub enum Backend { + Local(Box), + Broker(BrokerBackend), +} + +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, + ))) + } + + /// Local: hold the key and talk to the relay. + #[must_use] + pub fn local(client: BuzzClient) -> Self { + Self::Local(Box::new(LocalBackend::new(client))) + } +} + +impl AgentBackend for Backend { + async fn channel_read(&self, args: ChannelReadArgs) -> Result { + match self { + Self::Local(b) => b.channel_read(args).await, + Self::Broker(b) => b.channel_read(args).await, + } + } + + async fn message_post(&self, args: MessagePostArgs) -> Result { + match self { + Self::Local(b) => b.message_post(args).await, + Self::Broker(b) => b.message_post(args).await, + } + } + + async fn message_reply(&self, args: MessageReplyArgs) -> Result { + match self { + Self::Local(b) => b.message_reply(args).await, + Self::Broker(b) => b.message_reply(args).await, + } + } +} + +fn broker_verdict(status: &str, error: &BrokerError) -> CliError { + CliError::Other(format!( + "broker {status}: {} [{}]", + error.message, + error.code.as_str() + )) +} + +fn unexpected_outcome(action: &str) -> CliError { + CliError::Other(format!( + "broker returned an outcome that is not for {action}" + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::SocketAddr; + use std::sync::Arc; + + use axum::body::{Body, Bytes}; + use axum::extract::State; + use axum::http::StatusCode; + use axum::response::Response; + use axum::routing::post; + use axum::Router; + use nostr::{EventBuilder, Keys, Kind}; + use tokio::net::TcpListener; + + const CHANNEL: &str = "5df7dfa8-e919-43df-8efd-f1dcb8af7071"; + const EVENT_ID: &str = "cacf5f811cc8ef3f4af3f92cc222f92a86cdf6a26728a144c8e63b74ab6db359"; + + type Responder = Arc (StatusCode, String) + Send + Sync>; + + /// Spawn a broker host that echoes the request's `requestId`/`action` into + /// whatever `f` builds, so correlation always holds. + async fn spawn_host(f: F) -> BrokerBackend + where + F: Fn(&str, &str) -> (StatusCode, String) + Send + Sync + 'static, + { + let responder: Responder = Arc::new(f); + let app = Router::new() + .route( + "/v1/action", + post(|State(r): State, body: Bytes| async move { + let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let rid = v.get("requestId").and_then(|x| x.as_str()).unwrap_or(""); + let action = v.get("action").and_then(|x| x.as_str()).unwrap_or(""); + let (status, out) = r(rid, action); + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Body::from(out)) + .unwrap() + }), + ) + .with_state(responder); + + 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")) + } + + fn succeeded(rid: &str, action: &str, outcome: serde_json::Value) -> (StatusCode, String) { + let body = serde_json::json!({ + "type": "broker_result", + "protocolVersion": 1, + "requestId": rid, + "status": "succeeded", + "action": action, + "outcome": outcome, + }); + (StatusCode::OK, body.to_string()) + } + + #[tokio::test] + async fn broker_post_returns_the_published_event() { + let backend = spawn_host(|rid, action| { + succeeded( + rid, + action, + serde_json::json!({ "eventId": EVENT_ID, "kind": 9, "createdAt": 1_700_000_000u64 }), + ) + }) + .await; + + let published = backend + .message_post(MessagePostArgs { + channel_id: CHANNEL.into(), + content: "hello".into(), + mentions: Vec::new(), + }) + .await + .expect("published"); + + assert_eq!(published.event_id, EVENT_ID); + assert_eq!(published.kind, 9); + } + + #[tokio::test] + async fn broker_reply_returns_the_published_event() { + let backend = spawn_host(|rid, action| { + succeeded( + rid, + action, + serde_json::json!({ "eventId": EVENT_ID, "kind": 9, "createdAt": 1_700_000_000u64 }), + ) + }) + .await; + + let published = backend + .message_reply(MessageReplyArgs { + channel_id: CHANNEL.into(), + reply_to_event_id: EVENT_ID.into(), + content: "on it".into(), + mentions: Vec::new(), + }) + .await + .expect("published"); + + assert_eq!(published.event_id, EVENT_ID); + } + + #[tokio::test] + async fn broker_read_returns_a_page_of_signed_events() { + // A real signed event, so the strict event reader accepts it. + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "hi") + .sign_with_keys(&keys) + .unwrap(); + let event_json = serde_json::to_value(&event).unwrap(); + + let backend = spawn_host(move |rid, action| { + succeeded( + rid, + action, + serde_json::json!({ "messages": [event_json.clone()] }), + ) + }) + .await; + + let page = backend + .channel_read(ChannelReadArgs::channel(CHANNEL)) + .await + .expect("page"); + + assert_eq!(page.messages.len(), 1); + assert!(page.next_cursor.is_none()); + } + + #[tokio::test] + async fn broker_failure_is_surfaced_as_an_error() { + let backend = spawn_host(|rid, _action| { + let body = serde_json::json!({ + "type": "broker_result", + "protocolVersion": 1, + "requestId": rid, + "status": "failed", + "error": { "code": "unauthorized", "message": "not permitted" }, + }); + (StatusCode::OK, body.to_string()) + }) + .await; + + let err = backend + .message_post(MessagePostArgs { + channel_id: CHANNEL.into(), + content: "hello".into(), + mentions: Vec::new(), + }) + .await + .expect_err("a failure"); + + assert!(err.to_string().contains("unauthorized")); + } +} diff --git a/crates/buzz-cli/src/broker_client.rs b/crates/buzz-cli/src/broker_client.rs new file mode 100644 index 0000000000..3bf3f53236 --- /dev/null +++ b/crates/buzz-cli/src/broker_client.rs @@ -0,0 +1,212 @@ +//! HTTP transport for the agent broker — keyless client mode. +//! +//! 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. + +use buzz_sdk::broker::{ + BrokerClient, BrokerFuture, BrokerResponse, BrokerTransportError, Dispatch, PreparedRequest, + BROKER_ACTION_PATH, BROKER_CREDENTIAL_HEADER, +}; + +/// 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. +pub struct HttpBrokerClient { + base_url: String, + credential: String, + http: reqwest::Client, +} + +impl HttpBrokerClient { + /// A client posting to `base_url` (scheme + authority, no path) 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()) + } + + /// As [`Self::new`], reusing an existing reqwest client and its pool. + pub fn with_client( + base_url: impl Into, + credential: impl Into, + http: reqwest::Client, + ) -> Self { + Self { + base_url: base_url.into(), + credential: credential.into(), + http, + } + } +} + +impl BrokerClient for HttpBrokerClient { + fn send<'a>(&'a self, request: &'a PreparedRequest, _dispatch: Dispatch) -> BrokerFuture<'a> { + Box::pin(async move { + let url = format!( + "{}{BROKER_ACTION_PATH}", + self.base_url.trim_end_matches('/') + ); + let response = self + .http + .post(url) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .header( + BROKER_CREDENTIAL_HEADER, + format!("Bearer {}", self.credential), + ) + .body(request.body().to_vec()) + .send() + .await + .map_err(|e| BrokerTransportError::Unreachable(e.to_string()))?; + + let status = response.status().as_u16(); + let body = response + .bytes() + .await + .map_err(|e| BrokerTransportError::Unreachable(e.to_string()))?; + + // Parse an envelope whatever the status. Only its absence makes the + // status meaningful, and then only as operator detail. + serde_json::from_slice::(&body).map_err(|e| { + BrokerTransportError::NoEnvelope { + status, + detail: e.to_string(), + } + }) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::SocketAddr; + use std::sync::{Arc, Mutex}; + + use axum::body::{Body, Bytes}; + use axum::extract::State; + use axum::http::{HeaderMap, StatusCode}; + use axum::response::Response; + use axum::routing::post; + use axum::Router; + use buzz_sdk::broker::{ + ActionArgs, BrokerClientExt, BrokerRequest, BrokerResult, MessagePostArgs, + }; + use tokio::net::TcpListener; + + const CHANNEL: &str = "5df7dfa8-e919-43df-8efd-f1dcb8af7071"; + const CRED: &str = "test-cred"; + + /// Spawn a broker that answers `/v1/action` with a fixed `(status, body)` + /// and records the `Authorization` header it saw. + async fn spawn(status: StatusCode, body: String) -> (String, Arc>>) { + let seen_auth = Arc::new(Mutex::new(None)); + type S = (Arc<(StatusCode, String)>, Arc>>); + let state: S = (Arc::new((status, body)), seen_auth.clone()); + + let app = Router::new() + .route( + BROKER_ACTION_PATH, + post( + |State((canned, seen)): State, headers: HeaderMap, _body: Bytes| async move { + *seen.lock().unwrap() = headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .map(str::to_owned); + Response::builder() + .status(canned.0) + .header("content-type", "application/json") + .body(Body::from(canned.1.clone())) + .unwrap() + }, + ), + ) + .with_state(state); + + 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() }); + (format!("http://{addr}"), seen_auth) + } + + fn post_request() -> PreparedRequest { + let args = ActionArgs::MessagePost(MessagePostArgs { + channel_id: CHANNEL.to_string(), + content: "hi".to_string(), + mentions: Vec::new(), + }); + BrokerRequest::new("req-post-1", args) + .unwrap() + .prepare() + .unwrap() + } + + #[tokio::test] + async fn success_round_trips_and_sends_bearer_credential() { + let req = post_request(); + let body = format!( + r#"{{"type":"broker_result","protocolVersion":1,"requestId":"{}","status":"succeeded","action":"message.post","outcome":{{"eventId":"{}","kind":9,"createdAt":1700000000}}}}"#, + req.request_id(), + "a".repeat(64), + ); + let (base, seen_auth) = spawn(StatusCode::OK, body).await; + + let client = HttpBrokerClient::new(base, CRED); + let validated = client.execute(&req).await.expect("a verdict"); + + assert!(matches!(validated.result(), BrokerResult::Succeeded { .. })); + assert_eq!( + seen_auth.lock().unwrap().as_deref(), + Some("Bearer test-cred") + ); + } + + #[tokio::test] + async fn rejected_credential_is_a_verdict_not_a_transport_error() { + let req = post_request(); + let body = format!( + 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:?}"), + } + } + + #[tokio::test] + async fn non_envelope_response_is_a_transport_error() { + let req = post_request(); + let (base, _) = spawn(StatusCode::BAD_GATEWAY, "upstream boom".to_string()).await; + + let client = HttpBrokerClient::new(base, CRED); + let err = client.execute(&req).await.expect_err("no envelope"); + + assert!(matches!( + err, + BrokerTransportError::NoEnvelope { status: 502, .. } + )); + } + + #[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 err = client.execute(&req).await.expect_err("unreachable"); + + assert!(matches!(err, BrokerTransportError::Unreachable(_))); + } +} diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index d0155970fa..604dbd3a70 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1,4 +1,6 @@ pub mod agent_management; +pub mod backend; +pub mod broker_client; mod client; mod commands; mod error;