diff --git a/desktop/src-tauri/src/broker/host.rs b/desktop/src-tauri/src/broker/host.rs new file mode 100644 index 0000000000..0643e0d283 --- /dev/null +++ b/desktop/src-tauri/src/broker/host.rs @@ -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, + /// 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 { + 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 { + 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: ®istry, + 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(), + } + } + }) +} diff --git a/desktop/src-tauri/src/broker/ingress.rs b/desktop/src-tauri/src/broker/ingress.rs new file mode 100644 index 0000000000..5e114b1d16 --- /dev/null +++ b/desktop/src-tauri/src/broker/ingress.rs @@ -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) -> 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 { + 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 { + 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::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 { + 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, + /// 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 { + 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; diff --git a/desktop/src-tauri/src/broker/ingress/tests.rs b/desktop/src-tauri/src/broker/ingress/tests.rs new file mode 100644 index 0000000000..be265387ab --- /dev/null +++ b/desktop/src-tauri/src/broker/ingress/tests.rs @@ -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>, +} + +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>, + fail: Option, +} + +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 { + 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: ®istry, + 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()); +} diff --git a/desktop/src-tauri/src/broker/mod.rs b/desktop/src-tauri/src/broker/mod.rs index b5748d9d81..4727b1ad4a 100644 --- a/desktop/src-tauri/src/broker/mod.rs +++ b/desktop/src-tauri/src/broker/mod.rs @@ -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; diff --git a/desktop/src-tauri/src/broker/pipeline.rs b/desktop/src-tauri/src/broker/pipeline.rs index 52716fc6af..583bed49c1 100644 --- a/desktop/src-tauri/src/broker/pipeline.rs +++ b/desktop/src-tauri/src/broker/pipeline.rs @@ -63,6 +63,22 @@ pub struct VerifiedRequest { pub payload: Vec, } +/// 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 diff --git a/desktop/src-tauri/src/broker/store.rs b/desktop/src-tauri/src/broker/store.rs index 3c32fc3478..845c1f5200 100644 --- a/desktop/src-tauri/src/broker/store.rs +++ b/desktop/src-tauri/src/broker/store.rs @@ -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, String> { conn.query_row( "SELECT state, request_digest, capability_version, result_json diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 807f2dbc73..35bca57133 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -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, diff --git a/desktop/src/features/agents/brokerFrameForwarding.test.mjs b/desktop/src/features/agents/brokerFrameForwarding.test.mjs new file mode 100644 index 0000000000..c444a15baf --- /dev/null +++ b/desktop/src/features/agents/brokerFrameForwarding.test.mjs @@ -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(); + } +}); diff --git a/desktop/src/features/agents/brokerFrameForwarding.ts b/desktop/src/features/agents/brokerFrameForwarding.ts new file mode 100644 index 0000000000..b9fcc706b3 --- /dev/null +++ b/desktop/src/features/agents/brokerFrameForwarding.ts @@ -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 { + return invokeTauri("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 { + try { + return await forwardBrokerFrame(event); + } catch (error) { + console.error("Broker frame forwarding failed", error); + return null; + } +} diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 68fa290ad2..60143c422d 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -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;