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

broker: add agents.* capability handlers behind an async service seam

Browse Source 
The three capabilities dispatch to an AgentService trait rather than to
the Tauri commands directly. Those commands need an AppHandle and a
State, so depending on them here would make every handler untestable
without booting an app and would entangle the capability layer with the
desktop runtime. Behind the seam, handlers are pure and the tests below
drive the real pipeline against a fake.

The seams are async because the mutations they will call are: minting a
key, writing the agent store, and publishing to the relay all await.
That forces ExecutionLog to be a trait instead of a borrowed
rusqlite::Connection, which is !Sync and must not be held across an
.await. The production adapter confines each connection to one blocking
task; the test double serializes behind a Mutex. Cross-call exclusion is
SQLite's either way, since claim() does its read and insert inside a
single IMMEDIATE transaction.

ServiceError distinguishes Failed from Unknown. Failed promises no side
effects persisted; Unknown promises nothing and becomes indeterminate. A
service that mutated and then lost track must say Unknown, because
reporting a clean failure would invite a retry of something that may
already have happened.

Nothing in AgentService returns secret material: a create reports the new
agent's pubkey, never its nsec. A test serializes a create response and
asserts the absence of nsec, privateKey, private_key, and secret, so the
guarantee is checked rather than asserted in a comment.

Co-authored-by: Bradley Axen <baxen@squareup.com>
Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
ss-dev-00
2026-08-21 17:04:23 -07:00
parent b173f46e75
commit 86617f5f5f
8 changed files with 989 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
crates/buzz-core/src/observer.rs
View File

@@ -80,11 +80,22 @@ pub fn encrypt_observer_payload<T: Serialize>(
Ok(encrypted)
}
/// NIP-44 decrypt and deserialize an observer payload from `event`.
pub fn decrypt_observer_payload<T: DeserializeOwned>(
/// 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<T, ObserverPayloadError> {
) -> Result<String, ObserverPayloadError> {
if !content_looks_like_nip44(&event.content) {
return Err(ObserverPayloadError::InvalidCiphertextLength(
event.content.len(),
@@ -104,7 +115,15 @@ pub fn decrypt_observer_payload<T: DeserializeOwned>(
got,
});
}
Ok(plaintext)
}
/// NIP-44 decrypt and deserialize an observer payload from `event`.
pub fn decrypt_observer_payload<T: DeserializeOwned>(
recipient_keys: &Keys,
event: &Event,
) -> Result<T, ObserverPayloadError> {
let mut plaintext = decrypt_observer_plaintext(recipient_keys, event)?;
let result = serde_json::from_str(&plaintext);
plaintext.zeroize();
Ok(result?)

143
desktop/src-tauri/src/broker/agents_policy.rs
View File

@@ -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<Vec<String>, String>;
fn agent_pubkeys<'a>(
&'a self,
owner_pubkey: &'a str,
) -> BoxFuture<'a, Result<Vec<String>, String>>;
}
/// `agents.*` authorization backed by an [`AgentRoster`].
@@ -49,47 +52,51 @@ impl<R: AgentRoster> AgentsAuthorizer<R> {
}
impl<R: AgentRoster> Authorizer for AgentsAuthorizer<R> {
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<Vec<String>, String>);
impl AgentRoster for Roster {
fn agent_pubkeys(&self, _owner_pubkey: &str) -> Result<Vec<String>, String> {
self.0.clone()
fn agent_pubkeys<'a>(
&'a self,
_owner_pubkey: &'a str,
) -> BoxFuture<'a, Result<Vec<String>, 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<Vec<String>, 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());
}
}

199
desktop/src-tauri/src/broker/handlers.rs Normal file
View File

@@ -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<AgentsCreateOutcome, ServiceError>>;
/// Patch one of the owner's agents.
fn update<'a>(
&'a self,
owner_pubkey: &'a str,
args: &'a AgentsUpdateArgs,
) -> BoxFuture<'a, Result<AgentsUpdateOutcome, ServiceError>>;
/// Remove one of the owner's agents.
fn delete<'a>(
&'a self,
owner_pubkey: &'a str,
args: &'a AgentsDeleteArgs,
) -> BoxFuture<'a, Result<AgentsDeleteOutcome, ServiceError>>;
}
/// Dispatches the three `agents.*` capabilities to an [`AgentService`].
pub struct AgentsHandlers<S> {
service: S,
}
impl<S: AgentService> AgentsHandlers<S> {
/// 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<S: AgentService> 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<S: AgentService> AgentsHandlers<S> {
/// 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<S: AgentService> 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;

355
desktop/src-tauri/src/broker/handlers/tests.rs Normal file
View File

@@ -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<Vec<String>>,
create: Option<Result<AgentsCreateOutcome, ServiceError>>,
update: Option<Result<AgentsUpdateOutcome, ServiceError>>,
delete: Option<Result<AgentsDeleteOutcome, ServiceError>>,
}
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<String> {
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<AgentsCreateOutcome, ServiceError>> {
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<AgentsUpdateOutcome, ServiceError>> {
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<AgentsDeleteOutcome, ServiceError>> {
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<Vec<String>, String>> {
Box::pin(async { Ok(vec![REQUESTER.into()]) })
}
}
async fn run(payload: Vec<u8>, service: FakeService) -> (BrokerResponse, Vec<String>) {
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: &registry,
log: &log,
now: 1000,
},
)
.await
.unwrap();
let calls = handlers.service.calls();
(response, calls)
}
fn payload(args: CapabilityArgs) -> Vec<u8> {
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<Vec<String>, 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: &registry,
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: &registry,
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}"
);
}
}

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

@@ -33,6 +33,7 @@
use sha2::{Digest, Sha256};
pub mod agents_policy;
pub mod handlers;
pub mod pipeline;
pub mod store;

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

@@ -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<Box<dyn Future<Output = T> + 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<u8>,
}
/// 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<ClaimOutcome, String>>;
/// 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<BrokerResponse, String> {
@@ -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))
}

264
desktop/src-tauri/src/broker/pipeline/tests.rs
View File

@@ -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<u8> {
@@ -46,47 +48,57 @@ fn verified(payload: Vec<u8>) -> 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<usize>,
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, &registry, &mut conn);
let response = run(&request, &AllowAll, &registry, &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, &registry, &mut conn);
let second = run(&request, &AllowAll, &registry, &mut conn);
let first = run(&request, &AllowAll, &registry, &log).await;
let second = run(&request, &AllowAll, &registry, &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,
&registry,
&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,
&registry,
&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, &registry, &mut conn);
let denied = run(&request, &DenyAll, &registry, &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, &registry, &mut conn);
let allowed = run(&request, &AllowAll, &registry, &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, &registry, &mut conn);
let response = run(&request, &AllowAll, &registry, &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, &registry, &mut conn);
let first = run(&request, &AllowAll, &registry, &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, &registry, &mut conn);
let second = run(&request, &AllowAll, &registry, &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, &registry, &mut conn);
let response = run(&request, &AllowAll, &registry, &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, &registry, &mut conn).replayed);
assert!(!run(&first, &AllowAll, &registry, &log).await.replayed);
assert!(
!run(&second, &AllowAll, &registry, &mut conn).replayed,
!run(&second, &AllowAll, &registry, &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,
&registry,
&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<Vec<(String, String, String)>>);
#[tokio::test]
async fn the_authorizer_receives_transport_derived_identity() {
struct Recording(Mutex<Vec<(String, String, String)>>);
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,
&registry,
&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,
&registry,
&mut conn,
);
&log,
)
.await;
assert!(response.result.is_succeeded());
assert_eq!(*handler.calls.borrow(), 1);
assert_eq!(handler.calls(), 1);
}

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

@@ -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<Connection>);
#[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<T>(&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<ClaimOutcome, String>> {
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<ClaimOutcome, String>> {
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<Option<ExecutionRecord>, String> {
conn.query_row(