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

broker: verify inbound frames and deliver signed results

Browse Source 
Closes the loop from a signed relay event to a signed result frame. The
pipeline was already testable in isolation; this attaches the transport and
the host's real parts, so nothing above the ingress layer learns what Tauri
or the relay look like.

`ingress::verify_frame` derives every trusted field from the frame rather
than from anything a caller could assert. Owner, requester, and relay scope
are transport-derived, which is why the wire envelope has no field for them:
a payload that could name its own owner would be a payload that could pick
its own authority. Verification refuses duplicate routing tags — a frame with
two `agent` tags has two readings, and the one this host picks would not have
to match the one the relay routed on. Freshness is re-checked here rather
than trusted from the relay, because bounding replay of an old signed frame
is not a hop this host controls.

A frame that fails verification gets no answer at all. Emitting a signed
refusal at an unverified sender would make any host a signing oracle aimed at
a target of the sender's choosing. Frames that are simply not broker traffic —
the overwhelming majority, since ordinary telemetry arrives on this same
subscription — are ignored cheaply.

`host.rs` keeps the command a transport hand-off rather than an authority
hand-off: it takes a raw signed event and nothing else, so the renderer cannot
say who is asking or which owner to act as. The return value is deliberately
thin, since the requester's real answer travels in a signed frame; reporting
an outcome through IPC would make the reply path look like a return value it
is not. Results publish over the relay WebSocket because kind-24200 is
rejected on the HTTP bridge, and as the owner's own identity with no NIP-OA
auth tag — an auth tag is how a managed agent proves owner backing, and the
owner is not one.

On the TypeScript side the renderer forwards the *signed event*, never the
decrypted plaintext, so the host verifies and decrypts independently. The
routing check is shallow on purpose: it decides only whether a payload claims
to be a broker request, not whether it is valid, so an ill-formed request
still reaches the host and earns a structured refusal instead of silence. A
broker request also returns before the session journal, because a payload with
no seq, timestamp, or kind must not be filed as a transcript entry.
Forwarding is fault-isolated: a broker host fault is not an observer transport
fault, and must not be reported as one or tear down telemetry for every agent
over a single failed mutation.

`VerifiedRequest` gets a hand-written `Debug` that redacts the payload to a
byte count. Decrypted request content includes system prompts, and a derived
`Debug` would put them into any error or test failure that formats a request.

Verified at this tree: 2916 desktop Tauri Rust tests (74 broker, 18 ingress),
5311 desktop TS tests, workspace clippy and both fmt gates clean. The two
`buzz-pair-relay` integration failures are a pre-existing timing flake,
reproduced on clean origin/main at 074561233 and independent of these crates.

Signed-off-by: ss-dev-00 <a02c4e0850e5e612b4ddf95dbe2f5c56467cf27c6552203bc833ff438fb31971@buzz.block.builderlab.xyz>
Co-authored-by: Bradley Axen <baxen@squareup.com>
Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
ss-dev-00
2026-08-21 19:55:51 -07:00
parent 3a585fd8ed
commit 1938c1dbbd
10 changed files with 1361 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
desktop/src-tauri/src/broker/host.rs Normal file
View File

@@ -0,0 +1,178 @@
//! Wiring the broker into the running Desktop app.
//!
//! Everything above this file is testable without Tauri. This is where the
//! host's real parts get attached: the owner's keys, the agent roster, the agent
//! commands, the on-disk execution log, and the relay.
//!
//! # The command is a transport hand-off, not an authority hand-off
//!
//! [`broker_handle_observer_frame`] takes a raw signed event and nothing else.
//! The frontend cannot tell it who is asking, which owner to act as, or whether
//! a request is authorized — those come from the frame's signature and from
//! `AppState`. All the frontend does is forward bytes it already received on its
//! observer subscription, and it learns only whether the frame was broker
//! traffic. That keeps the trust boundary next to the credentials while reusing
//! the subscription that already exists.
use std::path::PathBuf;
use nostr::{Event, JsonUtil, Keys};
use tauri::{AppHandle, Manager, State};
use crate::app_state::AppState;
use super::agents_policy::AgentsAuthorizer;
use super::desktop_agents::{DesktopAgentRoster, DesktopAgentService};
use super::handlers::AgentsHandlers;
use super::ingress::{handle_frame, FrameRejection, Handled, ResultPublisher};
use super::pipeline::{BoxFuture, BrokerContext};
use super::store::{scoped_broker_db_path, SqliteExecutionLog};
/// Seconds allowed for a result frame to reach the relay and be acknowledged.
const RESULT_PUBLISH_TIMEOUT_SECS: u64 = 30;
/// Publishes result frames over the relay WebSocket.
///
/// Kind-24200 frames are only accepted on the relay's WebSocket path — the HTTP
/// bridge that every other Desktop publish uses rejects the kind outright — so
/// this opens a short-lived authenticated connection per result rather than
/// reusing `relay::submit_event`. Results are rare (one per brokered mutation),
/// which is what makes a per-result connection acceptable; a busier capability
/// would want a shared session instead.
pub struct WsResultPublisher {
relay_url: String,
keys: Keys,
}
impl WsResultPublisher {
/// Publish as `keys` against `relay_url`.
pub fn new(relay_url: String, keys: Keys) -> Self {
Self { relay_url, keys }
}
}
impl ResultPublisher for WsResultPublisher {
fn publish(&self, event: Event) -> BoxFuture<'_, Result<(), String>> {
Box::pin(async move {
buzz_ws_client_pkg::publish_event(
&self.relay_url,
event,
&self.keys,
// No NIP-OA auth tag: this publishes as the owner's own
// identity, which the relay authenticates directly. An auth tag
// is how a *managed agent* proves owner backing, and the owner
// is not one.
None,
RESULT_PUBLISH_TIMEOUT_SECS,
)
.await
.map_err(|error| format!("failed to publish broker result: {error}"))?;
Ok(())
})
}
}
/// What the caller learns about a forwarded frame.
///
/// Deliberately thin. A requester's outcome travels to the requester in a signed
/// result frame; telling the forwarding renderer what a brokered operation did
/// would make the reply path look like an IPC return value when it is not.
#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct FrameHandled {
/// True when the frame was a broker request this host executed.
pub broker_request: bool,
/// Correlation id, when there was a parseable one.
pub request_id: Option<String>,
/// True when the result frame reached the relay.
///
/// `false` with `broker_request: true` means the operation ran and its
/// outcome is recorded, but the requester did not hear it. A retry with the
/// same `requestId` replays the recorded outcome rather than re-running.
pub result_delivered: bool,
}
/// Resolve the execution log path for this owner and relay.
fn execution_log(app: &AppHandle, relay_url: &str, owner_pubkey: &str) -> Result<PathBuf, String> {
let base = app
.path()
.app_data_dir()
.map_err(|error| format!("failed to resolve app data dir: {error}"))?;
Ok(scoped_broker_db_path(&base, relay_url, owner_pubkey))
}
/// Execute a broker request carried by an inbound observer frame.
///
/// `event_json` is a signed relay event the caller received on its observer
/// subscription. Frames that are not broker requests are ignored cheaply; that
/// is the common case, since ordinary agent telemetry arrives here too.
///
/// # Errors
///
/// Returns an error for a host fault: unavailable owner keys, an unreadable
/// execution log, or an event that will not parse. A frame that fails
/// verification is not an error — it is reported as `brokerRequest: false`,
/// because a signed refusal addressed at an unverified sender is exactly what
/// this must not emit.
#[tauri::command]
pub async fn broker_handle_observer_frame(
event_json: String,
app: AppHandle,
state: State<'_, AppState>,
) -> Result<FrameHandled, String> {
let event = Event::from_json(&event_json)
.map_err(|error| format!("invalid observer event: {error}"))?;
let keys = state.signing_keys()?;
let relay_url = crate::relay::relay_ws_url_with_override(&state);
let owner_pubkey = keys.public_key().to_hex();
let log = SqliteExecutionLog::new(execution_log(&app, &relay_url, &owner_pubkey)?);
let authorizer = AgentsAuthorizer::new(DesktopAgentRoster::new(app.clone()));
let handlers = AgentsHandlers::new(DesktopAgentService::new(app.clone()));
let registry = handlers.registry();
let publisher = WsResultPublisher::new(relay_url.clone(), keys.clone());
let now = chrono::Utc::now().timestamp();
let handled = handle_frame(
&keys,
&event,
&relay_url,
now,
BrokerContext {
authorizer: &authorizer,
registry: &registry,
log: &log,
now,
},
&publisher,
)
.await?;
Ok(match handled {
Handled::Ignored(rejection) => {
if let FrameRejection::Rejected(message) = &rejection {
// A frame that claimed to be broker traffic and failed
// verification is worth seeing; ordinary telemetry is not.
eprintln!("buzz-desktop: broker: refused an inbound frame: {message}");
}
FrameHandled {
broker_request: false,
request_id: None,
result_delivered: false,
}
}
Handled::Executed { response, delivery } => {
if let Err(error) = &delivery {
eprintln!(
"buzz-desktop: broker: result for {} was not delivered: {error}",
response.request_id
);
}
FrameHandled {
broker_request: true,
request_id: Some(response.request_id.clone()),
result_delivered: delivery.is_ok(),
}
}
})
}

323
desktop/src-tauri/src/broker/ingress.rs Normal file
View File

@@ -0,0 +1,323 @@
//! Broker ingress: a verified frame in, a signed result out.
//!
//! This module owns the two transport edges the pipeline deliberately does not:
//! turning a received observer frame into a [`VerifiedRequest`], and turning the
//! resulting [`BrokerResponse`] into a frame addressed back to the requester.
//!
//! ```text
//! signed frame → verify → decrypt → pipeline::execute → encrypt → signed result
//! ```
//!
//! # Why the request is the whole plaintext
//!
//! A broker request is the *entire* decrypted frame payload — a bare
//! [`BrokerRequest`] JSON object. It is not wrapped in the ACP observer
//! envelope, and that is a load-bearing choice: the envelope carries per-attempt
//! fields (`seq`, `timestamp`), and the idempotency digest is a hash of the
//! bytes as received. Hashing an envelope would make a legitimate retry — same
//! `requestId`, same operation, a second later — hash differently and be refused
//! as a request-ID conflict. Keeping the envelope out of the plaintext means the
//! only way the digest changes is if the request itself changed.
//!
//! Frame-level identity (`created_at`, event id, signature) varies freely
//! between attempts and is *outside* the hashed bytes by construction.
//!
//! # Why a rejected frame is silent
//!
//! [`verify_frame`] failures produce no result frame. A frame that fails
//! verification has no trustworthy requester to address a reply to, and
//! answering one anyway would let anybody make this host emit signed frames at a
//! target of their choosing. Only a frame that verified — and that actually
//! claims to be a broker request — can be answered.
use buzz_core_pkg::observer::{
decrypt_observer_plaintext, encrypt_observer_payload, OBSERVER_AGENT_TAG,
OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY,
};
use buzz_sdk_pkg::broker::{BrokerResponse, BROKER_REQUEST_TYPE};
use buzz_sdk_pkg::kind::KIND_AGENT_OBSERVER_FRAME;
use nostr::{Event, Keys, Kind, PublicKey};
use zeroize::Zeroize;
use super::pipeline::{execute, BoxFuture, BrokerContext, VerifiedRequest};
use super::MAX_REQUEST_BYTES;
/// Largest accepted clock skew between a frame and this host, in seconds.
///
/// Matches the relay's observer-frame freshness window. Enforced again here
/// because the relay's check protects the relay: a host that trusted it would be
/// trusting a hop it does not control to bound replay of an old signed frame.
pub const MAX_FRAME_SKEW_SECS: i64 = 300;
/// Why an inbound frame is not a broker request this host will act on.
///
/// [`Self::NotBrokerTraffic`] is the ordinary case, not an error: every observer
/// telemetry frame an owner receives reaches this check, and almost none are
/// broker requests. It is separated from [`Self::Rejected`] so a caller can log
/// a real refusal without logging every agent heartbeat.
#[derive(Debug, PartialEq, Eq)]
pub enum FrameRejection {
/// The frame is well-formed observer traffic that is not a broker request.
NotBrokerTraffic,
/// The frame claims to be broker traffic but failed verification.
Rejected(String),
}
impl FrameRejection {
fn rejected(message: impl Into<String>) -> Self {
Self::Rejected(message.into())
}
}
/// Verify an inbound observer frame and decrypt it into a [`VerifiedRequest`].
///
/// Every field of the returned request is derived from the frame this host
/// verified, never from the request body:
///
/// - `owner_pubkey` is *this host's own* public key. It is not read from the
/// frame at all, so no signer can name a different owner.
/// - `requester_pubkey` is the verified signer.
/// - `relay_scope` is the relay the caller received the frame on.
///
/// # Errors
///
/// Returns [`FrameRejection::NotBrokerTraffic`] for a frame that is not a broker
/// request, and [`FrameRejection::Rejected`] when a frame that claims to be one
/// fails verification: bad id or signature, wrong kind, missing or duplicated
/// routing tags, a direction other than requester-to-owner telemetry, a
/// timestamp outside [`MAX_FRAME_SKEW_SECS`], an undecryptable body, or a
/// payload larger than [`MAX_REQUEST_BYTES`].
pub fn verify_frame(
owner_keys: &Keys,
event: &Event,
relay_scope: &str,
now: i64,
) -> Result<VerifiedRequest, FrameRejection> {
if event.kind != Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16) {
return Err(FrameRejection::NotBrokerTraffic);
}
// Cheap structural gates run before the expensive ones, but nothing that
// could admit a frame is skipped: id and signature are checked before the
// content is decrypted, and the content is bounded before it is parsed.
if single_tag(event, OBSERVER_FRAME_TAG).as_deref() != Some(OBSERVER_FRAME_TELEMETRY) {
return Err(FrameRejection::NotBrokerTraffic);
}
if !event.verify_id() {
return Err(FrameRejection::rejected("observer frame has an invalid id"));
}
if !event.verify_signature() {
return Err(FrameRejection::rejected(
"observer frame has an invalid signature",
));
}
let skew = event.created_at.as_secs() as i64 - now;
if skew.abs() > MAX_FRAME_SKEW_SECS {
return Err(FrameRejection::rejected(format!(
"observer frame timestamp is {skew}s from now, outside the ±{MAX_FRAME_SKEW_SECS}s window"
)));
}
let owner = owner_keys.public_key();
let recipient = single_pubkey_tag(event, "p")
.ok_or_else(|| FrameRejection::rejected("observer frame has no single p tag"))?;
if recipient != owner {
// Addressed to a different owner. This host holds no credentials that
// could act on it, and cannot decrypt it either.
return Err(FrameRejection::NotBrokerTraffic);
}
let agent = single_pubkey_tag(event, OBSERVER_AGENT_TAG)
.ok_or_else(|| FrameRejection::rejected("observer frame has no single agent tag"))?;
// Requester-to-owner telemetry only: the signer must be the agent the frame
// names. A frame signed by anyone else claiming to be an agent's telemetry
// would let a third party borrow that agent's authorization.
if event.pubkey != agent {
return Err(FrameRejection::rejected(
"observer frame signer is not the agent it names",
));
}
if agent == owner {
// The owner key signing owner-addressed telemetry is not a delegated
// request; policy refuses it too, but it never becomes a request here.
return Err(FrameRejection::NotBrokerTraffic);
}
let mut plaintext = decrypt_observer_plaintext(owner_keys, event)
.map_err(|error| FrameRejection::rejected(format!("frame decrypt failed: {error}")))?;
let outcome = classify_plaintext(&plaintext);
let verified = match outcome {
Ok(()) => Ok(VerifiedRequest {
owner_pubkey: owner.to_hex(),
requester_pubkey: event.pubkey.to_hex(),
relay_scope: relay_scope.to_string(),
payload: plaintext.as_bytes().to_vec(),
}),
Err(rejection) => Err(rejection),
};
plaintext.zeroize();
verified
}
/// Decide whether a decrypted payload is a broker request worth dispatching.
///
/// Deliberately does not validate the request: an ill-formed *broker* request
/// must reach the pipeline so the requester gets a `invalid_request` answer,
/// while a payload that never claimed to be one must not be answered at all.
fn classify_plaintext(plaintext: &str) -> Result<(), FrameRejection> {
if plaintext.len() > MAX_REQUEST_BYTES {
return Err(FrameRejection::rejected(format!(
"broker request payload exceeds {MAX_REQUEST_BYTES} bytes (got {})",
plaintext.len()
)));
}
let peeked: serde_json::Value =
serde_json::from_str(plaintext).map_err(|_| FrameRejection::NotBrokerTraffic)?;
if peeked.get("type").and_then(serde_json::Value::as_str) == Some(BROKER_REQUEST_TYPE) {
Ok(())
} else {
Err(FrameRejection::NotBrokerTraffic)
}
}
fn single_tag(event: &Event, name: &str) -> Option<String> {
let mut values = event
.tags
.iter()
.filter(|tag| tag.kind().to_string() == name)
.filter_map(|tag| tag.content());
let value = values.next()?;
// A duplicated routing tag is ambiguous, and picking one would let a frame
// present a different route to this host than it did to the relay.
if values.next().is_some() {
return None;
}
Some(value.to_string())
}
fn single_pubkey_tag(event: &Event, name: &str) -> Option<PublicKey> {
PublicKey::from_hex(&single_tag(event, name)?).ok()
}
/// Build the signed frame that carries `response` back to `requester`.
///
/// The result travels as an owner-to-agent *control* frame: encrypted to the
/// requester, signed by the owner, addressed with the requester as both
/// recipient and agent. That is the only direction the relay routes owner-signed
/// observer traffic in, and it means a result is readable by exactly the agent
/// that asked for it.
///
/// # Errors
///
/// Returns an error if `requester` is not a valid hex pubkey, or if the response
/// cannot be encrypted, built, or signed.
pub fn build_result_frame(
owner_keys: &Keys,
requester: &str,
response: &BrokerResponse,
) -> Result<Event, String> {
let requester = PublicKey::from_hex(requester.trim())
.map_err(|error| format!("invalid requester pubkey: {error}"))?;
let requester_hex = requester.to_hex();
let encrypted = encrypt_observer_payload(owner_keys, &requester, response)
.map_err(|error| format!("failed to encrypt broker result: {error}"))?;
buzz_sdk_pkg::build_agent_observer_frame(
&requester_hex,
&requester_hex,
OBSERVER_FRAME_CONTROL,
&encrypted,
)
.map_err(|error| format!("failed to build broker result frame: {error}"))?
.sign_with_keys(owner_keys)
.map_err(|error| format!("failed to sign broker result frame: {error}"))
}
/// Delivers a signed result frame to the relay.
///
/// A trait because kind-24200 frames are only accepted over the relay's
/// WebSocket path — the HTTP bridge rejects the kind outright — so delivery is a
/// different transport from every other Desktop publish. Isolating it here keeps
/// that fact in one place and lets the ingress path be tested without a relay.
pub trait ResultPublisher: Send + Sync {
/// Publish `event`, or report why delivery failed.
fn publish(&self, event: Event) -> BoxFuture<'_, Result<(), String>>;
}
/// The outcome of handling one inbound frame.
#[derive(Debug)]
pub enum Handled {
/// The frame was not a broker request for this host.
Ignored(FrameRejection),
/// The request produced `response`, and delivery reported `delivery`.
///
/// `delivery` is separate from the response on purpose: a request can
/// execute successfully and still fail to deliver its result. That must not
/// be reported as an execution failure, and it must not retry the execution
/// — the recorded outcome is already durable and a retry with the same
/// `requestId` replays it.
Executed {
/// The terminal response the pipeline produced.
response: Box<BrokerResponse>,
/// Whether the result frame reached the relay.
delivery: Result<(), String>,
},
}
/// Verify, execute, and answer one inbound frame.
///
/// # Errors
///
/// Returns `Err` only for a host fault that leaves no deliverable response —
/// an unreachable execution log, for instance. Refusals the requester should
/// hear about come back inside [`Handled::Executed`].
pub async fn handle_frame(
owner_keys: &Keys,
event: &Event,
relay_scope: &str,
now: i64,
ctx: BrokerContext<'_>,
publisher: &dyn ResultPublisher,
) -> Result<Handled, String> {
let request = match verify_frame(owner_keys, event, relay_scope, now) {
Ok(request) => request,
Err(rejection) => return Ok(Handled::Ignored(rejection)),
};
let requester = request.requester_pubkey.clone();
let response = execute(&request, ctx).await?;
// Audit line: identities, capability scope, and disposition. No prompts, no
// arguments, no decrypted payload — an audit trail that leaks the thing it
// audits is worse than none.
eprintln!(
"buzz-desktop: broker: request={} requester={} scope={} disposition={} replayed={}",
response.request_id,
requester,
relay_scope,
disposition(&response),
response.replayed
);
let delivery = match build_result_frame(owner_keys, &requester, &response) {
Ok(frame) => publisher.publish(frame).await,
Err(error) => Err(error),
};
Ok(Handled::Executed {
response: Box::new(response),
delivery,
})
}
/// Audit word for a response's terminal disposition.
fn disposition(response: &BrokerResponse) -> &'static str {
use buzz_sdk_pkg::broker::BrokerResult;
match response.result {
BrokerResult::Succeeded { .. } => "succeeded",
BrokerResult::Failed { .. } => "failed",
BrokerResult::Indeterminate { .. } => "indeterminate",
}
}
#[cfg(test)]
mod tests;

614
desktop/src-tauri/src/broker/ingress/tests.rs Normal file
View File

@@ -0,0 +1,614 @@
//! Ingress tests: frame verification, the digest boundary, and result delivery.
use std::sync::Mutex;
use buzz_sdk_pkg::broker::{
AgentsCreateArgs, AgentsCreateOutcome, BrokerErrorCode, BrokerRequest, BrokerResult,
Capability, CapabilityArgs, CapabilityOutcome,
};
use nostr::{EventBuilder, Tag, Timestamp};
use super::super::pipeline::{Authorizer, CapabilityHandler, CapabilityRegistry, HandlerOutcome};
use super::super::store::MemoryExecutionLog;
use super::*;
const CHANNEL: &str = "b2c38ca8-9ec3-411e-bab5-f9deab34d52e";
const RELAY: &str = "wss://relay.example";
const NOW: i64 = 1_700_000_000;
fn request(request_id: &str) -> BrokerRequest {
BrokerRequest::new(
request_id,
CapabilityArgs::AgentsCreate(AgentsCreateArgs {
channel_id: CHANNEL.into(),
display_name: "Helper".into(),
system_prompt: "Help.".into(),
runtime: None,
provider: None,
model: None,
respond_to: None,
}),
)
.expect("valid request")
}
/// Seal `payload` as requester-to-owner telemetry, the shape a real request has.
fn frame(requester: &Keys, owner: &Keys, payload: &serde_json::Value, created_at: i64) -> Event {
let encrypted =
encrypt_observer_payload(requester, &owner.public_key(), payload).expect("encrypt payload");
EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
.tags([
Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
Tag::parse([OBSERVER_AGENT_TAG, &requester.public_key().to_hex()]).expect("agent tag"),
Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
])
.custom_created_at(Timestamp::from(created_at as u64))
.sign_with_keys(requester)
.expect("sign frame")
}
fn request_frame(requester: &Keys, owner: &Keys, request: &BrokerRequest) -> Event {
frame(
requester,
owner,
&serde_json::to_value(request).expect("request json"),
NOW,
)
}
struct AllowAll;
impl Authorizer for AllowAll {
fn authorize<'a>(
&'a self,
_request: &'a VerifiedRequest,
_capability: Capability,
_parsed: &'a BrokerRequest,
) -> BoxFuture<'a, Result<(), buzz_sdk_pkg::broker::BrokerError>> {
Box::pin(async { Ok(()) })
}
}
/// Records every execution so a test can prove a retry did not re-run one.
struct RecordingHandler {
calls: Mutex<Vec<String>>,
}
impl RecordingHandler {
fn new() -> Self {
Self {
calls: Mutex::new(Vec::new()),
}
}
fn calls(&self) -> usize {
self.calls.lock().expect("calls").len()
}
}
impl CapabilityHandler for RecordingHandler {
fn execute<'a>(
&'a self,
_request: &'a VerifiedRequest,
parsed: &'a BrokerRequest,
) -> BoxFuture<'a, HandlerOutcome> {
Box::pin(async move {
self.calls
.lock()
.expect("calls")
.push(parsed.request_id.clone());
HandlerOutcome::Succeeded(CapabilityOutcome::AgentsCreate(AgentsCreateOutcome {
agent_pubkey: "44".repeat(32),
display_name: "Helper".into(),
channel_id: CHANNEL.into(),
}))
})
}
}
struct OneHandler<'a>(&'a RecordingHandler);
impl CapabilityRegistry for OneHandler<'_> {
fn handler(&self, _capability: Capability) -> Option<&dyn CapabilityHandler> {
Some(self.0)
}
}
/// Captures published frames instead of reaching a relay.
struct CapturingPublisher {
published: Mutex<Vec<Event>>,
fail: Option<String>,
}
impl CapturingPublisher {
fn new() -> Self {
Self {
published: Mutex::new(Vec::new()),
fail: None,
}
}
fn failing(message: &str) -> Self {
Self {
published: Mutex::new(Vec::new()),
fail: Some(message.to_string()),
}
}
fn frames(&self) -> Vec<Event> {
self.published.lock().expect("published").clone()
}
}
impl ResultPublisher for CapturingPublisher {
fn publish(&self, event: Event) -> BoxFuture<'_, Result<(), String>> {
Box::pin(async move {
self.published.lock().expect("published").push(event);
match &self.fail {
Some(message) => Err(message.clone()),
None => Ok(()),
}
})
}
}
struct Harness {
owner: Keys,
requester: Keys,
log: MemoryExecutionLog,
handler: RecordingHandler,
}
impl Harness {
fn new() -> Self {
Self {
owner: Keys::generate(),
requester: Keys::generate(),
log: MemoryExecutionLog::new(),
handler: RecordingHandler::new(),
}
}
async fn handle(&self, event: &Event, publisher: &dyn ResultPublisher) -> Handled {
let registry = OneHandler(&self.handler);
handle_frame(
&self.owner,
event,
RELAY,
NOW,
BrokerContext {
authorizer: &AllowAll,
registry: &registry,
log: &self.log,
now: NOW,
},
publisher,
)
.await
.expect("no host fault")
}
}
fn response_of(handled: &Handled) -> &BrokerResponse {
match handled {
Handled::Executed { response, .. } => response,
Handled::Ignored(rejection) => panic!("expected execution, got {rejection:?}"),
}
}
fn rejection_of(handled: &Handled) -> &FrameRejection {
match handled {
Handled::Ignored(rejection) => rejection,
Handled::Executed { response, .. } => panic!("expected rejection, got {response:?}"),
}
}
// ── identity is transport-derived ───────────────────────────────────────────
#[test]
fn a_verified_frame_yields_transport_derived_identity() {
let owner = Keys::generate();
let requester = Keys::generate();
let event = request_frame(&requester, &owner, &request("req-1"));
let verified = verify_frame(&owner, &event, RELAY, NOW).expect("verified");
assert_eq!(verified.owner_pubkey, owner.public_key().to_hex());
assert_eq!(verified.requester_pubkey, requester.public_key().to_hex());
assert_eq!(verified.relay_scope, RELAY);
}
/// The owner is this host's own key, never anything the frame carried — so a
/// request body that names another owner changes nothing about who is acted on.
#[test]
fn a_payload_cannot_name_its_own_owner_or_requester() {
let owner = Keys::generate();
let requester = Keys::generate();
let mut payload = serde_json::to_value(request("req-1")).expect("json");
payload["ownerPubkey"] = serde_json::json!("99".repeat(32));
payload["requesterPubkey"] = serde_json::json!("88".repeat(32));
let event = frame(&requester, &owner, &payload, NOW);
let verified = verify_frame(&owner, &event, RELAY, NOW).expect("verified");
assert_eq!(verified.owner_pubkey, owner.public_key().to_hex());
assert_eq!(verified.requester_pubkey, requester.public_key().to_hex());
}
// ── the digest boundary ─────────────────────────────────────────────────────
/// The load-bearing retry property: two attempts at the same operation differ in
/// every frame-level field — event id, signature, `created_at` — and must still
/// replay rather than conflict. If the digest ever covered frame or envelope
/// metadata, this test fails with `request_id_conflict`.
#[tokio::test]
async fn a_retry_with_a_new_frame_replays_instead_of_conflicting() {
let harness = Harness::new();
let request = request("req-retry");
let publisher = CapturingPublisher::new();
let first = request_frame(&harness.requester, &harness.owner, &request);
let second = frame(
&harness.requester,
&harness.owner,
&serde_json::to_value(&request).expect("json"),
NOW + 60,
);
assert_ne!(first.id, second.id, "retry must be a distinct frame");
assert_ne!(first.created_at, second.created_at);
let first = harness.handle(&first, &publisher).await;
assert!(response_of(&first).result.is_succeeded());
assert!(!response_of(&first).replayed);
let second = harness.handle(&second, &publisher).await;
let second = response_of(&second);
assert!(
second.result.is_succeeded(),
"retry must replay the recorded success, got {:?}",
second.result
);
assert!(second.replayed, "retry must be marked as a replay");
assert_eq!(harness.handler.calls(), 1, "the handler must run once");
}
/// The other half of the contract: same `requestId`, genuinely different
/// request. This must be refused rather than answered with the first outcome.
#[tokio::test]
async fn the_same_request_id_with_different_content_still_conflicts() {
let harness = Harness::new();
let publisher = CapturingPublisher::new();
let first = request_frame(&harness.requester, &harness.owner, &request("req-1"));
harness.handle(&first, &publisher).await;
let mut changed = request("req-1");
if let CapabilityArgs::AgentsCreate(args) = &mut changed.capability {
args.display_name = "Different".into();
}
let second = request_frame(&harness.requester, &harness.owner, &changed);
let response = harness.handle(&second, &publisher).await;
assert_eq!(
response_of(&response)
.result
.error()
.map(|error| error.code),
Some(BrokerErrorCode::RequestIdConflict)
);
assert_eq!(harness.handler.calls(), 1);
}
// ── what is refused, and what is silently ignored ───────────────────────────
#[test]
fn ordinary_telemetry_is_not_broker_traffic() {
let owner = Keys::generate();
let requester = Keys::generate();
let event = frame(
&requester,
&owner,
&serde_json::json!({"kind": "turn_started", "seq": 4}),
NOW,
);
assert_eq!(
verify_frame(&owner, &event, RELAY, NOW).unwrap_err(),
FrameRejection::NotBrokerTraffic
);
}
#[test]
fn a_frame_for_another_owner_is_not_this_hosts_traffic() {
let owner = Keys::generate();
let other_owner = Keys::generate();
let requester = Keys::generate();
let event = request_frame(&requester, &other_owner, &request("req-1"));
assert_eq!(
verify_frame(&owner, &event, RELAY, NOW).unwrap_err(),
FrameRejection::NotBrokerTraffic
);
}
/// A frame signed by someone other than the agent it names would let a third
/// party borrow that agent's authorization.
#[test]
fn a_signer_that_is_not_the_named_agent_is_refused() {
let owner = Keys::generate();
let requester = Keys::generate();
let impostor = Keys::generate();
let encrypted = encrypt_observer_payload(
&impostor,
&owner.public_key(),
&serde_json::to_value(request("req-1")).expect("json"),
)
.expect("encrypt");
let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
.tags([
Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
Tag::parse([OBSERVER_AGENT_TAG, &requester.public_key().to_hex()]).expect("agent tag"),
Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
])
.custom_created_at(Timestamp::from(NOW as u64))
.sign_with_keys(&impostor)
.expect("sign");
let error = verify_frame(&owner, &event, RELAY, NOW).unwrap_err();
assert!(
matches!(&error, FrameRejection::Rejected(message) if message.contains("not the agent it names")),
"unexpected: {error:?}"
);
}
#[test]
fn a_stale_frame_is_refused() {
let owner = Keys::generate();
let requester = Keys::generate();
let event = frame(
&requester,
&owner,
&serde_json::to_value(request("req-1")).expect("json"),
NOW - MAX_FRAME_SKEW_SECS - 1,
);
let error = verify_frame(&owner, &event, RELAY, NOW).unwrap_err();
assert!(
matches!(&error, FrameRejection::Rejected(message) if message.contains("outside the")),
"unexpected: {error:?}"
);
}
#[test]
fn a_future_dated_frame_is_refused() {
let owner = Keys::generate();
let requester = Keys::generate();
let event = frame(
&requester,
&owner,
&serde_json::to_value(request("req-1")).expect("json"),
NOW + MAX_FRAME_SKEW_SECS + 1,
);
assert!(matches!(
verify_frame(&owner, &event, RELAY, NOW),
Err(FrameRejection::Rejected(_))
));
}
/// A tampered frame must be caught here, not merely trusted from the relay.
#[test]
fn a_frame_with_a_broken_signature_is_refused() {
let owner = Keys::generate();
let requester = Keys::generate();
let other = Keys::generate();
let event = request_frame(&requester, &owner, &request("req-1"));
let mut forged = event.clone();
forged.pubkey = other.public_key();
let error = verify_frame(&owner, &forged, RELAY, NOW).unwrap_err();
assert!(
matches!(&error, FrameRejection::Rejected(message) if message.contains("invalid id")
|| message.contains("invalid signature")),
"unexpected: {error:?}"
);
}
#[test]
fn a_duplicated_routing_tag_is_refused() {
let owner = Keys::generate();
let requester = Keys::generate();
let encrypted = encrypt_observer_payload(
&requester,
&owner.public_key(),
&serde_json::to_value(request("req-1")).expect("json"),
)
.expect("encrypt");
let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
.tags([
Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
Tag::parse(["p", &Keys::generate().public_key().to_hex()]).expect("second p tag"),
Tag::parse([OBSERVER_AGENT_TAG, &requester.public_key().to_hex()]).expect("agent tag"),
Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
])
.custom_created_at(Timestamp::from(NOW as u64))
.sign_with_keys(&requester)
.expect("sign");
assert!(matches!(
verify_frame(&owner, &event, RELAY, NOW),
Err(FrameRejection::Rejected(_))
));
}
#[test]
fn a_control_frame_is_not_an_inbound_request() {
let owner = Keys::generate();
let requester = Keys::generate();
let encrypted = encrypt_observer_payload(
&owner,
&requester.public_key(),
&serde_json::to_value(request("req-1")).expect("json"),
)
.expect("encrypt");
let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
.tags([
Tag::parse(["p", &requester.public_key().to_hex()]).expect("p tag"),
Tag::parse([OBSERVER_AGENT_TAG, &requester.public_key().to_hex()]).expect("agent tag"),
Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_CONTROL]).expect("frame tag"),
])
.custom_created_at(Timestamp::from(NOW as u64))
.sign_with_keys(&owner)
.expect("sign");
assert_eq!(
verify_frame(&owner, &event, RELAY, NOW).unwrap_err(),
FrameRejection::NotBrokerTraffic
);
}
#[tokio::test]
async fn an_ignored_frame_never_reaches_a_handler_or_publishes() {
let harness = Harness::new();
let publisher = CapturingPublisher::new();
let event = frame(
&harness.requester,
&harness.owner,
&serde_json::json!({"kind": "turn_started"}),
NOW,
);
let handled = harness.handle(&event, &publisher).await;
assert_eq!(rejection_of(&handled), &FrameRejection::NotBrokerTraffic);
assert_eq!(harness.handler.calls(), 0);
assert!(publisher.frames().is_empty());
}
/// A malformed *broker* request must be answered — the requester needs to hear
/// `invalid_request` — while never reaching a capability.
#[tokio::test]
async fn a_malformed_broker_request_is_answered_but_not_executed() {
let harness = Harness::new();
let publisher = CapturingPublisher::new();
let event = frame(
&harness.requester,
&harness.owner,
&serde_json::json!({"type": "broker_request", "requestId": "req-1"}),
NOW,
);
let handled = harness.handle(&event, &publisher).await;
assert_eq!(
response_of(&handled).result.error().map(|error| error.code),
Some(BrokerErrorCode::InvalidRequest)
);
assert_eq!(harness.handler.calls(), 0);
assert_eq!(
publisher.frames().len(),
1,
"the refusal is still delivered"
);
}
// ── result delivery ─────────────────────────────────────────────────────────
/// The requester must be able to read its own result: the frame is a control
/// frame encrypted to the requester and signed by the owner.
#[tokio::test]
async fn the_result_frame_is_readable_by_the_requester_and_signed_by_the_owner() {
let harness = Harness::new();
let publisher = CapturingPublisher::new();
let event = request_frame(&harness.requester, &harness.owner, &request("req-1"));
harness.handle(&event, &publisher).await;
let frames = publisher.frames();
assert_eq!(frames.len(), 1);
let result_frame = &frames[0];
assert_eq!(result_frame.pubkey, harness.owner.public_key());
assert!(result_frame.verify_signature());
assert_eq!(
result_frame.kind,
Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16)
);
assert_eq!(
single_tag(result_frame, OBSERVER_FRAME_TAG).as_deref(),
Some(OBSERVER_FRAME_CONTROL),
"an owner-signed result must travel as a control frame"
);
assert_eq!(
single_pubkey_tag(result_frame, "p"),
Some(harness.requester.public_key())
);
let decrypted: BrokerResponse =
buzz_core_pkg::observer::decrypt_observer_payload(&harness.requester, result_frame)
.expect("requester can read its result");
decrypted.validate().expect("a valid result envelope");
assert_eq!(decrypted.request_id, "req-1");
assert!(decrypted.result.is_succeeded());
}
/// The published result carries no secret material — the create outcome names
/// the new agent's public key only.
#[tokio::test]
async fn a_result_frame_carries_no_secret_material() {
let harness = Harness::new();
let publisher = CapturingPublisher::new();
let event = request_frame(&harness.requester, &harness.owner, &request("req-1"));
harness.handle(&event, &publisher).await;
let decrypted: serde_json::Value = buzz_core_pkg::observer::decrypt_observer_payload(
&harness.requester,
&publisher.frames()[0],
)
.expect("decrypt");
let json = serde_json::to_string(&decrypted).expect("json");
for forbidden in ["nsec", "privateKey", "private_key", "secret"] {
assert!(
!json.contains(forbidden),
"result leaked {forbidden}: {json}"
);
}
}
/// A delivery failure must not be reported as an execution failure, and must not
/// undo the recorded outcome: the retry replays the same success.
#[tokio::test]
async fn a_delivery_failure_leaves_the_recorded_outcome_intact() {
let harness = Harness::new();
let failing = CapturingPublisher::failing("relay unreachable");
let request = request("req-1");
let event = request_frame(&harness.requester, &harness.owner, &request);
let handled = harness.handle(&event, &failing).await;
match &handled {
Handled::Executed { response, delivery } => {
assert!(response.result.is_succeeded());
assert_eq!(delivery.as_ref().unwrap_err(), "relay unreachable");
}
Handled::Ignored(rejection) => panic!("expected execution, got {rejection:?}"),
}
let ok = CapturingPublisher::new();
let retry = frame(
&harness.requester,
&harness.owner,
&serde_json::to_value(&request).expect("json"),
NOW + 5,
);
let handled = harness.handle(&retry, &ok).await;
let response = response_of(&handled);
assert!(response.result.is_succeeded());
assert!(response.replayed);
assert_eq!(
harness.handler.calls(),
1,
"a failed delivery must not cause a re-execution"
);
}
#[test]
fn a_result_frame_for_a_malformed_requester_is_an_error_not_a_panic() {
let owner = Keys::generate();
let response = BrokerResponse::new(
"req-1",
BrokerResult::succeeded(CapabilityOutcome::AgentsCreate(AgentsCreateOutcome {
agent_pubkey: "44".repeat(32),
display_name: "Helper".into(),
channel_id: CHANNEL.into(),
})),
);
assert!(build_result_frame(&owner, "not-a-pubkey", &response).is_err());
}

2
desktop/src-tauri/src/broker/mod.rs
View File

@@ -35,6 +35,8 @@ use sha2::{Digest, Sha256};
pub mod agents_policy;
pub mod desktop_agents;
pub mod handlers;
pub mod host;
pub mod ingress;
pub mod pipeline;
pub mod store;

16
desktop/src-tauri/src/broker/pipeline.rs
View File

@@ -63,6 +63,22 @@ pub struct VerifiedRequest {
pub payload: Vec<u8>,
}
/// Redacts the payload.
///
/// `payload` is decrypted request content — a system prompt, for instance. It is
/// not a credential, but it is not log material either, and a derived `Debug`
/// would put it into any error message or test failure that formats a request.
impl std::fmt::Debug for VerifiedRequest {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("VerifiedRequest")
.field("owner_pubkey", &self.owner_pubkey)
.field("requester_pubkey", &self.requester_pubkey)
.field("relay_scope", &self.relay_scope)
.field("payload_bytes", &self.payload.len())
.finish()
}
}
/// The durable execution log, as the pipeline needs it.
///
/// A trait rather than a `&mut Connection` because the pipeline is async and

5
desktop/src-tauri/src/broker/store.rs
View File

@@ -479,6 +479,11 @@ impl ExecutionLog for SqliteExecutionLog {
}
/// Read one execution row, if it exists.
///
/// Test-only: the pipeline never reads a record outside [`claim`], which returns
/// what it found in the same transaction that decided whether to claim. A
/// separate read would be a second, racier way to ask the same question.
#[cfg(test)]
pub fn get(conn: &Connection, key: &ExecutionKey) -> Result<Option<ExecutionRecord>, String> {
conn.query_row(
"SELECT state, request_digest, capability_version, result_json

1
desktop/src-tauri/src/lib.rs
View File

@@ -611,6 +611,7 @@ pub fn run() {
sign_out,
decrypt_observer_event,
build_observer_control_event,
broker::host::broker_handle_observer_frame,
create_auth_event,
nip44_encrypt_to_self,
nip44_decrypt_from_self,

126
desktop/src/features/agents/brokerFrameForwarding.test.mjs Normal file
View File

@@ -0,0 +1,126 @@
import assert from "node:assert/strict";
import test from "node:test";
import {
BROKER_REQUEST_TYPE,
isBrokerRequestPayload,
} from "./brokerFrameForwarding.ts";
test("a broker request payload is recognized by its type discriminator", () => {
assert.equal(
isBrokerRequestPayload({
type: BROKER_REQUEST_TYPE,
protocolVersion: 1,
requestId: "req-1",
}),
true,
);
});
test("an ill-formed broker request is still forwarded", () => {
// Validation belongs to the host: a request the renderer judged invalid and
// dropped would leave the requester waiting with no answer at all.
assert.equal(isBrokerRequestPayload({ type: BROKER_REQUEST_TYPE }), true);
});
test("agent telemetry is not forwarded", () => {
for (const payload of [
{ kind: "turn_started", seq: 3 },
{ type: "agent_management_request", action: "create" },
{ type: "broker_result", requestId: "req-1" },
null,
undefined,
"broker_request",
42,
]) {
assert.equal(
isBrokerRequestPayload(payload),
false,
`unexpectedly treated as a broker request: ${JSON.stringify(payload)}`,
);
}
});
// ── Forwarding: signed-event payload and fault isolation ──────────────────────
//
// The Tauri IPC bridge is stubbed at globalThis.__TAURI_INTERNALS__.invoke, the
// same pattern acpRuntimesQuery.test.mjs uses, so the command name and payload
// are observed directly.
const SIGNED_FRAME = {
id: "ffff",
pubkey: "aaaa",
created_at: 1,
kind: 24200,
tags: [["frame", "telemetry"]],
content: "nip44-ciphertext",
sig: "bbbb",
};
function stubInvoke(handler) {
// @tauri-apps/api reads window.__TAURI_INTERNALS__, so the shim needs a
// window; jsdom is not loaded for this plain-helper test.
const hadWindow = typeof globalThis.window !== "undefined";
if (!hadWindow) {
globalThis.window = globalThis;
}
const previous = globalThis.window.__TAURI_INTERNALS__;
const calls = [];
globalThis.window.__TAURI_INTERNALS__ = {
invoke: async (command, payload) => {
calls.push({ command, payload });
return handler(command, payload);
},
};
return {
calls,
restore() {
globalThis.window.__TAURI_INTERNALS__ = previous;
if (!hadWindow) {
delete globalThis.window;
}
},
};
}
test("forwarding hands the host the signed event, not the plaintext", async () => {
const stub = stubInvoke(() => ({
brokerRequest: true,
requestId: "req-1",
resultDelivered: true,
}));
try {
const { forwardBrokerFrame } = await import("./brokerFrameForwarding.ts");
const handled = await forwardBrokerFrame(SIGNED_FRAME);
assert.equal(stub.calls.length, 1);
assert.equal(stub.calls[0].command, "broker_handle_observer_frame");
// The host must re-verify and re-decrypt from the bytes the requester
// signed; a decrypted payload would make the renderer the authority on what
// the request says.
assert.deepEqual(JSON.parse(stub.calls[0].payload.eventJson), SIGNED_FRAME);
assert.equal(handled.requestId, "req-1");
} finally {
stub.restore();
}
});
test("a broker host fault does not escape to the observer subscription", async () => {
// Regression: forwarding used to run inside the observer decrypt try/catch, so
// a broker fault surfaced as "Observer event decrypt failed" and tore down the
// telemetry subscription for every agent over one failed mutation.
const stub = stubInvoke(() => {
throw new Error("owner keys unavailable");
});
const consoleError = console.error;
console.error = () => {};
try {
const { forwardBrokerFrameIsolated } = await import(
"./brokerFrameForwarding.ts"
);
assert.equal(await forwardBrokerFrameIsolated(SIGNED_FRAME), null);
} finally {
console.error = consoleError;
stub.restore();
}
});

76
desktop/src/features/agents/brokerFrameForwarding.ts Normal file
View File

@@ -0,0 +1,76 @@
import { invokeTauri } from "@/shared/api/tauri";
import type { RelayEvent } from "@/shared/api/types";
/**
* Wire `type` discriminator on a broker request payload. Must match
* `BROKER_REQUEST_TYPE` in `crates/buzz-sdk/src/broker/mod.rs`.
*/
export const BROKER_REQUEST_TYPE = "broker_request";
export type BrokerFrameHandled = {
brokerRequest: boolean;
requestId: string | null;
resultDelivered: boolean;
};
/**
* True when a decrypted observer payload is a broker request rather than agent
* telemetry.
*
* This is a **routing** decision, not a trust decision. Rust re-verifies the
* signature, re-derives the owner and requester from the frame, re-decrypts, and
* re-validates before anything executes — so a wrong answer here can only cause
* a broker request to be missed (the requester's wait expires) or a non-request
* to be forwarded and ignored. It cannot cause an unauthorized mutation.
*
* The check is deliberately shallow: the renderer must not decide whether a
* request is *valid*, only whether it claims to be one. An ill-formed broker
* request has to reach the host so the requester gets a structured refusal
* instead of silence.
*/
export function isBrokerRequestPayload(payload: unknown): boolean {
return (
typeof payload === "object" &&
payload !== null &&
(payload as { type?: unknown }).type === BROKER_REQUEST_TYPE
);
}
/**
* Hand a signed observer frame to the Rust broker for execution.
*
* The **raw signed event** is forwarded, not the decrypted payload: the host
* must verify and decrypt it itself, from bytes the requester signed. Passing
* the plaintext would make the renderer the thing that decides what a request
* says, which is precisely the authority split the broker exists to avoid.
*/
export async function forwardBrokerFrame(
event: RelayEvent,
): Promise<BrokerFrameHandled> {
return invokeTauri<BrokerFrameHandled>("broker_handle_observer_frame", {
eventJson: JSON.stringify(event),
});
}
/**
* Forward a broker frame without letting a host fault escape to the caller.
*
* The observer subscription carries agent telemetry for the whole app. A broker
* host fault is a different subsystem's failure, so it must not surface as an
* observer transport error — doing so would report a misleading cause and tear
* down telemetry for every agent over a failed mutation. The requester is
* unaffected either way: it learns the outcome from its own signed result frame,
* or hears nothing and retries with the same `requestId`.
*
* Returns `null` when forwarding failed.
*/
export async function forwardBrokerFrameIsolated(
event: RelayEvent,
): Promise<BrokerFrameHandled | null> {
try {
return await forwardBrokerFrame(event);
} catch (error) {
console.error("Broker frame forwarding failed", error);
return null;
}
}

22
desktop/src/features/agents/observerRelayStore.ts
View File

@@ -11,6 +11,10 @@ import {
parseAgentManagementRequest,
type AgentManagementRequest,
} from "./agentManagement";
import {
forwardBrokerFrameIsolated,
isBrokerRequestPayload,
} from "./brokerFrameForwarding";
import { normalizePubkey } from "@/shared/lib/pubkey";
import { useQueryClient } from "@tanstack/react-query";
import { agentConfigSurfaceQueryKey } from "@/features/agents/hooks";
@@ -559,11 +563,25 @@ async function handleRelayObserverEvent(
}
try {
const parsed = (await decryptObserverEvent(event)) as ObserverEvent;
const parsed = await decryptObserverEvent(event);
if (activeGeneration !== generation) {
return;
}
processLiveObserverEvents(agentPubkey, unwrapObserverBatch(parsed));
// Broker requests are not observer telemetry: the payload is a bare
// BrokerRequest, with no seq/timestamp/kind, so appending it to the session
// journal would file a mutation request as a transcript entry. Route it to
// the host instead, and forward the *signed event* rather than this
// plaintext — the host verifies and decrypts it again itself.
if (isBrokerRequestPayload(parsed)) {
// Isolated: a broker host fault must not be reported as — or tear down —
// the observer telemetry subscription. See forwardBrokerFrameIsolated.
await forwardBrokerFrameIsolated(event);
return;
}
processLiveObserverEvents(
agentPubkey,
unwrapObserverBatch(parsed as ObserverEvent),
);
} catch (error) {
if (activeGeneration !== generation) {
return;