keyless: add the client half of the broker (transport + backend)

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 <noreply@anthropic.com>
Signed-off-by: Joel Robotham <jrobotham@squareup.com>

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<MessagePage, CliError>;
+    async fn message_post(&self, args: MessagePostArgs) -> Result<EventPublished, CliError>;
+    async fn message_reply(&self, args: MessageReplyArgs) -> Result<EventPublished, CliError>;
+}
+
+/// 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<ActionOutcome, CliError> {
+        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<MessagePage, CliError> {
+        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<EventPublished, CliError> {
+        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<EventPublished, CliError> {
+        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<EventPublished, CliError> {
+        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<MessagePage, CliError> {
+        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::Value> =
+            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::<Event>(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<EventPublished, CliError> {
+        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<EventPublished, CliError> {
+        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<LocalBackend>),
+    Broker(BrokerBackend),
+}
+
+impl Backend {
+    /// Keyless: talk to a broker `base_url` with `credential`.
+    #[must_use]
+    pub fn broker(base_url: impl Into<String>, credential: impl Into<String>) -> 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<MessagePage, CliError> {
+        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<EventPublished, CliError> {
+        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<EventPublished, CliError> {
+        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<dyn Fn(&str, &str) -> (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: 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<Responder>, 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"));
+    }
+}

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<String>, credential: impl Into<String>) -> 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<String>,
+        credential: impl Into<String>,
+        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::<BrokerResponse>(&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<Mutex<Option<String>>>) {
+        let seen_auth = Arc::new(Mutex::new(None));
+        type S = (Arc<(StatusCode, String)>, Arc<Mutex<Option<String>>>);
+        let state: S = (Arc::new((status, body)), seen_auth.clone());
+
+        let app = Router::new()
+            .route(
+                BROKER_ACTION_PATH,
+                post(
+                    |State((canned, seen)): State<S>, 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(_)));
+    }
+}

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;