From b173f46e75c20699116ff46de14b7e71af858579 Mon Sep 17 00:00:00 2001 From: ss-dev-00 Date: Fri, 21 Aug 2026 16:31:29 -0700 Subject: [PATCH] broker: add host authority pipeline, execution log, and agents policy The trusted path lives in one Rust function: verified frame -> authorize -> validate -> claim -> dispatch -> complete It is a single function rather than coordinated steps because every stage boundary is somewhere a caller could otherwise substitute its own answer for "who is asking" or "has this already run", and a crash between two coordinated steps is a side effect nobody recorded. Identity arrives as VerifiedRequest, whose construction asserts that owner, requester, and relay scope came from the verified transport. Nothing in the request body can influence them, so a signer cannot name someone else as owner. A payload that tries fails to deserialize. The digest is computed here, over the exact decrypted bytes, after a size bound. No canonical encoding has to be agreed on and no second implementation can drift from this one. The execution log is keyed by (relay_scope, owner, requester, capability, request_id) and inserts `executing` before the first side effect. Terminal rows replay; a digest mismatch is a conflict, checked before state so a conflicting retry can never be answered with the first request's result; an existing `executing` row becomes `indeterminate` and is never blindly re-executed. Claim runs in an IMMEDIATE transaction -- a test races eight threads to show exactly one winner. This guarantees at-most-once plus indeterminate. Not exactly-once, and not durable completion: the log can prove execution started, never how far it got. Reconciling partial effects belongs to the capability that knows its own phases. Authorization is split from authentication. The broker proves who signed; whether that signer may manage agents is an agents.* scope rule, so a future capability cannot inherit it. The rule is that the requester must be one of this owner's managed agents, read from the authoritative Rust roster. It fails closed when the roster cannot be read, and refuses the owner key itself as a requester. Per the approved design, there is no channel restriction on the target of an update or delete: an owner's agent is theirs regardless of channel, and requiring co-membership would look tighter than it is while adding no real constraint. Channel appears only on create, where attachment needs it. Signed-off-by: ss-dev-00 Co-authored-by: Bradley Axen Signed-off-by: Bradley Axen --- desktop/src-tauri/src/broker/agents_policy.rs | 213 ++++++++ desktop/src-tauri/src/broker/mod.rs | 71 +++ desktop/src-tauri/src/broker/pipeline.rs | 252 +++++++++ .../src-tauri/src/broker/pipeline/tests.rs | 511 ++++++++++++++++++ desktop/src-tauri/src/broker/store.rs | 405 ++++++++++++++ desktop/src-tauri/src/broker/store/tests.rs | 297 ++++++++++ desktop/src-tauri/src/broker/tests.rs | 43 ++ desktop/src-tauri/src/lib.rs | 1 + 8 files changed, 1793 insertions(+) create mode 100644 desktop/src-tauri/src/broker/agents_policy.rs create mode 100644 desktop/src-tauri/src/broker/mod.rs create mode 100644 desktop/src-tauri/src/broker/pipeline.rs create mode 100644 desktop/src-tauri/src/broker/pipeline/tests.rs create mode 100644 desktop/src-tauri/src/broker/store.rs create mode 100644 desktop/src-tauri/src/broker/store/tests.rs create mode 100644 desktop/src-tauri/src/broker/tests.rs diff --git a/desktop/src-tauri/src/broker/agents_policy.rs b/desktop/src-tauri/src/broker/agents_policy.rs new file mode 100644 index 0000000000..ea83ea2faf --- /dev/null +++ b/desktop/src-tauri/src/broker/agents_policy.rs @@ -0,0 +1,213 @@ +//! Authorization policy for the `agents.*` capabilities. +//! +//! # Why this is not broker policy +//! +//! The broker authenticates: it proves who signed a frame and that the frame +//! was sealed for this owner. Whether that signer may *manage agents* is a +//! property of the `agents.*` capability family, not of brokering in general — +//! a future capability with different scope rules must not inherit these. +//! +//! # The rule +//! +//! A requester may manage this owner's agents when the requester **is** one of +//! this owner's managed agents. That is the owner↔requester binding, and it is +//! read from the authoritative Rust roster +//! (`managed_agents::storage::load_managed_agents`), never from anything the +//! caller sent. +//! +//! There is deliberately **no channel restriction on the target** of an update +//! or delete. An owner's agent is the owner's agent regardless of which channel +//! it happens to sit in, and requiring channel co-membership would have made the +//! rule look tighter than it is while adding no real constraint: a requester +//! that can create an agent in a channel could always add its target there +//! first. Channel appears only where the operation genuinely needs it — the +//! attachment on `agents.create`. + +use buzz_sdk_pkg::broker::{BrokerError, BrokerRequest, Capability}; + +use super::pipeline::{Authorizer, VerifiedRequest}; + +/// The set of agents an owner controls, as far as authorization is concerned. +/// +/// A trait so policy can be tested without a Tauri app handle or a real agent +/// store, and so a non-Desktop host can supply its own roster. +pub trait AgentRoster { + /// Pubkeys of every agent this owner manages. + fn agent_pubkeys(&self, owner_pubkey: &str) -> Result, String>; +} + +/// `agents.*` authorization backed by an [`AgentRoster`]. +pub struct AgentsAuthorizer { + roster: R, +} + +impl AgentsAuthorizer { + /// Build an authorizer over `roster`. + pub fn new(roster: R) -> Self { + Self { roster } + } +} + +impl Authorizer for AgentsAuthorizer { + fn authorize( + &self, + request: &VerifiedRequest, + capability: Capability, + _parsed: &BrokerRequest, + ) -> Result<(), BrokerError> { + match capability { + Capability::AgentsCreate | Capability::AgentsUpdate | Capability::AgentsDelete => {} + } + + // The requester must never be able to act as the owner directly: that + // would mean a frame signed by the owner key was treated as a delegated + // request, collapsing the distinction the broker exists to maintain. + if request + .requester_pubkey + .eq_ignore_ascii_case(&request.owner_pubkey) + { + return Err(BrokerError::unauthorized( + "the owner key is not a broker requester", + )); + } + + let agents = self + .roster + .agent_pubkeys(&request.owner_pubkey) + .map_err(|error| { + // A roster we cannot read is a closed door, not an open one. + BrokerError::unauthorized(format!("could not verify agent ownership: {error}")) + })?; + + let bound = agents + .iter() + .any(|pubkey| pubkey.eq_ignore_ascii_case(&request.requester_pubkey)); + + if bound { + Ok(()) + } else { + Err(BrokerError::unauthorized( + "requester is not an agent managed by this owner", + )) + } + } +} + +#[cfg(test)] +mod tests { + use buzz_sdk_pkg::broker::{AgentsCreateArgs, CapabilityArgs}; + + use super::*; + + const OWNER: &str = "1111111111111111111111111111111111111111111111111111111111111111"; + const AGENT: &str = "2222222222222222222222222222222222222222222222222222222222222222"; + const STRANGER: &str = "3333333333333333333333333333333333333333333333333333333333333333"; + const CHANNEL: &str = "b2c38ca8-9ec3-411e-bab5-f9deab34d52e"; + + struct Roster(Result, String>); + impl AgentRoster for Roster { + fn agent_pubkeys(&self, _owner_pubkey: &str) -> Result, String> { + self.0.clone() + } + } + + fn request(requester: &str) -> VerifiedRequest { + VerifiedRequest { + owner_pubkey: OWNER.into(), + requester_pubkey: requester.into(), + relay_scope: "wss://relay.example".into(), + payload: Vec::new(), + } + } + + fn parsed() -> BrokerRequest { + BrokerRequest::new( + "req-1", + CapabilityArgs::AgentsCreate(AgentsCreateArgs { + channel_id: CHANNEL.into(), + display_name: "Helper".into(), + system_prompt: "Help.".into(), + runtime: None, + provider: None, + model: None, + respond_to: None, + }), + ) + .unwrap() + } + + fn authorize( + roster: Result, String>, + requester: &str, + capability: Capability, + ) -> Result<(), BrokerError> { + AgentsAuthorizer::new(Roster(roster)).authorize(&request(requester), capability, &parsed()) + } + + #[test] + fn an_owners_own_agent_may_manage_agents() { + for capability in [ + Capability::AgentsCreate, + Capability::AgentsUpdate, + Capability::AgentsDelete, + ] { + assert!(authorize(Ok(vec![AGENT.into()]), AGENT, capability).is_ok()); + } + } + + #[test] + fn a_signer_who_is_not_this_owners_agent_is_refused() { + let error = + authorize(Ok(vec![AGENT.into()]), STRANGER, Capability::AgentsDelete).unwrap_err(); + assert!( + error.message.contains("not an agent managed by this owner"), + "unexpected: {}", + error.message + ); + } + + #[test] + fn an_empty_roster_refuses_everyone() { + assert!(authorize(Ok(Vec::new()), AGENT, Capability::AgentsCreate).is_err()); + } + + /// The owner key signing its own broker request would erase the delegation + /// boundary the broker exists to enforce. + #[test] + fn the_owner_key_is_not_a_requester() { + let error = authorize(Ok(vec![OWNER.into()]), OWNER, Capability::AgentsCreate).unwrap_err(); + assert!( + error + .message + .contains("owner key is not a broker requester"), + "unexpected: {}", + error.message + ); + } + + /// A roster that cannot be read must fail closed. + #[test] + fn an_unreadable_roster_fails_closed() { + let error = authorize( + Err("keyring locked".into()), + AGENT, + Capability::AgentsUpdate, + ) + .unwrap_err(); + assert!( + error.message.contains("could not verify agent ownership"), + "unexpected: {}", + error.message + ); + } + + #[test] + fn pubkey_comparison_ignores_hex_case() { + assert!(authorize( + Ok(vec![AGENT.to_ascii_uppercase()]), + AGENT, + Capability::AgentsCreate + ) + .is_ok()); + } +} diff --git a/desktop/src-tauri/src/broker/mod.rs b/desktop/src-tauri/src/broker/mod.rs new file mode 100644 index 0000000000..3c5bcbda7b --- /dev/null +++ b/desktop/src-tauri/src/broker/mod.rs @@ -0,0 +1,71 @@ +//! Buzz trusted-operation broker — host implementation. +//! +//! The broker performs a small set of named business operations that need the +//! owner's credentials, on behalf of a requester who cannot hold them. Desktop +//! is host #1; the boundary is drawn so a hosted authority can replace it. +//! +//! # Why this lives entirely in Rust +//! +//! Rust already owns every primitive this depends on: signature verification +//! and NIP-44 decrypt (`commands::identity`), owner key custody +//! (`AppState::signing_keys`), the authoritative agent roster +//! (`managed_agents::storage`), and every agent mutation +//! (`commands::agents`, `commands::agent_models_update`). Routing authority +//! through the frontend would move the trust boundary away from the credentials +//! and split one authority operation across IPC — where the inputs are +//! spoofable and a crash can land between the mutation and its record. The +//! frontend may forward frames and render diagnostics; it is not the security +//! principal. +//! +//! # Guarantee +//! +//! **At-most-once execution plus `indeterminate`** — not exactly-once, and not +//! durable completion. See [`store`] for why, and for what a capability must +//! own to do better. +//! +//! # Delivery is ephemeral +//! +//! Requests arrive as kind-24200 observer frames, which the relay routes to +//! connected subscribers without storing. Execution while the host is offline +//! is therefore unsupported: no queue holds the request, and the requester's +//! wait expires. That is a deliberate limit of this transport, not an oversight. + +use sha2::{Digest, Sha256}; + +pub mod agents_policy; +pub mod pipeline; +pub mod store; + +/// Maximum accepted size of a decrypted broker request payload, in bytes. +/// +/// Bounded before hashing or parsing so an oversized frame cannot force +/// unbounded work. Generous next to the largest legitimate request (a 20k-char +/// system prompt) and far below the observer plaintext ceiling. +pub const MAX_REQUEST_BYTES: usize = 96 * 1024; + +/// Hash the exact decrypted request bytes. +/// +/// The digest is computed **only here**, on the host, from the bytes as +/// received. Nothing re-serializes the request first, so there is no canonical +/// encoding to agree on and no second implementation that could drift from this +/// one. A retry that resends identical bytes hashes identically; a retry that +/// changes anything hashes differently and is refused as a conflict rather than +/// silently answered with the first request's outcome. +/// +/// # Errors +/// +/// Returns an error when the payload exceeds [`MAX_REQUEST_BYTES`]. +pub fn request_digest(payload: &[u8]) -> Result { + if payload.len() > MAX_REQUEST_BYTES { + return Err(format!( + "broker request exceeds {MAX_REQUEST_BYTES} bytes (got {})", + payload.len() + )); + } + let mut hasher = Sha256::new(); + hasher.update(payload); + Ok(hex::encode(hasher.finalize())) +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/broker/pipeline.rs b/desktop/src-tauri/src/broker/pipeline.rs new file mode 100644 index 0000000000..6c88c22a47 --- /dev/null +++ b/desktop/src-tauri/src/broker/pipeline.rs @@ -0,0 +1,252 @@ +//! The broker authority pipeline. +//! +//! One function owns the whole trusted path for a request: +//! +//! ```text +//! verified frame → authorize → validate → claim → dispatch → complete → response +//! ``` +//! +//! It is deliberately a single Rust function rather than a chain of steps +//! coordinated from elsewhere. Every stage boundary is a place where a caller +//! could otherwise substitute its own answer for "who is asking" or "has this +//! already run", and a crash between two coordinated steps is a side effect +//! nobody recorded. +//! +//! # What this module does not decide +//! +//! It does not verify signatures or decrypt frames — that already happened, and +//! its result arrives as [`VerifiedRequest`], which can only be built from +//! transport-derived identity. It does not implement capabilities either; +//! those are [`CapabilityHandler`]s. + +use buzz_sdk_pkg::broker::{ + BrokerError, BrokerErrorCode, BrokerRequest, BrokerResponse, BrokerResult, Capability, + CapabilityOutcome, +}; + +use super::store::{self, ClaimOutcome, ExecutionKey, ExecutionState}; + +/// A request whose frame has already been verified and decrypted. +/// +/// Constructing this type is the assertion that `owner_pubkey`, +/// `requester_pubkey`, and `relay_scope` came from the **verified transport**, +/// not from the request body. Nothing in [`BrokerRequest`] can influence them, +/// which is what prevents a signer from naming someone else as owner. +pub struct VerifiedRequest { + /// Owner whose credentials back this host. + pub owner_pubkey: String, + /// Verified signer of the frame. + pub requester_pubkey: String, + /// Relay the frame arrived on. + pub relay_scope: String, + /// The exact decrypted payload bytes, used for the idempotency digest. + pub payload: Vec, +} + +/// Whether a requester may invoke a capability. +/// +/// Base authentication (a verified signer, a real owner binding) is the +/// broker's job and has happened before this point. This trait is the +/// *capability scope* decision, which is domain policy: channel membership, for +/// instance, is an `agents.*` rule rather than something the broker asserts +/// about every capability it might ever host. +pub trait Authorizer { + /// Authorize `request` for `capability`, or explain the refusal. + fn authorize( + &self, + request: &VerifiedRequest, + capability: Capability, + parsed: &BrokerRequest, + ) -> Result<(), BrokerError>; +} + +/// The outcome of running one capability. +/// +/// `Failed` asserts no side effects persisted. `Indeterminate` asserts nothing: +/// a handler returns it when it cannot tell, and the broker records it so the +/// request is never silently retried. +pub enum HandlerOutcome { + /// Completed, with this outcome. + Succeeded(CapabilityOutcome), + /// Did not complete; no side effects persisted. + Failed(BrokerError), + /// Whether side effects persisted is unknown. + Indeterminate(BrokerError), +} + +/// One broker capability implementation. +pub trait CapabilityHandler { + /// Execute the capability. + /// + /// Called at most once per idempotency key. A handler that mutates in + /// several phases and cannot report which completed must return + /// [`HandlerOutcome::Indeterminate`] rather than guess. + fn execute(&self, request: &VerifiedRequest, parsed: &BrokerRequest) -> HandlerOutcome; +} + +/// Resolves a capability name to its handler. +pub trait CapabilityRegistry { + /// The handler for `capability`, or `None` when this host does not offer it. + fn handler(&self, capability: Capability) -> Option<&dyn CapabilityHandler>; +} + +/// Everything the pipeline needs that is not the request itself. +pub struct BrokerContext<'a> { + /// Capability scope policy. + pub authorizer: &'a dyn Authorizer, + /// Capability implementations. + pub registry: &'a dyn CapabilityRegistry, + /// Durable execution log. + pub conn: &'a mut rusqlite::Connection, + /// Current unix time, injected so tests are not clock-dependent. + pub now: i64, +} + +/// Run one broker request end to end and produce the response to deliver. +/// +/// Returns `Err` only for a host fault that leaves no deliverable response +/// (a database that cannot be reached, say). Every refusal a requester should +/// hear about — invalid, unauthorized, conflicting, interrupted — comes back as +/// an `Ok` response describing it. +pub fn execute( + request: &VerifiedRequest, + ctx: BrokerContext<'_>, +) -> Result { + // Bound and hash the payload before parsing it. The digest is over the + // exact bytes received, so a retry that resends them matches without any + // canonical encoding having to be agreed on. + let digest = match super::request_digest(&request.payload) { + Ok(digest) => digest, + Err(error) => { + return Ok(BrokerResponse::new( + UNKNOWN_REQUEST_ID, + BrokerResult::failed(BrokerError::invalid_request(error)), + )) + } + }; + + let parsed: BrokerRequest = match serde_json::from_slice(&request.payload) { + Ok(parsed) => parsed, + Err(error) => { + // Unparseable: there is no trustworthy request id to echo, and + // inventing one would let a malformed frame address someone + // else's execution record. + return Ok(BrokerResponse::new( + UNKNOWN_REQUEST_ID, + BrokerResult::failed(BrokerError::invalid_request(format!( + "malformed broker request: {error}" + ))), + )); + } + }; + + let request_id = parsed.request_id.clone(); + if let Err(error) = parsed.validate() { + return Ok(BrokerResponse::new( + &request_id, + BrokerResult::failed(BrokerError::invalid_request(error.to_string())), + )); + } + + let capability = parsed.capability(); + + // Authorize before claiming: a refused request must leave no trace that + // could later be replayed as though it had been accepted. + if let Err(error) = ctx.authorizer.authorize(request, capability, &parsed) { + return Ok(BrokerResponse::new( + &request_id, + BrokerResult::failed(error), + )); + } + + let Some(handler) = ctx.registry.handler(capability) else { + return Ok(BrokerResponse::new( + &request_id, + BrokerResult::failed(BrokerError::new( + BrokerErrorCode::UnknownCapability, + format!("this host does not offer {}", capability.as_str()), + )), + )); + }; + + let key = ExecutionKey { + relay_scope: store::normalized_relay_scope(&request.relay_scope).to_string(), + owner_pubkey: request.owner_pubkey.clone(), + requester_pubkey: request.requester_pubkey.clone(), + capability: capability.as_str().to_string(), + request_id: request_id.clone(), + }; + + match store::claim(ctx.conn, &key, &digest, parsed.capability_version, ctx.now)? { + ClaimOutcome::Claimed => {} + ClaimOutcome::Replay(record) => { + return Ok(replay_response(&request_id, record.result_json.as_deref())) + } + ClaimOutcome::Interrupted(_) => { + // A previous attempt started and never reported. Side effects may + // exist, so this is reported rather than retried. + return Ok(BrokerResponse::new( + &request_id, + BrokerResult::indeterminate(BrokerError::new( + BrokerErrorCode::OutcomeUnknown, + "a previous attempt at this request did not complete; \ + its effects are unknown and it will not be retried automatically", + )), + )); + } + ClaimOutcome::DigestConflict { .. } => { + return Ok(BrokerResponse::new( + &request_id, + BrokerResult::failed(BrokerError::new( + BrokerErrorCode::RequestIdConflict, + "this requestId was already used for a different request; \ + retry with the identical payload or choose a new requestId", + )), + )); + } + } + + // The row is now `executing`. Everything after this point must reach + // `complete`, or the next attempt correctly reports `indeterminate`. + let result = match handler.execute(request, &parsed) { + HandlerOutcome::Succeeded(outcome) => BrokerResult::succeeded(outcome), + HandlerOutcome::Failed(error) => BrokerResult::failed(error), + HandlerOutcome::Indeterminate(error) => BrokerResult::indeterminate(error), + }; + + let state = match &result { + BrokerResult::Succeeded { .. } => ExecutionState::Succeeded, + BrokerResult::Failed { .. } => ExecutionState::Failed, + BrokerResult::Indeterminate { .. } => ExecutionState::Indeterminate, + }; + let result_json = serde_json::to_string(&result) + .map_err(|error| format!("failed to encode broker result: {error}"))?; + store::complete(ctx.conn, &key, state, &result_json, ctx.now)?; + + Ok(BrokerResponse::new(&request_id, result)) +} + +/// Request id used when the payload is too malformed to yield one. +const UNKNOWN_REQUEST_ID: &str = "unknown"; + +/// Rebuild a response from a stored terminal result. +/// +/// A row that is terminal but has no decodable result cannot be re-executed — +/// its side effects already happened — so it degrades to `indeterminate`. +fn replay_response(request_id: &str, result_json: Option<&str>) -> BrokerResponse { + let stored = result_json.and_then(|json| serde_json::from_str::(json).ok()); + match stored { + Some(result) => BrokerResponse::new(request_id, result).replayed(), + None => BrokerResponse::new( + request_id, + BrokerResult::indeterminate(BrokerError::new( + BrokerErrorCode::OutcomeUnknown, + "this request already ran, but its recorded outcome could not be read", + )), + ) + .replayed(), + } +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/broker/pipeline/tests.rs b/desktop/src-tauri/src/broker/pipeline/tests.rs new file mode 100644 index 0000000000..70fb18b85d --- /dev/null +++ b/desktop/src-tauri/src/broker/pipeline/tests.rs @@ -0,0 +1,511 @@ +//! Pipeline tests: authority, idempotency, and crash behavior. + +use std::cell::RefCell; + +use buzz_sdk_pkg::broker::{ + AgentTarget, AgentsCreateArgs, AgentsCreateOutcome, AgentsDeleteArgs, AgentsDeleteOutcome, + CapabilityArgs, +}; + +use super::*; + +const CHANNEL: &str = "b2c38ca8-9ec3-411e-bab5-f9deab34d52e"; +const OWNER: &str = "1111111111111111111111111111111111111111111111111111111111111111"; +const REQUESTER: &str = "2222222222222222222222222222222222222222222222222222222222222222"; +const AGENT: &str = "3333333333333333333333333333333333333333333333333333333333333333"; + +fn db() -> rusqlite::Connection { + super::super::store::open_in_memory() +} + +fn create_request(request_id: &str) -> Vec { + let request = 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, + }), + ) + .unwrap(); + serde_json::to_vec(&request).unwrap() +} + +fn verified(payload: Vec) -> VerifiedRequest { + VerifiedRequest { + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + relay_scope: "wss://relay.example".into(), + payload, + } +} + +struct AllowAll; +impl Authorizer for AllowAll { + fn authorize( + &self, + _request: &VerifiedRequest, + _capability: Capability, + _parsed: &BrokerRequest, + ) -> Result<(), BrokerError> { + Ok(()) + } +} + +struct DenyAll; +impl Authorizer for DenyAll { + fn authorize( + &self, + _request: &VerifiedRequest, + _capability: Capability, + _parsed: &BrokerRequest, + ) -> Result<(), BrokerError> { + Err(BrokerError::unauthorized("not a member of that channel")) + } +} + +/// Counts executions so a test can prove a handler ran at most once. +struct CountingHandler { + calls: RefCell, + outcome: fn() -> HandlerOutcome, +} + +impl CountingHandler { + fn new(outcome: fn() -> HandlerOutcome) -> Self { + Self { + calls: RefCell::new(0), + outcome, + } + } +} + +impl CapabilityHandler for CountingHandler { + fn execute(&self, _request: &VerifiedRequest, _parsed: &BrokerRequest) -> HandlerOutcome { + *self.calls.borrow_mut() += 1; + (self.outcome)() + } +} + +struct OneHandler<'a> { + capability: Capability, + handler: &'a dyn CapabilityHandler, +} + +impl CapabilityRegistry for OneHandler<'_> { + fn handler(&self, capability: Capability) -> Option<&dyn CapabilityHandler> { + (capability == self.capability).then_some(self.handler) + } +} + +struct NoHandlers; +impl CapabilityRegistry for NoHandlers { + fn handler(&self, _capability: Capability) -> Option<&dyn CapabilityHandler> { + None + } +} + +fn success() -> HandlerOutcome { + HandlerOutcome::Succeeded(CapabilityOutcome::AgentsCreate(AgentsCreateOutcome { + agent_pubkey: AGENT.into(), + display_name: "Helper".into(), + channel_id: CHANNEL.into(), + })) +} + +fn run<'a>( + request: &VerifiedRequest, + authorizer: &'a dyn Authorizer, + registry: &'a dyn CapabilityRegistry, + conn: &'a mut rusqlite::Connection, +) -> BrokerResponse { + execute( + request, + BrokerContext { + authorizer, + registry, + conn, + now: 1000, + }, + ) + .unwrap() +} + +#[test] +fn a_valid_authorized_request_executes_once_and_returns_its_outcome() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + let response = run(&request, &AllowAll, ®istry, &mut conn); + assert_eq!(response.request_id, "req-1"); + assert!(response.result.is_succeeded()); + assert!(!response.replayed); + assert_eq!(*handler.calls.borrow(), 1); +} + +/// The central idempotency property: replaying the identical request returns +/// the recorded outcome and does not run the handler a second time. +#[test] +fn replaying_an_identical_request_returns_the_record_without_re_executing() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + let first = run(&request, &AllowAll, ®istry, &mut conn); + let second = run(&request, &AllowAll, ®istry, &mut conn); + + assert_eq!(*handler.calls.borrow(), 1, "handler must not run twice"); + assert!(second.replayed); + assert_eq!(first.result, second.result, "replay must be identical"); +} + +/// Same request id, different content: the caller must be told, not handed +/// someone else's outcome. +#[test] +fn a_different_payload_under_the_same_request_id_is_refused() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + + run( + &verified(create_request("req-1")), + &AllowAll, + ®istry, + &mut conn, + ); + + let mut different = BrokerRequest::new( + "req-1", + CapabilityArgs::AgentsCreate(AgentsCreateArgs { + channel_id: CHANNEL.into(), + display_name: "Different agent".into(), + system_prompt: "Help.".into(), + runtime: None, + provider: None, + model: None, + respond_to: None, + }), + ) + .unwrap(); + different.request_id = "req-1".into(); + let response = run( + &verified(serde_json::to_vec(&different).unwrap()), + &AllowAll, + ®istry, + &mut conn, + ); + + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::RequestIdConflict + ); + assert_eq!(*handler.calls.borrow(), 1, "the conflict must not execute"); +} + +/// An unauthorized request must not leave a claim behind: otherwise a later +/// authorized retry would be answered from a refusal. +#[test] +fn an_unauthorized_request_does_not_execute_or_leave_a_claim() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + let denied = run(&request, &DenyAll, ®istry, &mut conn); + assert_eq!( + denied.result.error().unwrap().code, + BrokerErrorCode::Unauthorized + ); + assert_eq!(*handler.calls.borrow(), 0); + + // The same request, now authorized, still runs normally. + let allowed = run(&request, &AllowAll, ®istry, &mut conn); + assert!(allowed.result.is_succeeded()); + assert!(!allowed.replayed); + assert_eq!(*handler.calls.borrow(), 1); +} + +/// A crash between the mutation and its record must surface as indeterminate, +/// never as a silent re-run. +#[test] +fn an_interrupted_execution_reports_indeterminate_and_never_re_executes() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + // Simulate a process that claimed the key and died before completing. + let key = ExecutionKey { + relay_scope: "wss://relay.example".into(), + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + capability: "agents.create".into(), + request_id: "req-1".into(), + }; + let digest = super::super::request_digest(&request.payload).unwrap(); + store::claim(&mut conn, &key, &digest, 1, 500).unwrap(); + + let response = run(&request, &AllowAll, ®istry, &mut conn); + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::OutcomeUnknown + ); + assert_eq!( + *handler.calls.borrow(), + 0, + "an interrupted request must not be re-executed" + ); +} + +#[test] +fn a_failed_handler_is_recorded_and_replayed_as_failed() { + fn failure() -> HandlerOutcome { + HandlerOutcome::Failed(BrokerError::new( + BrokerErrorCode::CapabilityFailed, + "runtime not installed", + )) + } + let mut conn = db(); + let handler = CountingHandler::new(failure); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + let first = run(&request, &AllowAll, ®istry, &mut conn); + assert_eq!( + first.result.error().unwrap().code, + BrokerErrorCode::CapabilityFailed + ); + + // A failure is terminal: retrying replays it rather than re-running. + let second = run(&request, &AllowAll, ®istry, &mut conn); + assert!(second.replayed); + assert_eq!(first.result, second.result); + assert_eq!(*handler.calls.borrow(), 1); +} + +#[test] +fn an_indeterminate_handler_outcome_is_recorded_as_indeterminate() { + fn unknown() -> HandlerOutcome { + HandlerOutcome::Indeterminate(BrokerError::new( + BrokerErrorCode::OutcomeUnknown, + "published but could not confirm", + )) + } + let mut conn = db(); + let handler = CountingHandler::new(unknown); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let request = verified(create_request("req-1")); + + let response = run(&request, &AllowAll, ®istry, &mut conn); + assert!(matches!( + response.result, + BrokerResult::Indeterminate { .. } + )); + assert_eq!(*handler.calls.borrow(), 1); +} + +#[test] +fn a_malformed_payload_is_refused_without_a_claim() { + let mut conn = db(); + let response = run( + &verified(b"not json".to_vec()), + &AllowAll, + &NoHandlers, + &mut conn, + ); + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::InvalidRequest + ); +} + +#[test] +fn an_oversized_payload_is_refused() { + let mut conn = db(); + let response = run( + &verified(vec![b'x'; super::super::MAX_REQUEST_BYTES + 1]), + &AllowAll, + &NoHandlers, + &mut conn, + ); + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::InvalidRequest + ); +} + +#[test] +fn a_capability_this_host_does_not_offer_is_refused() { + let mut conn = db(); + let response = run( + &verified(create_request("req-1")), + &AllowAll, + &NoHandlers, + &mut conn, + ); + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::UnknownCapability + ); +} + +/// Two requesters using the same request id are separate executions: the key +/// includes the verified requester, so one cannot read the other's outcome. +#[test] +fn the_same_request_id_from_two_requesters_are_separate_executions() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + + let payload = create_request("req-1"); + let first = verified(payload.clone()); + let second = VerifiedRequest { + requester_pubkey: AGENT.into(), + ..verified(payload) + }; + + assert!(!run(&first, &AllowAll, ®istry, &mut conn).replayed); + assert!( + !run(&second, &AllowAll, ®istry, &mut conn).replayed, + "a different requester must not replay another's record" + ); + assert_eq!(*handler.calls.borrow(), 2); +} + +/// The pipeline must take identity from the verified transport, never from the +/// payload. A body claiming another owner cannot even deserialize. +#[test] +fn identity_in_the_payload_cannot_override_transport_identity() { + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + + let mut json: serde_json::Value = serde_json::from_slice(&create_request("req-1")).unwrap(); + json.as_object_mut() + .unwrap() + .insert("ownerPubkey".into(), serde_json::json!(AGENT)); + + let response = run( + &verified(serde_json::to_vec(&json).unwrap()), + &AllowAll, + ®istry, + &mut conn, + ); + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::InvalidRequest + ); + assert_eq!(*handler.calls.borrow(), 0); +} + +/// The authorizer sees the transport-derived identity, so policy is written +/// against who actually signed. +#[test] +fn the_authorizer_receives_transport_derived_identity() { + struct Recording(RefCell>); + impl Authorizer for Recording { + fn authorize( + &self, + request: &VerifiedRequest, + capability: Capability, + _parsed: &BrokerRequest, + ) -> Result<(), BrokerError> { + self.0.borrow_mut().push(( + request.owner_pubkey.clone(), + request.requester_pubkey.clone(), + capability.as_str().to_string(), + )); + Ok(()) + } + } + + let mut conn = db(); + let handler = CountingHandler::new(success); + let registry = OneHandler { + capability: Capability::AgentsCreate, + handler: &handler, + }; + let authorizer = Recording(RefCell::new(Vec::new())); + + run( + &verified(create_request("req-1")), + &authorizer, + ®istry, + &mut conn, + ); + + assert_eq!( + authorizer.0.borrow().as_slice(), + &[( + OWNER.to_string(), + REQUESTER.to_string(), + "agents.create".to_string() + )] + ); +} + +#[test] +fn delete_requests_route_to_the_delete_capability() { + fn deleted() -> HandlerOutcome { + HandlerOutcome::Succeeded(CapabilityOutcome::AgentsDelete(AgentsDeleteOutcome { + agent_pubkey: AGENT.into(), + display_name: "Gone".into(), + })) + } + let mut conn = db(); + let handler = CountingHandler::new(deleted); + let registry = OneHandler { + capability: Capability::AgentsDelete, + handler: &handler, + }; + + let request = BrokerRequest::new( + "req-del", + CapabilityArgs::AgentsDelete(AgentsDeleteArgs { + target: AgentTarget::Pubkey(AGENT.into()), + }), + ) + .unwrap(); + let response = run( + &verified(serde_json::to_vec(&request).unwrap()), + &AllowAll, + ®istry, + &mut conn, + ); + + assert!(response.result.is_succeeded()); + assert_eq!(*handler.calls.borrow(), 1); +} diff --git a/desktop/src-tauri/src/broker/store.rs b/desktop/src-tauri/src/broker/store.rs new file mode 100644 index 0000000000..aa417ebb8b --- /dev/null +++ b/desktop/src-tauri/src/broker/store.rs @@ -0,0 +1,405 @@ +//! Durable execution log for broker requests. +//! +//! One row per `(relay_scope, owner, requester, capability, request_id)`. The +//! row is inserted in state `executing` **before the first side effect**, so a +//! request that crashes mid-flight leaves evidence behind rather than looking +//! like it never ran. +//! +//! # What this buys, and what it does not +//! +//! This log delivers **at-most-once execution plus `indeterminate`**. It does +//! not deliver exactly-once or durable completion. A crash between a mutation +//! and the recording of its result is not recoverable from here — the log can +//! prove that execution *started*, never how far it got. Reconciling partial +//! side effects is the owning capability's job, because only it knows what its +//! phases were. +//! +//! WAL + `busy_timeout=5000` matches `managed_agents/retention.rs` and +//! `archive/store.rs`. + +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use rusqlite::{params, Connection, OptionalExtension}; +use sha2::{Digest, Sha256}; + +/// Terminal or in-flight state of a broker request. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ExecutionState { + /// A handler was dispatched and has not reported back. + Executing, + /// The handler completed successfully. + Succeeded, + /// The handler failed without leaving side effects. + Failed, + /// Whether side effects took hold is unknown. + Indeterminate, +} + +impl ExecutionState { + /// Stable database encoding. + pub fn as_str(self) -> &'static str { + match self { + Self::Executing => "executing", + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::Indeterminate => "indeterminate", + } + } + + /// Parse the database encoding. + fn parse(value: &str) -> Result { + match value { + "executing" => Ok(Self::Executing), + "succeeded" => Ok(Self::Succeeded), + "failed" => Ok(Self::Failed), + "indeterminate" => Ok(Self::Indeterminate), + other => Err(format!("unknown broker execution state \"{other}\"")), + } + } + + /// Whether this state is final. + pub fn is_terminal(self) -> bool { + !matches!(self, Self::Executing) + } +} + +/// Identity of one broker request, as derived by the host. +/// +/// Owner, requester, and relay scope come from the verified frame — never from +/// the request body — so a caller cannot address another owner's execution log. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExecutionKey { + /// Relay the request arrived on, normalized. + pub relay_scope: String, + /// Owner whose credentials would be used. + pub owner_pubkey: String, + /// Verified signer of the request frame. + pub requester_pubkey: String, + /// Capability wire name. + pub capability: String, + /// Caller-chosen idempotency key. + pub request_id: String, +} + +/// A recorded execution row. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExecutionRecord { + /// Current state. + pub state: ExecutionState, + /// Digest of the request that claimed this key. + pub request_digest: String, + /// Capability contract version that claimed this key. + pub capability_version: u16, + /// Serialized terminal result, absent while `executing`. + pub result_json: Option, +} + +/// What a claim attempt established. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ClaimOutcome { + /// This caller owns the execution and must proceed. + Claimed, + /// A terminal result already exists; replay it without re-executing. + Replay(ExecutionRecord), + /// A previous attempt started and never finished. + /// + /// The row has been moved to `indeterminate` and must not be re-executed: + /// side effects may already exist. + Interrupted(ExecutionRecord), + /// The key was claimed by a request with different content. + DigestConflict { + /// Digest recorded by the first request. + expected: String, + }, +} + +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS broker_executions ( + relay_scope TEXT NOT NULL, + owner_pubkey TEXT NOT NULL, + requester_pubkey TEXT NOT NULL, + capability TEXT NOT NULL, + request_id TEXT NOT NULL, + request_digest TEXT NOT NULL, + capability_version INTEGER NOT NULL, + state TEXT NOT NULL, + result_json TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (relay_scope, owner_pubkey, requester_pubkey, capability, request_id) +); +"; + +/// Relay-URL form that identifies a broker scope. +/// +/// Mirrors `managed_agents::retention`: equivalent workspace URLs (surrounding +/// space, trailing slash) must resolve to one scope, so "same relay" can never +/// disagree with "same database". +pub fn normalized_relay_scope(relay_url: &str) -> &str { + relay_url.trim().trim_end_matches('/') +} + +/// Resolve the broker database path for a relay + owner pair. +/// +/// The normalized scope is hashed so relay URLs never become path components. +pub fn scoped_broker_db_path(base_dir: &Path, relay_url: &str, owner_pubkey: &str) -> PathBuf { + let normalized_relay = normalized_relay_scope(relay_url); + let mut hasher = Sha256::new(); + hasher.update(owner_pubkey.trim().to_ascii_lowercase().as_bytes()); + hasher.update(b"\0"); + hasher.update(normalized_relay.as_bytes()); + let scope_id = hex::encode(hasher.finalize()); + base_dir.join("broker").join(format!("{scope_id}.db")) +} + +/// Open (or create) the broker execution database. +pub fn open_broker_db(path: &Path) -> Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| format!("failed to create broker dir: {e}"))?; + } + + let conn = Connection::open(path).map_err(|e| format!("failed to open broker db: {e}"))?; + + conn.pragma_update(None, "busy_timeout", 5000) + .map_err(|e| format!("failed to set busy_timeout: {e}"))?; + set_wal_mode(&conn)?; + + conn.execute_batch(SCHEMA) + .map_err(|e| format!("failed to initialize broker schema: {e}"))?; + + Ok(conn) +} + +fn set_wal_mode(conn: &Connection) -> Result<(), String> { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + match conn.pragma_update(None, "journal_mode", "WAL") { + Ok(()) => return Ok(()), + Err(error) if sqlite_is_busy(&error) && Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(25)); + } + Err(error) => return Err(format!("failed to set WAL mode: {error}")), + } + } +} + +fn sqlite_is_busy(error: &rusqlite::Error) -> bool { + matches!( + error, + rusqlite::Error::SqliteFailure(inner, _) + if matches!( + inner.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + ) + ) +} + +/// Claim `key` for execution, or report why it cannot be claimed. +/// +/// The insert and the read happen in one `IMMEDIATE` transaction, so two +/// concurrent frames carrying the same request cannot both believe they claimed +/// it. This is the only function that may transition a row into `executing`, +/// and it never returns [`ClaimOutcome::Claimed`] for a key that already has a +/// row. +/// +/// An existing `executing` row is treated as interrupted: it is moved to +/// `indeterminate` and returned. Re-executing it would risk duplicating side +/// effects that the log cannot see. +pub fn claim( + conn: &mut Connection, + key: &ExecutionKey, + request_digest: &str, + capability_version: u16, + now: i64, +) -> Result { + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|e| format!("failed to begin broker claim: {e}"))?; + + let existing = tx + .query_row( + "SELECT state, request_digest, capability_version, result_json + FROM broker_executions + WHERE relay_scope = ?1 AND owner_pubkey = ?2 AND requester_pubkey = ?3 + AND capability = ?4 AND request_id = ?5", + params![ + key.relay_scope, + key.owner_pubkey, + key.requester_pubkey, + key.capability, + key.request_id + ], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, Option>(3)?, + )) + }, + ) + .optional() + .map_err(|e| format!("failed to read broker execution: {e}"))?; + + let outcome = match existing { + None => { + tx.execute( + "INSERT INTO broker_executions ( + relay_scope, owner_pubkey, requester_pubkey, capability, request_id, + request_digest, capability_version, state, result_json, + created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?9)", + params![ + key.relay_scope, + key.owner_pubkey, + key.requester_pubkey, + key.capability, + key.request_id, + request_digest, + i64::from(capability_version), + ExecutionState::Executing.as_str(), + now + ], + ) + .map_err(|e| format!("failed to claim broker execution: {e}"))?; + ClaimOutcome::Claimed + } + Some((state, digest, version, result_json)) => { + // A different payload under the same id is a caller bug, and the + // recorded outcome does not describe it. Checked before state so a + // conflicting retry can never be answered with someone else's result. + if digest != request_digest { + ClaimOutcome::DigestConflict { expected: digest } + } else { + let state = ExecutionState::parse(&state)?; + let version = u16::try_from(version) + .map_err(|_| format!("stored capability_version {version} out of range"))?; + let record = ExecutionRecord { + state, + request_digest: digest, + capability_version: version, + result_json, + }; + if state.is_terminal() { + ClaimOutcome::Replay(record) + } else { + tx.execute( + "UPDATE broker_executions SET state = ?1, updated_at = ?2 + WHERE relay_scope = ?3 AND owner_pubkey = ?4 + AND requester_pubkey = ?5 AND capability = ?6 + AND request_id = ?7", + params![ + ExecutionState::Indeterminate.as_str(), + now, + key.relay_scope, + key.owner_pubkey, + key.requester_pubkey, + key.capability, + key.request_id + ], + ) + .map_err(|e| format!("failed to mark broker execution interrupted: {e}"))?; + ClaimOutcome::Interrupted(ExecutionRecord { + state: ExecutionState::Indeterminate, + ..record + }) + } + } + } + }; + + tx.commit() + .map_err(|e| format!("failed to commit broker claim: {e}"))?; + Ok(outcome) +} + +/// Record the terminal state and result for a claimed key. +/// +/// Rejects a non-terminal state: only [`claim`] may write `executing`. +pub fn complete( + conn: &Connection, + key: &ExecutionKey, + state: ExecutionState, + result_json: &str, + now: i64, +) -> Result<(), String> { + if !state.is_terminal() { + return Err(format!( + "cannot complete a broker execution as \"{}\"", + state.as_str() + )); + } + let changed = conn + .execute( + "UPDATE broker_executions + SET state = ?1, result_json = ?2, updated_at = ?3 + WHERE relay_scope = ?4 AND owner_pubkey = ?5 AND requester_pubkey = ?6 + AND capability = ?7 AND request_id = ?8", + params![ + state.as_str(), + result_json, + now, + key.relay_scope, + key.owner_pubkey, + key.requester_pubkey, + key.capability, + key.request_id + ], + ) + .map_err(|e| format!("failed to complete broker execution: {e}"))?; + if changed == 0 { + return Err("broker execution row is missing; refusing to record a result".into()); + } + Ok(()) +} + +/// Open an in-memory execution log with the real schema. +/// +/// For tests that need the real claim/complete behavior without a temp file. +#[cfg(test)] +pub fn open_in_memory() -> Connection { + let conn = Connection::open_in_memory().expect("in-memory sqlite"); + conn.execute_batch(SCHEMA).expect("broker schema"); + conn +} + +/// Read one execution row, if it exists. +pub fn get(conn: &Connection, key: &ExecutionKey) -> Result, String> { + conn.query_row( + "SELECT state, request_digest, capability_version, result_json + FROM broker_executions + WHERE relay_scope = ?1 AND owner_pubkey = ?2 AND requester_pubkey = ?3 + AND capability = ?4 AND request_id = ?5", + params![ + key.relay_scope, + key.owner_pubkey, + key.requester_pubkey, + key.capability, + key.request_id + ], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, Option>(3)?, + )) + }, + ) + .optional() + .map_err(|e| format!("failed to read broker execution: {e}"))? + .map(|(state, request_digest, version, result_json)| { + Ok(ExecutionRecord { + state: ExecutionState::parse(&state)?, + request_digest, + capability_version: u16::try_from(version) + .map_err(|_| format!("stored capability_version {version} out of range"))?, + result_json, + }) + }) + .transpose() +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/broker/store/tests.rs b/desktop/src-tauri/src/broker/store/tests.rs new file mode 100644 index 0000000000..9157d28af1 --- /dev/null +++ b/desktop/src-tauri/src/broker/store/tests.rs @@ -0,0 +1,297 @@ +//! Durability and idempotency tests for the broker execution log. + +use super::*; + +const RELAY: &str = "wss://relay.example"; +const OWNER: &str = "1111111111111111111111111111111111111111111111111111111111111111"; +const REQUESTER: &str = "2222222222222222222222222222222222222222222222222222222222222222"; +const OTHER: &str = "3333333333333333333333333333333333333333333333333333333333333333"; +const DIGEST: &str = "aaaa"; + +fn db() -> Connection { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch(SCHEMA).unwrap(); + conn +} + +fn key() -> ExecutionKey { + ExecutionKey { + relay_scope: RELAY.into(), + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + capability: "agents.create".into(), + request_id: "req-1".into(), + } +} + +#[test] +fn first_claim_wins_and_records_executing_before_any_side_effect() { + let mut conn = db(); + assert_eq!( + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(), + ClaimOutcome::Claimed + ); + + // The row exists and is `executing` even though nothing has completed: + // that is what makes a crash visible instead of invisible. + let record = get(&conn, &key()).unwrap().unwrap(); + assert_eq!(record.state, ExecutionState::Executing); + assert_eq!(record.request_digest, DIGEST); + assert!(record.result_json.is_none()); +} + +#[test] +fn terminal_row_replays_the_recorded_result_instead_of_re_executing() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + complete( + &conn, + &key(), + ExecutionState::Succeeded, + r#"{"status":"succeeded"}"#, + 200, + ) + .unwrap(); + + match claim(&mut conn, &key(), DIGEST, 1, 300).unwrap() { + ClaimOutcome::Replay(record) => { + assert_eq!(record.state, ExecutionState::Succeeded); + assert_eq!(record.result_json.unwrap(), r#"{"status":"succeeded"}"#); + } + other => panic!("expected replay, got {other:?}"), + } +} + +#[test] +fn a_failed_result_also_replays_rather_than_retrying() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + complete( + &conn, + &key(), + ExecutionState::Failed, + r#"{"status":"failed"}"#, + 200, + ) + .unwrap(); + + match claim(&mut conn, &key(), DIGEST, 1, 300).unwrap() { + ClaimOutcome::Replay(record) => assert_eq!(record.state, ExecutionState::Failed), + other => panic!("expected replay, got {other:?}"), + } +} + +/// The core safety property: a request that started and never finished must +/// never be blindly re-executed, because side effects may already exist. +#[test] +fn an_interrupted_execution_becomes_indeterminate_and_is_never_re_executed() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + // No `complete` — simulates a crash mid-execution. + + match claim(&mut conn, &key(), DIGEST, 1, 200).unwrap() { + ClaimOutcome::Interrupted(record) => { + assert_eq!(record.state, ExecutionState::Indeterminate); + } + other => panic!("expected interrupted, got {other:?}"), + } + assert_eq!( + get(&conn, &key()).unwrap().unwrap().state, + ExecutionState::Indeterminate + ); + + // And it stays indeterminate: a third attempt still refuses to run. + match claim(&mut conn, &key(), DIGEST, 1, 300).unwrap() { + ClaimOutcome::Replay(record) => { + assert_eq!(record.state, ExecutionState::Indeterminate) + } + other => panic!("expected replay of indeterminate, got {other:?}"), + } +} + +#[test] +fn same_request_id_with_different_content_is_a_conflict() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + + match claim(&mut conn, &key(), "bbbb", 1, 200).unwrap() { + ClaimOutcome::DigestConflict { expected } => assert_eq!(expected, DIGEST), + other => panic!("expected conflict, got {other:?}"), + } +} + +/// A conflicting retry must never be answered with the first request's result. +#[test] +fn digest_conflict_is_reported_even_when_a_result_already_exists() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + complete( + &conn, + &key(), + ExecutionState::Succeeded, + r#"{"status":"succeeded"}"#, + 200, + ) + .unwrap(); + + match claim(&mut conn, &key(), "different", 1, 300).unwrap() { + ClaimOutcome::DigestConflict { expected } => assert_eq!(expected, DIGEST), + other => panic!("a conflicting payload must not receive the stored result: {other:?}"), + } +} + +/// The key includes requester and owner, so one requester's id cannot collide +/// with another's — or leak another's recorded outcome. +#[test] +fn the_key_separates_requesters_owners_relays_and_capabilities() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + + let variants = [ + ExecutionKey { + requester_pubkey: OTHER.into(), + ..key() + }, + ExecutionKey { + owner_pubkey: OTHER.into(), + ..key() + }, + ExecutionKey { + relay_scope: "wss://other.example".into(), + ..key() + }, + ExecutionKey { + capability: "agents.delete".into(), + ..key() + }, + ]; + for variant in variants { + assert_eq!( + claim(&mut conn, &variant, DIGEST, 1, 100).unwrap(), + ClaimOutcome::Claimed, + "{variant:?} must be a distinct execution" + ); + } +} + +#[test] +fn complete_refuses_a_non_terminal_state() { + let mut conn = db(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + let error = complete(&conn, &key(), ExecutionState::Executing, "{}", 200).unwrap_err(); + assert!(error.contains("cannot complete"), "unexpected: {error}"); +} + +#[test] +fn complete_refuses_an_unclaimed_key() { + let conn = db(); + let error = complete(&conn, &key(), ExecutionState::Succeeded, "{}", 200).unwrap_err(); + assert!(error.contains("missing"), "unexpected: {error}"); +} + +#[test] +fn states_round_trip_through_their_database_encoding() { + for state in [ + ExecutionState::Executing, + ExecutionState::Succeeded, + ExecutionState::Failed, + ExecutionState::Indeterminate, + ] { + assert_eq!(ExecutionState::parse(state.as_str()).unwrap(), state); + } + assert!(ExecutionState::parse("bogus").is_err()); + assert!(!ExecutionState::Executing.is_terminal()); + for terminal in [ + ExecutionState::Succeeded, + ExecutionState::Failed, + ExecutionState::Indeterminate, + ] { + assert!(terminal.is_terminal()); + } +} + +/// Equivalent relay URLs must resolve to one scope, so "same relay" cannot +/// disagree with "same database". +#[test] +fn relay_scope_and_db_path_normalize_equivalent_urls() { + assert_eq!( + normalized_relay_scope(" wss://r.example/ "), + "wss://r.example" + ); + + let base = Path::new("/tmp/nest"); + let a = scoped_broker_db_path(base, "wss://r.example", OWNER); + let b = scoped_broker_db_path(base, " wss://r.example/ ", &OWNER.to_ascii_uppercase()); + assert_eq!(a, b); + + // Different owner or relay must not share a database. + assert_ne!(a, scoped_broker_db_path(base, "wss://r.example", OTHER)); + assert_ne!(a, scoped_broker_db_path(base, "wss://other.example", OWNER)); + + // The relay URL never becomes a path component. + assert!(!a.to_string_lossy().contains("r.example")); +} + +#[test] +fn opening_the_db_twice_is_idempotent_and_preserves_rows() { + let dir = tempfile::tempdir().unwrap(); + let path = scoped_broker_db_path(dir.path(), RELAY, OWNER); + + let mut conn = open_broker_db(&path).unwrap(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap(); + complete(&conn, &key(), ExecutionState::Succeeded, "{}", 200).unwrap(); + drop(conn); + + let conn = open_broker_db(&path).unwrap(); + assert_eq!( + get(&conn, &key()).unwrap().unwrap().state, + ExecutionState::Succeeded + ); +} + +/// Two racing frames carrying the same request must not both believe they +/// claimed it — exactly one executes, the other is told to stand down. +#[test] +fn concurrent_claims_produce_exactly_one_winner() { + use std::sync::{Arc, Barrier}; + + let dir = tempfile::tempdir().unwrap(); + let path = scoped_broker_db_path(dir.path(), RELAY, OWNER); + // Materialize the schema before the threads race on it. + drop(open_broker_db(&path).unwrap()); + + let threads = 8; + let barrier = Arc::new(Barrier::new(threads)); + let handles: Vec<_> = (0..threads) + .map(|_| { + let path = path.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + let mut conn = open_broker_db(&path).unwrap(); + barrier.wait(); + claim(&mut conn, &key(), DIGEST, 1, 100).unwrap() + }) + }) + .collect(); + + let outcomes: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect(); + let claimed = outcomes + .iter() + .filter(|o| matches!(o, ClaimOutcome::Claimed)) + .count(); + assert_eq!(claimed, 1, "exactly one claim must win: {outcomes:?}"); + + // Every loser saw the in-flight row rather than a free slot. + for outcome in outcomes + .iter() + .filter(|o| !matches!(o, ClaimOutcome::Claimed)) + { + assert!( + matches!( + outcome, + ClaimOutcome::Interrupted(_) | ClaimOutcome::Replay(_) + ), + "unexpected loser outcome: {outcome:?}" + ); + } +} diff --git a/desktop/src-tauri/src/broker/tests.rs b/desktop/src-tauri/src/broker/tests.rs new file mode 100644 index 0000000000..2ec0f04562 --- /dev/null +++ b/desktop/src-tauri/src/broker/tests.rs @@ -0,0 +1,43 @@ +//! Digest tests. + +use super::*; + +#[test] +fn identical_bytes_hash_identically_and_any_change_does_not() { + let payload = br#"{"type":"broker_request","requestId":"req-1"}"#; + assert_eq!( + request_digest(payload).unwrap(), + request_digest(payload).unwrap() + ); + + // Byte-level difference, including key order, is a different request. + let reordered = br#"{"requestId":"req-1","type":"broker_request"}"#; + assert_ne!( + request_digest(payload).unwrap(), + request_digest(reordered).unwrap() + ); +} + +#[test] +fn digest_is_hex_sha256_of_the_exact_bytes() { + // Pinned against an independent SHA-256 of the empty input, so a change to + // the hashing scheme fails loudly rather than silently invalidating every + // recorded digest. + assert_eq!( + request_digest(b"").unwrap(), + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ); + assert_eq!( + request_digest(b"abc").unwrap(), + "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" + ); +} + +#[test] +fn an_oversized_payload_is_refused_before_hashing() { + let too_big = vec![b'x'; MAX_REQUEST_BYTES + 1]; + let error = request_digest(&too_big).unwrap_err(); + assert!(error.contains("exceeds"), "unexpected: {error}"); + + assert!(request_digest(&vec![b'x'; MAX_REQUEST_BYTES]).is_ok()); +} diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 71a5eb3806..807f2dbc73 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -2,6 +2,7 @@ mod app_menu; mod app_state; mod archive; +mod broker; mod builderlab; mod commands; mod deep_link;