diff --git a/crates/buzz-core/src/observer.rs b/crates/buzz-core/src/observer.rs index 8347bda054..d9ec08c60b 100644 --- a/crates/buzz-core/src/observer.rs +++ b/crates/buzz-core/src/observer.rs @@ -80,11 +80,22 @@ pub fn encrypt_observer_payload( Ok(encrypted) } -/// NIP-44 decrypt and deserialize an observer payload from `event`. -pub fn decrypt_observer_payload( +/// NIP-44 decrypt an observer payload from `event` and return the plaintext +/// JSON **verbatim**. +/// +/// Callers that only need a typed value should prefer +/// [`decrypt_observer_payload`]. This exists for the one caller that must hash +/// the bytes it received — the trusted-operation broker's idempotency digest. +/// Deserializing and re-serializing first would make that digest depend on a +/// canonical encoding both ends have to agree on; hashing the plaintext as it +/// arrived needs no such agreement. +/// +/// The returned `String` is the caller's to zeroize if it is sensitive; the +/// intermediate copies held here are wiped before return. +pub fn decrypt_observer_plaintext( recipient_keys: &Keys, event: &Event, -) -> Result { +) -> Result { if !content_looks_like_nip44(&event.content) { return Err(ObserverPayloadError::InvalidCiphertextLength( event.content.len(), @@ -104,7 +115,15 @@ pub fn decrypt_observer_payload( got, }); } + Ok(plaintext) +} +/// NIP-44 decrypt and deserialize an observer payload from `event`. +pub fn decrypt_observer_payload( + recipient_keys: &Keys, + event: &Event, +) -> Result { + let mut plaintext = decrypt_observer_plaintext(recipient_keys, event)?; let result = serde_json::from_str(&plaintext); plaintext.zeroize(); Ok(result?) diff --git a/desktop/src-tauri/src/broker/agents_policy.rs b/desktop/src-tauri/src/broker/agents_policy.rs index ea83ea2faf..a634347b23 100644 --- a/desktop/src-tauri/src/broker/agents_policy.rs +++ b/desktop/src-tauri/src/broker/agents_policy.rs @@ -25,15 +25,18 @@ use buzz_sdk_pkg::broker::{BrokerError, BrokerRequest, Capability}; -use super::pipeline::{Authorizer, VerifiedRequest}; +use super::pipeline::{Authorizer, BoxFuture, 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 { +pub trait AgentRoster: Send + Sync { /// Pubkeys of every agent this owner manages. - fn agent_pubkeys(&self, owner_pubkey: &str) -> Result, String>; + fn agent_pubkeys<'a>( + &'a self, + owner_pubkey: &'a str, + ) -> BoxFuture<'a, Result, String>>; } /// `agents.*` authorization backed by an [`AgentRoster`]. @@ -49,47 +52,51 @@ impl AgentsAuthorizer { } impl Authorizer for AgentsAuthorizer { - fn authorize( - &self, - request: &VerifiedRequest, + fn authorize<'a>( + &'a self, + request: &'a VerifiedRequest, capability: Capability, - _parsed: &BrokerRequest, - ) -> Result<(), BrokerError> { - match capability { - Capability::AgentsCreate | Capability::AgentsUpdate | Capability::AgentsDelete => {} - } + _parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, Result<(), BrokerError>> { + Box::pin(async move { + 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", - )); - } + // 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 agents = self + .roster + .agent_pubkeys(&request.owner_pubkey) + .await + .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)); + 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", - )) - } + if bound { + Ok(()) + } else { + Err(BrokerError::unauthorized( + "requester is not an agent managed by this owner", + )) + } + }) } } @@ -106,8 +113,12 @@ mod tests { struct Roster(Result, String>); impl AgentRoster for Roster { - fn agent_pubkeys(&self, _owner_pubkey: &str) -> Result, String> { - self.0.clone() + fn agent_pubkeys<'a>( + &'a self, + _owner_pubkey: &'a str, + ) -> BoxFuture<'a, Result, String>> { + let result = self.0.clone(); + Box::pin(async move { result }) } } @@ -136,29 +147,35 @@ mod tests { .unwrap() } - fn authorize( + async fn authorize( roster: Result, String>, requester: &str, capability: Capability, ) -> Result<(), BrokerError> { - AgentsAuthorizer::new(Roster(roster)).authorize(&request(requester), capability, &parsed()) + let authorizer = AgentsAuthorizer::new(Roster(roster)); + let request = request(requester); + let parsed = parsed(); + authorizer.authorize(&request, capability, &parsed).await } - #[test] - fn an_owners_own_agent_may_manage_agents() { + #[tokio::test] + async 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()); + assert!(authorize(Ok(vec![AGENT.into()]), AGENT, capability) + .await + .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(); + #[tokio::test] + async fn a_signer_who_is_not_this_owners_agent_is_refused() { + let error = authorize(Ok(vec![AGENT.into()]), STRANGER, Capability::AgentsDelete) + .await + .unwrap_err(); assert!( error.message.contains("not an agent managed by this owner"), "unexpected: {}", @@ -166,16 +183,20 @@ mod tests { ); } - #[test] - fn an_empty_roster_refuses_everyone() { - assert!(authorize(Ok(Vec::new()), AGENT, Capability::AgentsCreate).is_err()); + #[tokio::test] + async fn an_empty_roster_refuses_everyone() { + assert!(authorize(Ok(Vec::new()), AGENT, Capability::AgentsCreate) + .await + .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(); + #[tokio::test] + async fn the_owner_key_is_not_a_requester() { + let error = authorize(Ok(vec![OWNER.into()]), OWNER, Capability::AgentsCreate) + .await + .unwrap_err(); assert!( error .message @@ -186,13 +207,14 @@ mod tests { } /// A roster that cannot be read must fail closed. - #[test] - fn an_unreadable_roster_fails_closed() { + #[tokio::test] + async fn an_unreadable_roster_fails_closed() { let error = authorize( Err("keyring locked".into()), AGENT, Capability::AgentsUpdate, ) + .await .unwrap_err(); assert!( error.message.contains("could not verify agent ownership"), @@ -201,13 +223,14 @@ mod tests { ); } - #[test] - fn pubkey_comparison_ignores_hex_case() { + #[tokio::test] + async fn pubkey_comparison_ignores_hex_case() { assert!(authorize( Ok(vec![AGENT.to_ascii_uppercase()]), AGENT, Capability::AgentsCreate ) + .await .is_ok()); } } diff --git a/desktop/src-tauri/src/broker/handlers.rs b/desktop/src-tauri/src/broker/handlers.rs new file mode 100644 index 0000000000..fcb0e3f031 --- /dev/null +++ b/desktop/src-tauri/src/broker/handlers.rs @@ -0,0 +1,199 @@ +//! The three `agents.*` capability handlers. +//! +//! Each handler translates a validated broker request into a call on +//! [`AgentService`] and translates the result back into a +//! [`HandlerOutcome`]. They contain no policy (that is `agents_policy`), no +//! idempotency logic (that is `store`), and no transport concerns. +//! +//! # Why a trait instead of calling the commands directly +//! +//! The agent mutations live behind Tauri command signatures that need an +//! `AppHandle` and a `State`. Depending on those directly here would make every +//! handler untestable without booting an app, and would leave the capability +//! layer entangled with the desktop runtime. [`AgentService`] is the seam: the +//! handlers are pure and tested against a fake, and the Tauri-backed +//! implementation stays thin enough to read in one screen. + +use buzz_sdk_pkg::broker::{ + AgentsCreateArgs, AgentsCreateOutcome, AgentsDeleteArgs, AgentsDeleteOutcome, AgentsUpdateArgs, + AgentsUpdateOutcome, BrokerError, BrokerErrorCode, BrokerRequest, Capability, CapabilityArgs, + CapabilityOutcome, +}; + +use super::pipeline::{ + BoxFuture, CapabilityHandler, CapabilityRegistry, HandlerOutcome, VerifiedRequest, +}; + +/// Why a domain mutation did not succeed. +/// +/// The distinction is the whole point: [`Self::Failed`] asserts that no side +/// effects persisted, while [`Self::Unknown`] asserts nothing. A service that +/// mutated and then lost track must return `Unknown` so the broker records +/// `indeterminate` instead of implying a clean failure. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ServiceError { + /// The operation did not take effect. + Failed(String), + /// Whether the operation took effect is unknown. + Unknown(String), +} + +impl ServiceError { + fn into_outcome(self) -> HandlerOutcome { + match self { + Self::Failed(message) => { + HandlerOutcome::Failed(BrokerError::new(BrokerErrorCode::CapabilityFailed, message)) + } + Self::Unknown(message) => HandlerOutcome::Indeterminate(BrokerError::new( + BrokerErrorCode::OutcomeUnknown, + message, + )), + } + } +} + +/// The agent mutations the broker can perform on an owner's behalf. +/// +/// Implementations own credential access. Nothing in this trait returns secret +/// material: a create reports the new agent's public key, never its nsec. +/// +/// Async because the real implementation mints a key, writes the agent store, +/// and publishes to the relay. `Send + Sync` so a handler can be dispatched +/// from the async pipeline. +pub trait AgentService: Send + Sync { + /// Mint an agent and attach it to the requested channel. + fn create<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsCreateArgs, + ) -> BoxFuture<'a, Result>; + + /// Patch one of the owner's agents. + fn update<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsUpdateArgs, + ) -> BoxFuture<'a, Result>; + + /// Remove one of the owner's agents. + fn delete<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsDeleteArgs, + ) -> BoxFuture<'a, Result>; +} + +/// Dispatches the three `agents.*` capabilities to an [`AgentService`]. +pub struct AgentsHandlers { + service: S, +} + +impl AgentsHandlers { + /// Wrap `service`. + pub fn new(service: S) -> Self { + Self { service } + } +} + +/// Handler for one capability, borrowing the shared service. +struct Handler<'a, S> { + service: &'a S, + capability: Capability, +} + +impl CapabilityHandler for Handler<'_, S> { + fn execute<'a>( + &'a self, + request: &'a VerifiedRequest, + parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, HandlerOutcome> { + Box::pin(async move { + // The pipeline validated the envelope, but the args it validated are + // the ones this handler must act on -- so it re-normalizes rather + // than trusting a separately parsed copy. + let args = match parsed.capability.validated() { + Ok(args) => args, + Err(error) => { + return HandlerOutcome::Failed(BrokerError::invalid_request(error.to_string())) + } + }; + + // A capability name that disagrees with its args is a routing bug, + // not a caller error, and must never reach a mutation. + if args.capability() != self.capability { + return HandlerOutcome::Failed(BrokerError::new( + BrokerErrorCode::Internal, + "broker routed a request to the wrong capability handler", + )); + } + + let owner = &request.owner_pubkey; + match &args { + CapabilityArgs::AgentsCreate(args) => { + match self.service.create(owner, args).await { + Ok(outcome) => { + HandlerOutcome::Succeeded(CapabilityOutcome::AgentsCreate(outcome)) + } + Err(error) => error.into_outcome(), + } + } + CapabilityArgs::AgentsUpdate(args) => { + match self.service.update(owner, args).await { + Ok(outcome) => { + HandlerOutcome::Succeeded(CapabilityOutcome::AgentsUpdate(outcome)) + } + Err(error) => error.into_outcome(), + } + } + CapabilityArgs::AgentsDelete(args) => { + match self.service.delete(owner, args).await { + Ok(outcome) => { + HandlerOutcome::Succeeded(CapabilityOutcome::AgentsDelete(outcome)) + } + Err(error) => error.into_outcome(), + } + } + } + }) + } +} + +/// Registry over the three `agents.*` handlers. +pub struct AgentsRegistry<'a, S> { + create: Handler<'a, S>, + update: Handler<'a, S>, + delete: Handler<'a, S>, +} + +impl AgentsHandlers { + /// Build the registry this host offers. + pub fn registry(&self) -> AgentsRegistry<'_, S> { + AgentsRegistry { + create: Handler { + service: &self.service, + capability: Capability::AgentsCreate, + }, + update: Handler { + service: &self.service, + capability: Capability::AgentsUpdate, + }, + delete: Handler { + service: &self.service, + capability: Capability::AgentsDelete, + }, + } + } +} + +impl CapabilityRegistry for AgentsRegistry<'_, S> { + fn handler(&self, capability: Capability) -> Option<&dyn CapabilityHandler> { + Some(match capability { + Capability::AgentsCreate => &self.create, + Capability::AgentsUpdate => &self.update, + Capability::AgentsDelete => &self.delete, + }) + } +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/broker/handlers/tests.rs b/desktop/src-tauri/src/broker/handlers/tests.rs new file mode 100644 index 0000000000..9bec4ab386 --- /dev/null +++ b/desktop/src-tauri/src/broker/handlers/tests.rs @@ -0,0 +1,355 @@ +//! Handler tests, driven through the real pipeline against a fake service. + +use std::sync::Mutex; + +use buzz_sdk_pkg::broker::{AgentTarget, BrokerResponse, BrokerResult}; + +use super::super::agents_policy::{AgentRoster, AgentsAuthorizer}; +use super::super::pipeline::{execute, BrokerContext}; +use super::super::store::MemoryExecutionLog; +use super::*; + +const CHANNEL: &str = "b2c38ca8-9ec3-411e-bab5-f9deab34d52e"; +const OWNER: &str = "1111111111111111111111111111111111111111111111111111111111111111"; +const REQUESTER: &str = "2222222222222222222222222222222222222222222222222222222222222222"; +const AGENT: &str = "3333333333333333333333333333333333333333333333333333333333333333"; + +/// Records every call so a test can assert what the domain layer was asked to +/// do -- and that it was asked exactly once. +/// +/// `Mutex` rather than `RefCell` because [`AgentService`] is `Send + Sync`: the +/// pipeline dispatches handlers from async code, so a fake has to be shareable +/// the same way a real service is. +#[derive(Default)] +struct FakeService { + calls: Mutex>, + create: Option>, + update: Option>, + delete: Option>, +} + +impl FakeService { + fn creating() -> Self { + Self { + create: Some(Ok(AgentsCreateOutcome { + agent_pubkey: AGENT.into(), + display_name: "Helper".into(), + channel_id: CHANNEL.into(), + })), + ..Self::default() + } + } + + fn record(&self, call: String) { + self.calls.lock().expect("fake service poisoned").push(call); + } + + fn calls(&self) -> Vec { + self.calls.lock().expect("fake service poisoned").clone() + } +} + +impl AgentService for FakeService { + fn create<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsCreateArgs, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + self.record(format!("create:{owner_pubkey}:{}", args.display_name)); + self.create + .clone() + .unwrap_or(Err(ServiceError::Failed("no create configured".into()))) + }) + } + + fn update<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsUpdateArgs, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + self.record(format!("update:{owner_pubkey}:{:?}", args.target)); + self.update + .clone() + .unwrap_or(Err(ServiceError::Failed("no update configured".into()))) + }) + } + + fn delete<'a>( + &'a self, + owner_pubkey: &'a str, + args: &'a AgentsDeleteArgs, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + self.record(format!("delete:{owner_pubkey}:{:?}", args.target)); + self.delete + .clone() + .unwrap_or(Err(ServiceError::Failed("no delete configured".into()))) + }) + } +} + +struct OwnerRoster; +impl AgentRoster for OwnerRoster { + fn agent_pubkeys<'a>( + &'a self, + _owner_pubkey: &'a str, + ) -> BoxFuture<'a, Result, String>> { + Box::pin(async { Ok(vec![REQUESTER.into()]) }) + } +} + +async fn run(payload: Vec, service: FakeService) -> (BrokerResponse, Vec) { + let handlers = AgentsHandlers::new(service); + let registry = handlers.registry(); + let authorizer = AgentsAuthorizer::new(OwnerRoster); + let log = MemoryExecutionLog::new(); + + let request = VerifiedRequest { + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + relay_scope: "wss://relay.example".into(), + payload, + }; + let response = execute( + &request, + BrokerContext { + authorizer: &authorizer, + registry: ®istry, + log: &log, + now: 1000, + }, + ) + .await + .unwrap(); + let calls = handlers.service.calls(); + (response, calls) +} + +fn payload(args: CapabilityArgs) -> Vec { + serde_json::to_vec(&BrokerRequest::new("req-1", args).unwrap()).unwrap() +} + +fn create_args() -> CapabilityArgs { + CapabilityArgs::AgentsCreate(AgentsCreateArgs { + channel_id: CHANNEL.into(), + display_name: "Helper".into(), + system_prompt: "Help.".into(), + runtime: None, + provider: None, + model: None, + respond_to: None, + }) +} + +#[tokio::test] +async fn create_reaches_the_service_and_reports_the_new_agent() { + let (response, calls) = run(payload(create_args()), FakeService::creating()).await; + + match response.result { + BrokerResult::Succeeded { + outcome: CapabilityOutcome::AgentsCreate(outcome), + } => { + assert_eq!(outcome.agent_pubkey, AGENT); + assert_eq!(outcome.channel_id, CHANNEL); + } + other => panic!("expected create success, got {other:?}"), + } + assert_eq!(calls, vec![format!("create:{OWNER}:Helper")]); +} + +/// The owner the service acts for comes from the verified transport, not the +/// payload -- so a handler can never be aimed at another owner's agents. +#[tokio::test] +async fn the_service_receives_the_transport_derived_owner() { + let (_, calls) = run(payload(create_args()), FakeService::creating()).await; + assert!(calls[0].contains(OWNER), "unexpected call: {}", calls[0]); +} + +#[tokio::test] +async fn update_and_delete_route_to_their_own_service_methods() { + let update = FakeService { + update: Some(Ok(AgentsUpdateOutcome { + agent_pubkey: AGENT.into(), + display_name: "Renamed".into(), + updated_fields: vec!["displayName".into()], + })), + ..FakeService::default() + }; + let (response, calls) = run( + payload(CapabilityArgs::AgentsUpdate(AgentsUpdateArgs { + target: AgentTarget::Pubkey(AGENT.into()), + display_name: Some("Renamed".into()), + system_prompt: None, + runtime: None, + provider: None, + model: None, + respond_to: None, + })), + update, + ) + .await; + assert!(response.result.is_succeeded()); + assert!(calls[0].starts_with("update:"), "unexpected: {}", calls[0]); + + let delete = FakeService { + delete: Some(Ok(AgentsDeleteOutcome { + agent_pubkey: AGENT.into(), + display_name: "Gone".into(), + })), + ..FakeService::default() + }; + let (response, calls) = run( + payload(CapabilityArgs::AgentsDelete(AgentsDeleteArgs { + target: AgentTarget::Pubkey(AGENT.into()), + })), + delete, + ) + .await; + assert!(response.result.is_succeeded()); + assert!(calls[0].starts_with("delete:"), "unexpected: {}", calls[0]); +} + +/// A service that failed cleanly must produce `failed`, which promises no side +/// effects took hold. +#[tokio::test] +async fn a_clean_service_failure_becomes_failed() { + let service = FakeService { + create: Some(Err(ServiceError::Failed("runtime not installed".into()))), + ..FakeService::default() + }; + let (response, _) = run(payload(create_args()), service).await; + + match &response.result { + BrokerResult::Failed { error } => { + assert_eq!(error.code, BrokerErrorCode::CapabilityFailed); + assert!(error.message.contains("runtime not installed")); + } + other => panic!("expected failed, got {other:?}"), + } +} + +/// A service that cannot tell whether it mutated must produce `indeterminate`, +/// never `failed` -- the difference is what stops a caller from safely retrying +/// something that may already have happened. +#[tokio::test] +async fn an_unknown_service_outcome_becomes_indeterminate_not_failed() { + let service = FakeService { + create: Some(Err(ServiceError::Unknown( + "agent minted but profile publish unconfirmed".into(), + ))), + ..FakeService::default() + }; + let (response, _) = run(payload(create_args()), service).await; + + match &response.result { + BrokerResult::Indeterminate { error } => { + assert_eq!(error.code, BrokerErrorCode::OutcomeUnknown); + } + other => panic!("expected indeterminate, got {other:?}"), + } +} + +/// A requester that is not one of the owner's agents must never reach the +/// service at all. +#[tokio::test] +async fn an_unauthorized_requester_never_reaches_the_service() { + struct EmptyRoster; + impl AgentRoster for EmptyRoster { + fn agent_pubkeys<'a>( + &'a self, + _owner: &'a str, + ) -> BoxFuture<'a, Result, String>> { + Box::pin(async { Ok(Vec::new()) }) + } + } + + let handlers = AgentsHandlers::new(FakeService::creating()); + let registry = handlers.registry(); + let authorizer = AgentsAuthorizer::new(EmptyRoster); + let log = MemoryExecutionLog::new(); + + let request = VerifiedRequest { + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + relay_scope: "wss://relay.example".into(), + payload: payload(create_args()), + }; + let response = execute( + &request, + BrokerContext { + authorizer: &authorizer, + registry: ®istry, + log: &log, + now: 1000, + }, + ) + .await + .unwrap(); + + assert_eq!( + response.result.error().unwrap().code, + BrokerErrorCode::Unauthorized + ); + assert!( + handlers.service.calls().is_empty(), + "an unauthorized request must not touch the domain layer" + ); +} + +/// End-to-end idempotency through the real handler stack: the domain mutation +/// must run once even though the request arrives twice. +#[tokio::test] +async fn a_replayed_request_does_not_mutate_twice() { + let handlers = AgentsHandlers::new(FakeService::creating()); + let registry = handlers.registry(); + let authorizer = AgentsAuthorizer::new(OwnerRoster); + let log = MemoryExecutionLog::new(); + + let request = VerifiedRequest { + owner_pubkey: OWNER.into(), + requester_pubkey: REQUESTER.into(), + relay_scope: "wss://relay.example".into(), + payload: payload(create_args()), + }; + let run_once = || async { + execute( + &request, + BrokerContext { + authorizer: &authorizer, + registry: ®istry, + log: &log, + now: 1000, + }, + ) + .await + .unwrap() + }; + + let first = run_once().await; + let second = run_once().await; + + assert!(first.result.is_succeeded()); + assert_eq!(first.result, second.result); + assert!(second.replayed); + assert_eq!( + handlers.service.calls().len(), + 1, + "the domain mutation must run exactly once" + ); +} + +/// A create outcome cannot carry the minted secret, by construction. +#[tokio::test] +async fn a_create_outcome_never_carries_the_minted_nsec() { + let (response, _) = run(payload(create_args()), FakeService::creating()).await; + let json = serde_json::to_string(&response).unwrap(); + assert!(!json.contains("nsec"), "response leaked a secret: {json}"); + for forbidden in ["privateKey", "private_key", "secret"] { + assert!( + !json.contains(forbidden), + "response leaked {forbidden}: {json}" + ); + } +} diff --git a/desktop/src-tauri/src/broker/mod.rs b/desktop/src-tauri/src/broker/mod.rs index 3c5bcbda7b..af99b3659b 100644 --- a/desktop/src-tauri/src/broker/mod.rs +++ b/desktop/src-tauri/src/broker/mod.rs @@ -33,6 +33,7 @@ use sha2::{Digest, Sha256}; pub mod agents_policy; +pub mod handlers; pub mod pipeline; pub mod store; diff --git a/desktop/src-tauri/src/broker/pipeline.rs b/desktop/src-tauri/src/broker/pipeline.rs index 6c88c22a47..52716fc6af 100644 --- a/desktop/src-tauri/src/broker/pipeline.rs +++ b/desktop/src-tauri/src/broker/pipeline.rs @@ -12,6 +12,20 @@ //! already run", and a crash between two coordinated steps is a side effect //! nobody recorded. //! +//! # Why the seams are async +//! +//! The capability this ships first mutates through `create_managed_agent` / +//! `update_managed_agent` / `delete_managed_agent`, which are `async` and +//! publish to the relay. So [`CapabilityHandler`], [`Authorizer`], and +//! [`ExecutionLog`] are all async — modelled on [`crate::archive::sync`]'s +//! `ArchiveSyncIo`, the repo's existing injected-async-IO seam. +//! +//! `rusqlite::Connection` is `!Sync` and must not be held across an `.await`, +//! so the pipeline never touches one: it talks to [`ExecutionLog`], whose +//! production implementation confines each connection to a single blocking +//! task. That is why the durable log is a trait here rather than a borrowed +//! connection. +//! //! # What this module does not decide //! //! It does not verify signatures or decrypt frames — that already happened, and @@ -19,6 +33,9 @@ //! transport-derived identity. It does not implement capabilities either; //! those are [`CapabilityHandler`]s. +use std::future::Future; +use std::pin::Pin; + use buzz_sdk_pkg::broker::{ BrokerError, BrokerErrorCode, BrokerRequest, BrokerResponse, BrokerResult, Capability, CapabilityOutcome, @@ -26,6 +43,9 @@ use buzz_sdk_pkg::broker::{ use super::store::{self, ClaimOutcome, ExecutionKey, ExecutionState}; +/// Boxed future alias, matching [`crate::archive::sync`]'s injected-IO seams. +pub type BoxFuture<'a, T> = Pin + Send + 'a>>; + /// A request whose frame has already been verified and decrypted. /// /// Constructing this type is the assertion that `owner_pubkey`, @@ -43,6 +63,34 @@ pub struct VerifiedRequest { pub payload: Vec, } +/// The durable execution log, as the pipeline needs it. +/// +/// A trait rather than a `&mut Connection` because the pipeline is async and +/// `rusqlite::Connection` is `!Sync`: an implementation must keep each +/// connection inside one blocking task, which only it can arrange. +pub trait ExecutionLog: Send + Sync { + /// Claim `key` for execution, or report why it cannot be claimed. + /// + /// Must insert `executing` before returning [`ClaimOutcome::Claimed`], and + /// must never return `Claimed` twice for one key. + fn claim( + &self, + key: ExecutionKey, + request_digest: String, + capability_version: u16, + now: i64, + ) -> BoxFuture<'_, Result>; + + /// Record the terminal `state` and `result_json` for a claimed key. + fn complete( + &self, + key: ExecutionKey, + state: ExecutionState, + result_json: String, + now: i64, + ) -> BoxFuture<'_, Result<(), String>>; +} + /// Whether a requester may invoke a capability. /// /// Base authentication (a verified signer, a real owner binding) is the @@ -50,14 +98,14 @@ pub struct VerifiedRequest { /// *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 { +pub trait Authorizer: Send + Sync { /// Authorize `request` for `capability`, or explain the refusal. - fn authorize( - &self, - request: &VerifiedRequest, + fn authorize<'a>( + &'a self, + request: &'a VerifiedRequest, capability: Capability, - parsed: &BrokerRequest, - ) -> Result<(), BrokerError>; + parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, Result<(), BrokerError>>; } /// The outcome of running one capability. @@ -75,17 +123,21 @@ pub enum HandlerOutcome { } /// One broker capability implementation. -pub trait CapabilityHandler { +pub trait CapabilityHandler: Send + Sync { /// 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; + fn execute<'a>( + &'a self, + request: &'a VerifiedRequest, + parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, HandlerOutcome>; } /// Resolves a capability name to its handler. -pub trait CapabilityRegistry { +pub trait CapabilityRegistry: Send + Sync { /// The handler for `capability`, or `None` when this host does not offer it. fn handler(&self, capability: Capability) -> Option<&dyn CapabilityHandler>; } @@ -97,7 +149,7 @@ pub struct BrokerContext<'a> { /// Capability implementations. pub registry: &'a dyn CapabilityRegistry, /// Durable execution log. - pub conn: &'a mut rusqlite::Connection, + pub log: &'a dyn ExecutionLog, /// Current unix time, injected so tests are not clock-dependent. pub now: i64, } @@ -108,7 +160,7 @@ pub struct BrokerContext<'a> { /// (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( +pub async fn execute( request: &VerifiedRequest, ctx: BrokerContext<'_>, ) -> Result { @@ -152,7 +204,7 @@ pub fn execute( // 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) { + if let Err(error) = ctx.authorizer.authorize(request, capability, &parsed).await { return Ok(BrokerResponse::new( &request_id, BrokerResult::failed(error), @@ -177,7 +229,11 @@ pub fn execute( request_id: request_id.clone(), }; - match store::claim(ctx.conn, &key, &digest, parsed.capability_version, ctx.now)? { + let claim = ctx + .log + .claim(key.clone(), digest, parsed.capability_version, ctx.now) + .await?; + match claim { ClaimOutcome::Claimed => {} ClaimOutcome::Replay(record) => { return Ok(replay_response(&request_id, record.result_json.as_deref())) @@ -208,7 +264,7 @@ pub fn execute( // 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) { + let result = match handler.execute(request, &parsed).await { HandlerOutcome::Succeeded(outcome) => BrokerResult::succeeded(outcome), HandlerOutcome::Failed(error) => BrokerResult::failed(error), HandlerOutcome::Indeterminate(error) => BrokerResult::indeterminate(error), @@ -221,7 +277,7 @@ pub fn execute( }; 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)?; + ctx.log.complete(key, state, result_json, ctx.now).await?; Ok(BrokerResponse::new(&request_id, result)) } diff --git a/desktop/src-tauri/src/broker/pipeline/tests.rs b/desktop/src-tauri/src/broker/pipeline/tests.rs index 70fb18b85d..5fbf08f58c 100644 --- a/desktop/src-tauri/src/broker/pipeline/tests.rs +++ b/desktop/src-tauri/src/broker/pipeline/tests.rs @@ -1,12 +1,14 @@ //! Pipeline tests: authority, idempotency, and crash behavior. -use std::cell::RefCell; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Mutex; use buzz_sdk_pkg::broker::{ AgentTarget, AgentsCreateArgs, AgentsCreateOutcome, AgentsDeleteArgs, AgentsDeleteOutcome, CapabilityArgs, }; +use super::super::store::MemoryExecutionLog; use super::*; const CHANNEL: &str = "b2c38ca8-9ec3-411e-bab5-f9deab34d52e"; @@ -14,8 +16,8 @@ const OWNER: &str = "11111111111111111111111111111111111111111111111111111111111 const REQUESTER: &str = "2222222222222222222222222222222222222222222222222222222222222222"; const AGENT: &str = "3333333333333333333333333333333333333333333333333333333333333333"; -fn db() -> rusqlite::Connection { - super::super::store::open_in_memory() +fn db() -> MemoryExecutionLog { + MemoryExecutionLog::new() } fn create_request(request_id: &str) -> Vec { @@ -46,47 +48,57 @@ fn verified(payload: Vec) -> VerifiedRequest { struct AllowAll; impl Authorizer for AllowAll { - fn authorize( - &self, - _request: &VerifiedRequest, + fn authorize<'a>( + &'a self, + _request: &'a VerifiedRequest, _capability: Capability, - _parsed: &BrokerRequest, - ) -> Result<(), BrokerError> { - Ok(()) + _parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, Result<(), BrokerError>> { + Box::pin(async { Ok(()) }) } } struct DenyAll; impl Authorizer for DenyAll { - fn authorize( - &self, - _request: &VerifiedRequest, + fn authorize<'a>( + &'a self, + _request: &'a VerifiedRequest, _capability: Capability, - _parsed: &BrokerRequest, - ) -> Result<(), BrokerError> { - Err(BrokerError::unauthorized("not a member of that channel")) + _parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, Result<(), BrokerError>> { + Box::pin(async { 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, + calls: AtomicUsize, outcome: fn() -> HandlerOutcome, } impl CountingHandler { fn new(outcome: fn() -> HandlerOutcome) -> Self { Self { - calls: RefCell::new(0), + calls: AtomicUsize::new(0), outcome, } } + + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } } impl CapabilityHandler for CountingHandler { - fn execute(&self, _request: &VerifiedRequest, _parsed: &BrokerRequest) -> HandlerOutcome { - *self.calls.borrow_mut() += 1; - (self.outcome)() + fn execute<'a>( + &'a self, + _request: &'a VerifiedRequest, + _parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, HandlerOutcome> { + Box::pin(async move { + self.calls.fetch_add(1, Ordering::SeqCst); + (self.outcome)() + }) } } @@ -116,27 +128,28 @@ fn success() -> HandlerOutcome { })) } -fn run<'a>( +async fn run( request: &VerifiedRequest, - authorizer: &'a dyn Authorizer, - registry: &'a dyn CapabilityRegistry, - conn: &'a mut rusqlite::Connection, + authorizer: &dyn Authorizer, + registry: &dyn CapabilityRegistry, + log: &dyn ExecutionLog, ) -> BrokerResponse { execute( request, BrokerContext { authorizer, registry, - conn, + log, now: 1000, }, ) + .await .unwrap() } -#[test] -fn a_valid_authorized_request_executes_once_and_returns_its_outcome() { - let mut conn = db(); +#[tokio::test] +async fn a_valid_authorized_request_executes_once_and_returns_its_outcome() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -144,18 +157,18 @@ fn a_valid_authorized_request_executes_once_and_returns_its_outcome() { }; let request = verified(create_request("req-1")); - let response = run(&request, &AllowAll, ®istry, &mut conn); + let response = run(&request, &AllowAll, ®istry, &log).await; assert_eq!(response.request_id, "req-1"); assert!(response.result.is_succeeded()); assert!(!response.replayed); - assert_eq!(*handler.calls.borrow(), 1); + assert_eq!(handler.calls(), 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(); +#[tokio::test] +async fn replaying_an_identical_request_returns_the_record_without_re_executing() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -163,19 +176,19 @@ fn replaying_an_identical_request_returns_the_record_without_re_executing() { }; let request = verified(create_request("req-1")); - let first = run(&request, &AllowAll, ®istry, &mut conn); - let second = run(&request, &AllowAll, ®istry, &mut conn); + let first = run(&request, &AllowAll, ®istry, &log).await; + let second = run(&request, &AllowAll, ®istry, &log).await; - assert_eq!(*handler.calls.borrow(), 1, "handler must not run twice"); + assert_eq!(handler.calls(), 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(); +#[tokio::test] +async fn a_different_payload_under_the_same_request_id_is_refused() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -186,8 +199,9 @@ fn a_different_payload_under_the_same_request_id_is_refused() { &verified(create_request("req-1")), &AllowAll, ®istry, - &mut conn, - ); + &log, + ) + .await; let mut different = BrokerRequest::new( "req-1", @@ -207,21 +221,22 @@ fn a_different_payload_under_the_same_request_id_is_refused() { &verified(serde_json::to_vec(&different).unwrap()), &AllowAll, ®istry, - &mut conn, - ); + &log, + ) + .await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::RequestIdConflict ); - assert_eq!(*handler.calls.borrow(), 1, "the conflict must not execute"); + assert_eq!(handler.calls(), 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(); +#[tokio::test] +async fn an_unauthorized_request_does_not_execute_or_leave_a_claim() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -229,25 +244,25 @@ fn an_unauthorized_request_does_not_execute_or_leave_a_claim() { }; let request = verified(create_request("req-1")); - let denied = run(&request, &DenyAll, ®istry, &mut conn); + let denied = run(&request, &DenyAll, ®istry, &log).await; assert_eq!( denied.result.error().unwrap().code, BrokerErrorCode::Unauthorized ); - assert_eq!(*handler.calls.borrow(), 0); + assert_eq!(handler.calls(), 0); // The same request, now authorized, still runs normally. - let allowed = run(&request, &AllowAll, ®istry, &mut conn); + let allowed = run(&request, &AllowAll, ®istry, &log).await; assert!(allowed.result.is_succeeded()); assert!(!allowed.replayed); - assert_eq!(*handler.calls.borrow(), 1); + assert_eq!(handler.calls(), 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(); +#[tokio::test] +async fn an_interrupted_execution_reports_indeterminate_and_never_re_executes() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -264,29 +279,30 @@ fn an_interrupted_execution_reports_indeterminate_and_never_re_executes() { request_id: "req-1".into(), }; let digest = super::super::request_digest(&request.payload).unwrap(); - store::claim(&mut conn, &key, &digest, 1, 500).unwrap(); + log.with_conn(|conn| store::claim(conn, &key, &digest, 1, 500)) + .unwrap(); - let response = run(&request, &AllowAll, ®istry, &mut conn); + let response = run(&request, &AllowAll, ®istry, &log).await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::OutcomeUnknown ); assert_eq!( - *handler.calls.borrow(), + handler.calls(), 0, "an interrupted request must not be re-executed" ); } -#[test] -fn a_failed_handler_is_recorded_and_replayed_as_failed() { +#[tokio::test] +async 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 log = db(); let handler = CountingHandler::new(failure); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -294,28 +310,28 @@ fn a_failed_handler_is_recorded_and_replayed_as_failed() { }; let request = verified(create_request("req-1")); - let first = run(&request, &AllowAll, ®istry, &mut conn); + let first = run(&request, &AllowAll, ®istry, &log).await; 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); + let second = run(&request, &AllowAll, ®istry, &log).await; assert!(second.replayed); assert_eq!(first.result, second.result); - assert_eq!(*handler.calls.borrow(), 1); + assert_eq!(handler.calls(), 1); } -#[test] -fn an_indeterminate_handler_outcome_is_recorded_as_indeterminate() { +#[tokio::test] +async 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 log = db(); let handler = CountingHandler::new(unknown); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -323,53 +339,56 @@ fn an_indeterminate_handler_outcome_is_recorded_as_indeterminate() { }; let request = verified(create_request("req-1")); - let response = run(&request, &AllowAll, ®istry, &mut conn); + let response = run(&request, &AllowAll, ®istry, &log).await; assert!(matches!( response.result, BrokerResult::Indeterminate { .. } )); - assert_eq!(*handler.calls.borrow(), 1); + assert_eq!(handler.calls(), 1); } -#[test] -fn a_malformed_payload_is_refused_without_a_claim() { - let mut conn = db(); +#[tokio::test] +async fn a_malformed_payload_is_refused_without_a_claim() { + let log = db(); let response = run( &verified(b"not json".to_vec()), &AllowAll, &NoHandlers, - &mut conn, - ); + &log, + ) + .await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::InvalidRequest ); } -#[test] -fn an_oversized_payload_is_refused() { - let mut conn = db(); +#[tokio::test] +async fn an_oversized_payload_is_refused() { + let log = db(); let response = run( &verified(vec![b'x'; super::super::MAX_REQUEST_BYTES + 1]), &AllowAll, &NoHandlers, - &mut conn, - ); + &log, + ) + .await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::InvalidRequest ); } -#[test] -fn a_capability_this_host_does_not_offer_is_refused() { - let mut conn = db(); +#[tokio::test] +async fn a_capability_this_host_does_not_offer_is_refused() { + let log = db(); let response = run( &verified(create_request("req-1")), &AllowAll, &NoHandlers, - &mut conn, - ); + &log, + ) + .await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::UnknownCapability @@ -378,9 +397,9 @@ fn a_capability_this_host_does_not_offer_is_refused() { /// 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(); +#[tokio::test] +async fn the_same_request_id_from_two_requesters_are_separate_executions() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -394,19 +413,19 @@ fn the_same_request_id_from_two_requesters_are_separate_executions() { ..verified(payload) }; - assert!(!run(&first, &AllowAll, ®istry, &mut conn).replayed); + assert!(!run(&first, &AllowAll, ®istry, &log).await.replayed); assert!( - !run(&second, &AllowAll, ®istry, &mut conn).replayed, + !run(&second, &AllowAll, ®istry, &log).await.replayed, "a different requester must not replay another's record" ); - assert_eq!(*handler.calls.borrow(), 2); + assert_eq!(handler.calls(), 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(); +#[tokio::test] +async fn identity_in_the_payload_cannot_override_transport_identity() { + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, @@ -422,53 +441,57 @@ fn identity_in_the_payload_cannot_override_transport_identity() { &verified(serde_json::to_vec(&json).unwrap()), &AllowAll, ®istry, - &mut conn, - ); + &log, + ) + .await; assert_eq!( response.result.error().unwrap().code, BrokerErrorCode::InvalidRequest ); - assert_eq!(*handler.calls.borrow(), 0); + assert_eq!(handler.calls(), 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>); +#[tokio::test] +async fn the_authorizer_receives_transport_derived_identity() { + struct Recording(Mutex>); impl Authorizer for Recording { - fn authorize( - &self, - request: &VerifiedRequest, + fn authorize<'a>( + &'a self, + request: &'a 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(()) + _parsed: &'a BrokerRequest, + ) -> BoxFuture<'a, Result<(), BrokerError>> { + Box::pin(async move { + self.0.lock().unwrap().push(( + request.owner_pubkey.clone(), + request.requester_pubkey.clone(), + capability.as_str().to_string(), + )); + Ok(()) + }) } } - let mut conn = db(); + let log = db(); let handler = CountingHandler::new(success); let registry = OneHandler { capability: Capability::AgentsCreate, handler: &handler, }; - let authorizer = Recording(RefCell::new(Vec::new())); + let authorizer = Recording(Mutex::new(Vec::new())); run( &verified(create_request("req-1")), &authorizer, ®istry, - &mut conn, - ); + &log, + ) + .await; assert_eq!( - authorizer.0.borrow().as_slice(), + authorizer.0.lock().unwrap().as_slice(), &[( OWNER.to_string(), REQUESTER.to_string(), @@ -477,15 +500,15 @@ fn the_authorizer_receives_transport_derived_identity() { ); } -#[test] -fn delete_requests_route_to_the_delete_capability() { +#[tokio::test] +async 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 log = db(); let handler = CountingHandler::new(deleted); let registry = OneHandler { capability: Capability::AgentsDelete, @@ -503,9 +526,10 @@ fn delete_requests_route_to_the_delete_capability() { &verified(serde_json::to_vec(&request).unwrap()), &AllowAll, ®istry, - &mut conn, - ); + &log, + ) + .await; assert!(response.result.is_succeeded()); - assert_eq!(*handler.calls.borrow(), 1); + assert_eq!(handler.calls(), 1); } diff --git a/desktop/src-tauri/src/broker/store.rs b/desktop/src-tauri/src/broker/store.rs index aa417ebb8b..3c32fc3478 100644 --- a/desktop/src-tauri/src/broker/store.rs +++ b/desktop/src-tauri/src/broker/store.rs @@ -23,6 +23,8 @@ use std::time::{Duration, Instant}; use rusqlite::{params, Connection, OptionalExtension}; use sha2::{Digest, Sha256}; +use super::pipeline::{BoxFuture, ExecutionLog}; + /// Terminal or in-flight state of a broker request. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ExecutionState { @@ -364,6 +366,118 @@ pub fn open_in_memory() -> Connection { conn } +/// In-memory [`ExecutionLog`] running the real [`claim`] / [`complete`] logic. +/// +/// Tests get the production idempotency behavior without touching the +/// filesystem. `Connection` is `Send` but not `Sync`, so the mutex is what makes +/// it shareable behind the `Send + Sync` trait — and holding the lock for the +/// whole call serializes claims the same way the file-backed `IMMEDIATE` +/// transaction does. +#[cfg(test)] +pub struct MemoryExecutionLog(std::sync::Mutex); + +#[cfg(test)] +impl MemoryExecutionLog { + /// A fresh, empty log. + pub fn new() -> Self { + Self(std::sync::Mutex::new(open_in_memory())) + } + + /// Run `body` against the underlying connection. + /// + /// Lets a test plant a row directly — simulating a process that claimed a + /// key and died — without exposing the connection itself. + pub fn with_conn(&self, body: impl FnOnce(&mut Connection) -> T) -> T { + body(&mut self.0.lock().expect("broker test log poisoned")) + } +} + +#[cfg(test)] +impl ExecutionLog for MemoryExecutionLog { + fn claim( + &self, + key: ExecutionKey, + request_digest: String, + capability_version: u16, + now: i64, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + self.with_conn(|conn| claim(conn, &key, &request_digest, capability_version, now)) + }) + } + + fn complete( + &self, + key: ExecutionKey, + state: ExecutionState, + result_json: String, + now: i64, + ) -> BoxFuture<'_, Result<(), String>> { + Box::pin( + async move { self.with_conn(|conn| complete(conn, &key, state, &result_json, now)) }, + ) + } +} + +/// File-backed [`ExecutionLog`] that keeps every `!Sync` connection inside one +/// blocking task. +/// +/// The pipeline is async, so it can never hold a `rusqlite::Connection` across +/// an `.await`. This adapter is where that constraint is discharged: each call +/// opens its own connection on the blocking pool, finishes its transaction, and +/// drops it before the future resolves. Cross-call exclusion is SQLite's, not +/// this type's — [`claim`] does its read and insert inside one `IMMEDIATE` +/// transaction, which is what makes two concurrent claims safe even though they +/// use different connections. +pub struct SqliteExecutionLog { + path: PathBuf, +} + +impl SqliteExecutionLog { + /// Log stored at `path`. The file and its parent are created on first use. + pub fn new(path: PathBuf) -> Self { + Self { path } + } +} + +impl ExecutionLog for SqliteExecutionLog { + fn claim( + &self, + key: ExecutionKey, + request_digest: String, + capability_version: u16, + now: i64, + ) -> BoxFuture<'_, Result> { + let path = self.path.clone(); + Box::pin(async move { + tokio::task::spawn_blocking(move || { + let mut conn = open_broker_db(&path)?; + claim(&mut conn, &key, &request_digest, capability_version, now) + }) + .await + .map_err(|error| format!("broker claim task failed: {error}"))? + }) + } + + fn complete( + &self, + key: ExecutionKey, + state: ExecutionState, + result_json: String, + now: i64, + ) -> BoxFuture<'_, Result<(), String>> { + let path = self.path.clone(); + Box::pin(async move { + tokio::task::spawn_blocking(move || { + let conn = open_broker_db(&path)?; + complete(&conn, &key, state, &result_json, now) + }) + .await + .map_err(|error| format!("broker complete task failed: {error}"))? + }) + } +} + /// Read one execution row, if it exists. pub fn get(conn: &Connection, key: &ExecutionKey) -> Result, String> { conn.query_row(