feat(agents): persist bounded shared-task admissions

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};

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<Request, serde_json::Value> {

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 {

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");
+}

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;

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,

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<String, Budget>,
+}
+
+/// 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<Secret> = 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;