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;