From 34d29841cecd1bb24ed189e12ee7131edf8807c1 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 3 Nov 2026 18:59:10 -0500 Subject: [PATCH] feat(agents): persist bounded shared-task admissions Signed-off-by: Tee8z --- crates/buzz-sandbox-reaper/src/main.rs | 2 + crates/buzz-sandbox-reaper/src/manager.rs | 61 ++++++ .../buzz-sandbox-reaper/src/manager_tests.rs | 3 + .../src/manager_tests/shared.rs | 58 ++++++ crates/buzz-sandbox-reaper/src/mentions.rs | 1 + .../src/mentions/registry.rs | 2 +- .../buzz-sandbox-reaper/src/shared_budget.rs | 123 ++++++++++++ .../src/shared_budget/tests.rs | 127 +++++++++++++ .../buzz-sandbox-reaper/src/shared_tasks.rs | 132 +++++++++++++ crates/buzz-sdk/src/sharing.rs | 1 + crates/buzz-sdk/src/sharing/budget.rs | 178 ++++++++++++++++++ crates/buzz-sdk/src/sharing/budget/tests.rs | 119 ++++++++++++ crates/buzz-sdk/src/sharing/policy.rs | 3 + crates/buzz-sdk/src/sharing/task.rs | 7 +- 14 files changed, 814 insertions(+), 3 deletions(-) create mode 100644 crates/buzz-sandbox-reaper/src/manager_tests/shared.rs create mode 100644 crates/buzz-sandbox-reaper/src/shared_budget.rs create mode 100644 crates/buzz-sandbox-reaper/src/shared_budget/tests.rs create mode 100644 crates/buzz-sandbox-reaper/src/shared_tasks.rs create mode 100644 crates/buzz-sdk/src/sharing/budget.rs create mode 100644 crates/buzz-sdk/src/sharing/budget/tests.rs diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index e465564e7a7..2b995a37606 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/crates/buzz-sandbox-reaper/src/main.rs @@ -21,6 +21,8 @@ mod profile; mod reconcile; mod serve; mod server; +mod shared_budget; +mod shared_tasks; use std::collections::BTreeMap; use std::future::{Future, IntoFuture}; diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index f3a4bec686e..c9cd70f4614 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/crates/buzz-sandbox-reaper/src/manager.rs @@ -259,6 +259,13 @@ pub struct Job { /// What a job runs. pub enum Work { + /// Reserve one shared task. This operation cannot start or stop a provider. + Admit { + owner: PublicKey, + agent: PublicKey, + namespace: String, + request: crate::shared_tasks::Request, + }, /// A provider protocol request, against the owner's namespace. Provider { request: Request, @@ -282,6 +289,16 @@ impl Work { /// Run to the `done` response. pub async fn run(self, manager: &Manager) -> serde_json::Value { match self { + Work::Admit { + owner, + agent, + namespace, + request, + } => { + manager + .admit_shared(owner, agent, &namespace, &request) + .await + } Work::Provider { request, target, @@ -553,6 +570,21 @@ impl Manager { Work::Profile { developer: developer.clone(), } + } else if op == Op::Admit { + let request = match serde_json::from_value(opened.request.payload.clone()) { + Ok(request) => request, + Err(_) => { + return answer( + serde_json::json!({"ok":false,"code":"shared_task_denied","error":"admit requires relay_url and the original signed trigger"}), + ) + } + }; + Work::Admit { + owner, + agent: opened.signer, + namespace: developer.namespace.clone(), + request, + } } else if op == Op::Start { let start = match Start::admit( opened.request.payload.clone(), @@ -746,6 +778,35 @@ impl Manager { }) } + pub(crate) async fn admit_shared( + &self, + owner: PublicKey, + agent: PublicKey, + namespace: &str, + request: &crate::shared_tasks::Request, + ) -> serde_json::Value { + use buzz_sdk::sharing::policy::Error as PolicyError; + let result=tokio::time::timeout(std::time::Duration::from_secs(40),async { + let task=match crate::shared_tasks::verify(self.client.clone(),namespace,owner,agent,request).await { + Ok(task)=>task, + Err(PolicyError::Denied(reason))=>return serde_json::json!({"ok":false,"code":"shared_task_denied","error":reason}), + Err(PolicyError::Unavailable)=>return serde_json::json!({"ok":false,"code":"shared_task_retry","error":"current sharing authorization is unavailable; retry this task"}), + }; + let Some(origin)=self.config.developer(&task.origin_owner) else { + return serde_json::json!({"ok":false,"code":"shared_task_denied","error":"the initial agent must use this session manager"}); + }; + match crate::shared_budget::admit(self.client.clone(),&origin.namespace,&request.relay_url,&task,Timestamp::now().as_secs()).await { + Ok(())=>serde_json::json!({"ok":true,"request":task.origin.request,"origin_agent":task.origin.agent, + "human":task.human,"channel":task.scope.channel,"thread_root":task.scope.root}), + Err(error)=>{ + let code=if matches!(error,crate::shared_budget::Error::Storage) {"shared_task_retry"} else {"shared_task_denied"}; + serde_json::json!({"ok":false,"code":code,"error":error.to_string()}) + } + } + }).await; + result.unwrap_or_else(|_|serde_json::json!({"ok":false,"code":"shared_task_retry","error":"task authorization timed out; retry this task"})) + } + /// Step 3: parse the provider request and bind its agent key to the /// signer. The response is the refusal to send otherwise. fn payload(&self, opened: &OpenedRequest) -> Result { diff --git a/crates/buzz-sandbox-reaper/src/manager_tests.rs b/crates/buzz-sandbox-reaper/src/manager_tests.rs index f0d02a616ae..1bb39799ccf 100644 --- a/crates/buzz-sandbox-reaper/src/manager_tests.rs +++ b/crates/buzz-sandbox-reaper/src/manager_tests.rs @@ -5,6 +5,9 @@ use nostr::nips::nip44; use nostr::{EventBuilder, Kind, Tag}; use serde_json::{json, Value}; +#[path = "manager_tests/shared.rs"] +mod shared; + const NAMESPACE: &str = "buzz-agent-staging-dev"; fn keys(byte: &str) -> Keys { diff --git a/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs b/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs new file mode 100644 index 00000000000..e931d3c68aa --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs @@ -0,0 +1,58 @@ +use super::*; + +fn task_request() -> Sent { + let trigger = EventBuilder::new(Kind::Custom(9), "teammate task") + .tags([ + Tag::parse(["h", "0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10"]).unwrap(), + Tag::public_key(agent().public_key()), + ]) + .sign_with_keys(&stranger()) + .unwrap(); + Sent { + op: Op::Admit, + payload: json!({"relay_url":"wss://relay.example.invalid","trigger":trigger}), + ..Sent::deploy() + } +} + +#[tokio::test] +async fn shared_admission_is_agent_attested_and_confined_to_a_registered_identity() { + let manager = manager(); + let sent = task_request(); + let Admission::Start { job, .. } = manager.admit(&sent.event(), now()) else { + panic!("task was not admitted for verification") + }; + let Work::Admit { + owner: admitted_owner, + agent: admitted_agent, + namespace, + request, + } = &job.work + else { + panic!("shared task acquired a lifecycle capability") + }; + assert_eq!(*admitted_owner, owner().public_key()); + assert_eq!(*admitted_agent, agent().public_key()); + assert_eq!(namespace, NAMESPACE); + assert_eq!(request.trigger.pubkey, stranger().public_key()); + let response = job.work.run(&manager).await; + assert_eq!(response["ok"], false); + assert_eq!( + response["code"], "shared_task_denied", + "unregistered identities must fail before connecting to the requested relay" + ); +} + +#[tokio::test] +async fn shared_grants_cannot_replace_owner_identity_or_inject_lifecycle_payloads() { + let manager = manager(); + let mut sent = task_request(); + sent.auth = None; + assert_eq!(admit(&manager, &sent), "unauthorized_owner"); + let mut sent = task_request(); + sent.payload["provider_config"] = json!({"namespace":"someone-else"}); + assert_eq!(admit(&manager, &sent), "shared_task_denied"); + let mut sent = task_request(); + sent.op = Op::Start; + assert_eq!(admit(&manager, &sent), "signer_mismatch"); +} diff --git a/crates/buzz-sandbox-reaper/src/mentions.rs b/crates/buzz-sandbox-reaper/src/mentions.rs index bb343091f4d..67cdc36335a 100644 --- a/crates/buzz-sandbox-reaper/src/mentions.rs +++ b/crates/buzz-sandbox-reaper/src/mentions.rs @@ -4,6 +4,7 @@ mod journal; mod registry; +pub(crate) use registry::Identity; mod worker; use std::collections::BTreeMap; diff --git a/crates/buzz-sandbox-reaper/src/mentions/registry.rs b/crates/buzz-sandbox-reaper/src/mentions/registry.rs index 90b5fc0db7c..4ca4383aaec 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/registry.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/registry.rs @@ -11,7 +11,7 @@ use crate::manager::Manager; use crate::mobile_launch::{relay_endpoint, Registration, DATA_KEY, LABEL}; /// No Debug: this holds the agent's credentials, never the owner's key. -pub(super) struct Identity { +pub(crate) struct Identity { pub name: String, pub namespace: String, pub revision: String, diff --git a/crates/buzz-sandbox-reaper/src/shared_budget.rs b/crates/buzz-sandbox-reaper/src/shared_budget.rs new file mode 100644 index 00000000000..8f08c146ee1 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/shared_budget.rs @@ -0,0 +1,123 @@ +//! One bounded, CAS-protected admission journal per origin owner and community. +//! A missing/ambiguous write acknowledgement propagates; replay converges on the +//! same receipt rather than replenishing the task's budget. + +use buzz_sdk::sharing::{budget::Budget, policy::Task}; +use k8s_openapi::api::core::v1::Secret; +use kube::{api::PostParams, Api}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::collections::BTreeMap; + +const LABEL: &str = "buzz.voltage.cloud/shared-admissions"; +const DATA: &str = "admissions.json"; +const MAX_TASKS: usize = 512; +const MAX_BYTES: usize = 900 * 1024; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum Error { + #[error(transparent)] + Refused(#[from] buzz_sdk::sharing::budget::Error), + #[error("task admission storage unavailable; retry this task")] + Storage, + #[error("task admission capacity reached; wait for older tasks to expire")] + Capacity, + #[error("task admission journal is invalid; operator repair is required")] + Journal, +} + +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct Journal { + version: u32, + relay: String, + tasks: BTreeMap, +} + +/// Persist an authorized task before either startup or harness delivery. The +/// namespace comes from the initial agent's verified, configured owner. +pub(crate) async fn admit( + client: kube::Client, + namespace: &str, + relay: &str, + task: &Task, + now: u64, +) -> Result<(), Error> { + let relay = crate::mobile_launch::relay_endpoint(relay).map_err(|_| Error::Journal)?; + let digest = Sha256::digest(relay.as_bytes()); + let hash: String = digest[..20].iter().map(|b| format!("{b:02x}")).collect(); + let name = format!("buzz-shared-admissions-{hash}"); + let api: Api = Api::namespaced(client, namespace); + for _ in 0..3 { + let prior = api.get_opt(&name).await.map_err(|_| Error::Storage)?; + let mut journal = if let Some(secret) = &prior { + if secret + .metadata + .labels + .as_ref() + .and_then(|l| l.get(LABEL)) + .map(String::as_str) + != Some("v1") + { + return Err(Error::Journal); + } + let bytes = &secret + .data + .as_ref() + .and_then(|d| d.get(DATA)) + .ok_or(Error::Journal)? + .0; + if bytes.len() > MAX_BYTES { + return Err(Error::Capacity); + } + let journal: Journal = serde_json::from_slice(bytes).map_err(|_| Error::Journal)?; + if journal.version != 1 || journal.relay != relay || journal.tasks.len() > MAX_TASKS { + return Err(Error::Journal); + } + journal + } else { + Journal { + version: 1, + relay: relay.clone(), + tasks: BTreeMap::new(), + } + }; + journal.tasks.retain(|_, budget| !budget.expired(now)); + let key = Budget::key(task); + if !journal.tasks.contains_key(&key) { + if journal.tasks.len() >= MAX_TASKS { + return Err(Error::Capacity); + } + journal.tasks.insert(key.clone(), Budget::new(task, now)?); + } + let budget = journal.tasks.get_mut(&key).ok_or(Error::Journal)?; + budget.admit(task, now)?; + let body = serde_json::to_string(&journal).map_err(|_| Error::Journal)?; + if body.len() > MAX_BYTES { + return Err(Error::Capacity); + } + let mut secret = Secret { + type_: Some("Opaque".into()), + string_data: Some(BTreeMap::from([(DATA.into(), body)])), + ..Default::default() + }; + secret.metadata.name = Some(name.clone()); + secret.metadata.namespace = Some(namespace.into()); + secret.metadata.labels = Some(BTreeMap::from([(LABEL.into(), "v1".into())])); + secret.metadata.resource_version = prior.and_then(|s| s.metadata.resource_version); + let saved = if secret.metadata.resource_version.is_some() { + api.replace(&name, &PostParams::default(), &secret).await + } else { + api.create(&PostParams::default(), &secret).await + }; + match saved { + Ok(_) => return Ok(()), + Err(kube::Error::Api(e)) if e.code == 409 => continue, + Err(_) => return Err(Error::Storage), + } + } + Err(Error::Storage) +} + +#[cfg(test)] +mod tests; diff --git a/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs b/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs new file mode 100644 index 00000000000..297a9ebb7fb --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/shared_budget/tests.rs @@ -0,0 +1,127 @@ +use super::*; +use buzz_backend_kubernetes::sandbox_fake; +use buzz_sdk::sharing::task::{Origin, Scope}; +use nostr::{EventId, Keys}; + +fn task() -> Task { + let human = Keys::generate().public_key(); + let agent = Keys::generate().public_key(); + let id = EventId::from_byte_array([1; 32]); + Task { + scope: Scope { + channel: uuid::Uuid::new_v4(), + root: id, + }, + requester: human, + agent, + owner: human, + origin: Origin { + request: id, + agent, + parent: id, + }, + human, + origin_owner: human, + trigger: id, + handoff: false, + requested_at: 100, + } +} + +#[tokio::test] +async fn journal_round_trip_replay_and_failed_write_preserve_budget() { + let state = sandbox_fake::Shared::default(); + let client = sandbox_fake::client(&state); + let base = task(); + state.lock().unwrap().fail_secret_create_responses = 1; + assert!(matches!( + admit(client.clone(), "agents", "ws://relay.test", &base, 101).await, + Err(Error::Storage) + )); + admit(client.clone(), "agents", "ws://relay.test", &base, 101) + .await + .unwrap(); + let first = Task { + handoff: true, + requester: base.agent, + agent: Keys::generate().public_key(), + trigger: EventId::from_byte_array([2; 32]), + ..base + }; + // The real kube request path must return an error when persistence failed. + state.lock().unwrap().fail_secret_updates = 1; + assert!(matches!( + admit(client.clone(), "agents", "ws://relay.test", &first, 102).await, + Err(Error::Storage) + )); + admit(client.clone(), "agents", "ws://relay.test", &first, 102) + .await + .unwrap(); + admit(client.clone(), "agents", "ws://relay.test", &first, 103) + .await + .unwrap(); + let secrets = state.lock().unwrap().secrets.clone(); + assert_eq!(secrets.len(), 1); + let stored: Secret = serde_json::from_value(secrets.values().next().unwrap().clone()).unwrap(); + let journal: Journal = serde_json::from_slice(&stored.data.unwrap()[DATA].0).unwrap(); + let encoded = serde_json::to_value(&journal).unwrap(); + assert_eq!( + encoded["tasks"][Budget::key(&base)]["receipts"] + .as_array() + .unwrap() + .len(), + 2 + ); + let expired = base.requested_at + buzz_sdk::sharing::budget::LIFETIME_SECONDS; + assert!(matches!( + admit(client, "agents", "ws://relay.test", &base, expired).await, + Err(Error::Refused(_)) + )); +} + +#[tokio::test] +async fn concurrent_distinct_handoffs_cannot_lose_a_reserved_admission() { + let state = sandbox_fake::Shared::default(); + let client = sandbox_fake::client(&state); + let base = task(); + admit(client.clone(), "agents", "ws://relay.test", &base, 101) + .await + .unwrap(); + let first = Task { + handoff: true, + requester: base.agent, + agent: Keys::generate().public_key(), + trigger: EventId::from_byte_array([2; 32]), + ..base + }; + let second = Task { + agent: Keys::generate().public_key(), + trigger: EventId::from_byte_array([3; 32]), + ..first + }; + let (a, b) = tokio::join!( + admit(client.clone(), "agents", "ws://relay.test", &first, 102), + admit(client.clone(), "agents", "ws://relay.test", &second, 102) + ); + a.unwrap(); + b.unwrap(); + let stored: Secret = serde_json::from_value( + state + .lock() + .unwrap() + .secrets + .values() + .next() + .unwrap() + .clone(), + ) + .unwrap(); + let journal: Journal = serde_json::from_slice(&stored.data.unwrap()[DATA].0).unwrap(); + assert_eq!( + serde_json::to_value(journal).unwrap()["tasks"][Budget::key(&base)]["receipts"] + .as_array() + .unwrap() + .len(), + 3 + ); +} diff --git a/crates/buzz-sandbox-reaper/src/shared_tasks.rs b/crates/buzz-sandbox-reaper/src/shared_tasks.rs new file mode 100644 index 00000000000..535e5363728 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/shared_tasks.rs @@ -0,0 +1,132 @@ +//! Agent-authenticated relay reads for one shared admission. No manager channel +//! membership or new relay read permission is required. + +use crate::{mentions::Identity, mobile_launch::Registration}; +use buzz_sdk::sharing::policy::{self, Task}; +use buzz_ws_client::{NostrWsConnection, RelayMessage}; +use k8s_openapi::api::core::v1::Secret; +use kube::Api; +use nostr::{Event, PublicKey, Timestamp}; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::time::Duration; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Request { + pub relay_url: String, + pub trigger: Event, +} + +pub(crate) async fn verify( + client: kube::Client, + namespace: &str, + owner: PublicKey, + agent: PublicKey, + request: &Request, +) -> Result { + let relay = crate::mobile_launch::relay_endpoint(&request.relay_url) + .map_err(|_| policy::Error::Denied("invalid relay"))?; + let api: Api = Api::namespaced(client, namespace); + let secret = api + .get_opt(&Registration::name(owner, agent, &relay)) + .await + .map_err(|_| policy::Error::Unavailable)? + .ok_or(policy::Error::Denied( + "start this agent once from desktop before sharing", + ))?; + let identity = + Identity::read(secret, namespace, owner, &relay).map_err(|_| policy::Error::Unavailable)?; + if identity.keys.public_key() != agent { + return Err(policy::Error::Unavailable); + } + let relay_key = relay_key(&identity.relay).await?; + policy::authorize( + |filter| query(&identity, filter), + relay_key, + owner, + agent, + &request.trigger, + Timestamp::now().as_secs(), + ) + .await +} + +async fn relay_key(relay: &str) -> Result { + let mut url = url::Url::parse(relay).map_err(|_| policy::Error::Unavailable)?; + let scheme = match url.scheme() { + "wss" => "https", + "ws" => "http", + _ => return Err(policy::Error::Unavailable), + }; + url.set_scheme(scheme) + .map_err(|_| policy::Error::Unavailable)?; + let http = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .map_err(|_| policy::Error::Unavailable)?; + let mut response = http + .get(url) + .header("accept", "application/nostr+json") + .send() + .await + .map_err(|_| policy::Error::Unavailable)? + .error_for_status() + .map_err(|_| policy::Error::Unavailable)?; + let mut bytes = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|_| policy::Error::Unavailable)? + { + if bytes.len() + chunk.len() > 64 * 1024 { + return Err(policy::Error::Unavailable); + } + bytes.extend_from_slice(&chunk); + } + let info: Value = serde_json::from_slice(&bytes).map_err(|_| policy::Error::Unavailable)?; + PublicKey::from_hex(info["self"].as_str().ok_or(policy::Error::Unavailable)?) + .map_err(|_| policy::Error::Unavailable) +} + +async fn query(identity: &Identity, filter: Value) -> Result, ()> { + tokio::time::timeout(Duration::from_secs(10), async { + let mut connection = NostrWsConnection::connect_authenticated( + &identity.relay, + &identity.keys, + Some(&identity.auth), + ) + .await + .map_err(|_| ())?; + connection + .send_raw(&json!(["REQ", "shared-admission", filter])) + .await + .map_err(|_| ())?; + let mut events = Vec::new(); + for _ in 0..8 { + match connection + .next_event(Duration::from_secs(10)) + .await + .map_err(|_| ())? + { + RelayMessage::Event { + subscription_id, + event, + } if subscription_id == "shared-admission" => { + if !events.is_empty() { + return Err(()); + } + events.push(*event); + } + RelayMessage::Eose { subscription_id } if subscription_id == "shared-admission" => { + return Ok(events) + } + RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(()), + _ => {} + } + } + Err(()) + }) + .await + .map_err(|_| ())? +} diff --git a/crates/buzz-sdk/src/sharing.rs b/crates/buzz-sdk/src/sharing.rs index 5248ee47ac2..f3ea2148ac0 100644 --- a/crates/buzz-sdk/src/sharing.rs +++ b/crates/buzz-sdk/src/sharing.rs @@ -10,6 +10,7 @@ use uuid::Uuid; use crate::{kind::KIND_AGENT_CHANNEL_GRANT, nip_oa, SdkError}; +pub mod budget; pub mod policy; pub mod task; diff --git a/crates/buzz-sdk/src/sharing/budget.rs b/crates/buzz-sdk/src/sharing/budget.rs new file mode 100644 index 00000000000..ed4394387dd --- /dev/null +++ b/crates/buzz-sdk/src/sharing/budget.rs @@ -0,0 +1,178 @@ +//! Bounded task admissions. Persist a successful mutation atomically before any +//! launch or turn; retries of the same (task, agent) consume no further budget. + +use super::{policy::Task, task::Scope}; +use nostr::{EventId, PublicKey}; +use serde::{Deserialize, Serialize}; + +/// Maximum lifetime of one human request's handoff budget. +pub const LIFETIME_SECONDS: u64 = 7 * 24 * 60 * 60; +const MAX_HANDOFFS: usize = 4; +const MAX_AGENT_TASKS: usize = 2; + +/// A fixed refusal, safe to report without exposing any credential or task text. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +pub enum Error { + /// A fresh human mention is required after the original request expires. + #[error("task expired; ask a human to mention the agent again")] + Expired, + /// The agent-to-agent budget has been exhausted. + #[error("four agent handoffs already admitted; ask a human before continuing")] + Handoffs, + /// One agent has already handled two tasks under this request. + #[error("agent already admitted twice for this request; ask a human before continuing")] + RepeatedAgent, + /// A handoff must refer to an admitted incoming task for its signer. + #[error("handoff parent was not admitted for the sending agent")] + Parent, + /// Stored provenance does not match the verified task. + #[error("task admission journal is invalid")] + Journal, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct Receipt { + event: EventId, + agent: PublicKey, + requester: PublicKey, + handoff: bool, +} + +/// The durable admission record for one initial human-to-agent task. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Budget { + version: u32, + scope: Scope, + request: EventId, + agent: PublicKey, + human: PublicKey, + owner: PublicKey, + requested_at: u64, + receipts: Vec, +} + +impl Budget { + /// Start from the verified human task. Seeding its initial receipt lets an + /// owner-only agent make a handoff without having used shared admission itself. + pub fn new(task: &Task, now: u64) -> Result { + let budget = Self { + version: 1, + scope: task.scope, + request: task.origin.request, + agent: task.origin.agent, + human: task.human, + owner: task.origin_owner, + requested_at: task.requested_at, + receipts: vec![Receipt { + event: task.origin.request, + agent: task.origin.agent, + requester: task.human, + handoff: false, + }], + }; + budget.validate(task, now)?; + Ok(budget) + } + + /// The stable storage key. Each explicit initial agent has its own task. + pub fn key(task: &Task) -> String { + format!("{}:{}", task.origin.request, task.origin.agent) + } + + /// Whether this record can be pruned. The original signed request remains + /// ineligible after pruning, preventing a restart from replenishing its budget. + pub fn expired(&self, now: u64) -> bool { + now >= self.requested_at.saturating_add(LIFETIME_SECONDS) + } + + fn validate(&self, task: &Task, now: u64) -> Result<(), Error> { + if self.expired(now) || self.requested_at > now.saturating_add(60) { + return Err(Error::Expired); + } + if self.version != 1 + || self.scope != task.scope + || self.request != task.origin.request + || self.agent != task.origin.agent + || self.human != task.human + || self.owner != task.origin_owner + || self.requested_at != task.requested_at + || self.receipts.is_empty() + || self.receipts.len() > MAX_HANDOFFS + 1 + { + return Err(Error::Journal); + } + let first = &self.receipts[0]; + if first.event != self.request + || first.agent != self.agent + || first.requester != self.human + || first.handoff + { + return Err(Error::Journal); + } + for (index, receipt) in self.receipts.iter().enumerate() { + if (index > 0 && !receipt.handoff) + || self.receipts[..index] + .iter() + .any(|r| r.event == receipt.event && r.agent == receipt.agent) + || self + .receipts + .iter() + .filter(|r| r.agent == receipt.agent) + .count() + > MAX_AGENT_TASKS + { + return Err(Error::Journal); + } + } + Ok(()) + } + + /// Reserve a task in memory; caller must persist the resulting record with + /// compare-and-swap before running it. Returns false for an identical retry. + pub fn admit(&mut self, task: &Task, now: u64) -> Result { + self.validate(task, now)?; + if let Some(prior) = self + .receipts + .iter() + .find(|r| r.event == task.trigger && r.agent == task.agent) + { + return if prior.requester == task.requester && prior.handoff == task.handoff { + Ok(false) + } else { + Err(Error::Journal) + }; + } + if !task.handoff + || !self + .receipts + .iter() + .any(|r| r.event == task.origin.parent && r.agent == task.requester) + { + return Err(Error::Parent); + } + if self.receipts.len() > MAX_HANDOFFS { + return Err(Error::Handoffs); + } + if self + .receipts + .iter() + .filter(|r| r.agent == task.agent) + .count() + >= MAX_AGENT_TASKS + { + return Err(Error::RepeatedAgent); + } + self.receipts.push(Receipt { + event: task.trigger, + agent: task.agent, + requester: task.requester, + handoff: true, + }); + Ok(true) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/buzz-sdk/src/sharing/budget/tests.rs b/crates/buzz-sdk/src/sharing/budget/tests.rs new file mode 100644 index 00000000000..ac4629e0eb8 --- /dev/null +++ b/crates/buzz-sdk/src/sharing/budget/tests.rs @@ -0,0 +1,119 @@ +use super::*; +use crate::sharing::task::Origin; +use nostr::Keys; + +pub fn task() -> Task { + let human = Keys::generate().public_key(); + let agent = Keys::generate().public_key(); + let id = EventId::from_byte_array([1; 32]); + Task { + scope: Scope { + channel: uuid::Uuid::new_v4(), + root: id, + }, + requester: human, + agent, + owner: human, + origin: Origin { + request: id, + agent, + parent: id, + }, + human, + origin_owner: human, + trigger: id, + handoff: false, + requested_at: 100, + } +} +fn next(base: &Task, source: PublicKey, target: PublicKey, parent: EventId, byte: u8) -> Task { + Task { + agent: target, + requester: source, + trigger: EventId::from_byte_array([byte; 32]), + handoff: true, + origin: Origin { + parent, + ..base.origin + }, + ..*base + } +} + +#[test] +fn receipts_survive_restart_and_enforce_parent_and_per_agent_limits() { + let base = task(); + let peer = Keys::generate().public_key(); + let mut budget = Budget::new(&base, 101).unwrap(); + assert!(!budget.admit(&base, 101).unwrap()); + let first = next(&base, base.agent, peer, base.trigger, 2); + assert!(budget.admit(&first, 102).unwrap()); + let second = next(&base, peer, base.agent, first.trigger, 3); + assert!(budget.admit(&second, 103).unwrap()); + let mut recovered: Budget = + serde_json::from_str(&serde_json::to_string(&budget).unwrap()).unwrap(); + assert!(!recovered.admit(&first, 104).unwrap()); + let forged = next( + &base, + peer, + Keys::generate().public_key(), + EventId::from_byte_array([77; 32]), + 4, + ); + assert_eq!(recovered.admit(&forged, 104), Err(Error::Parent)); + let reset = next(&base, peer, base.agent, first.trigger, 5); + assert_eq!(recovered.admit(&reset, 104), Err(Error::RepeatedAgent)); + let spoofed = next( + &base, + Keys::generate().public_key(), + peer, + second.trigger, + 6, + ); + assert_eq!(recovered.admit(&spoofed, 104), Err(Error::Parent)); +} + +#[test] +fn fifth_handoff_is_refused_even_after_restart_but_exact_retry_is_free() { + let base = task(); + let mut budget = Budget::new(&base, 101).unwrap(); + let mut prior = base; + for byte in 2..=5 { + let next = next( + &base, + prior.agent, + Keys::generate().public_key(), + prior.trigger, + byte, + ); + assert!(budget.admit(&next, 101).unwrap()); + prior = next; + } + let mut budget: Budget = + serde_json::from_str(&serde_json::to_string(&budget).unwrap()).unwrap(); + assert!(!budget.admit(&prior, 102).unwrap()); + let fifth = next( + &base, + prior.agent, + Keys::generate().public_key(), + prior.trigger, + 6, + ); + assert_eq!(budget.admit(&fifth, 102), Err(Error::Handoffs)); +} + +#[test] +fn expiry_cannot_replenish_an_old_request_and_different_humans_cannot_share_a_budget() { + let base = task(); + let mut budget = Budget::new(&base, 101).unwrap(); + let expiry = base.requested_at + LIFETIME_SECONDS; + assert!(budget.expired(expiry)); + assert_eq!(budget.admit(&base, expiry), Err(Error::Expired)); + assert!(matches!(Budget::new(&base, expiry), Err(Error::Expired))); + let mut forged = base; + forged.human = Keys::generate().public_key(); + assert_eq!(budget.admit(&forged, 102), Err(Error::Journal)); + forged = base; + forged.scope.channel = uuid::Uuid::new_v4(); + assert_eq!(budget.admit(&forged, 102), Err(Error::Journal)); +} diff --git a/crates/buzz-sdk/src/sharing/policy.rs b/crates/buzz-sdk/src/sharing/policy.rs index 356ff41df3b..b1a455f2864 100644 --- a/crates/buzz-sdk/src/sharing/policy.rs +++ b/crates/buzz-sdk/src/sharing/policy.rs @@ -43,6 +43,8 @@ pub struct Task { pub trigger: EventId, /// Whether the requester is a verified agent. pub handoff: bool, + /// Signed creation time of the original human task. Expiry cannot reset. + pub requested_at: u64, } /// Load current grants, identities, metadata and membership, then verify task @@ -183,6 +185,7 @@ where origin_owner, trigger: event.id, handoff: automated, + requested_at: root.created_at.as_secs(), }) } diff --git a/crates/buzz-sdk/src/sharing/task.rs b/crates/buzz-sdk/src/sharing/task.rs index befd70bb727..9ceb53b26bb 100644 --- a/crates/buzz-sdk/src/sharing/task.rs +++ b/crates/buzz-sdk/src/sharing/task.rs @@ -2,10 +2,12 @@ use crate::SdkError; use nostr::{Event, EventBuilder, EventId, Kind, PublicKey, Tag}; +use serde::{Deserialize, Serialize}; use uuid::Uuid; /// A mention's signed channel and canonical thread binding. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] pub struct Scope { /// The sole canonical channel tag. pub channel: Uuid, @@ -14,7 +16,8 @@ pub struct Scope { } /// Provenance retained across cross-owner agent handoffs. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] pub struct Origin { /// The original human's signed task message. pub request: EventId,