feat(agents): wake shared agents from channel-scoped mentions

crates/buzz-backend-kubernetes/src/sandbox_fake.rs

@@ -159,12 +159,21 @@ impl State {
         let path = path.split('?').next().unwrap_or_default();
         let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect();
         match segments.as_slice() {
-            ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes", rest @ ..] => {
-                self.sandbox(method, rest.first().copied(), body)
+            ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", namespace, "sandboxes", rest @ ..] => {
+                self.sandbox(method, namespace, rest.first().copied(), body)
             }
-            ["api", "v1", "namespaces", _, "pods"] => {
+            ["api", "v1", "namespaces", namespace, "pods"] => {
                 let names: Vec<String> = self.pods.keys().cloned().collect();
-                let items: Vec<Value> = names.iter().filter_map(|n| self.pod(n)).collect();
+                let items: Vec<Value> = names
+                    .iter()
+                    .filter(|name| {
+                        self.sandboxes
+                            .get(*name)
+                            .and_then(|s| s["metadata"]["namespace"].as_str())
+                            .is_none_or(|stored| stored == *namespace)
+                    })
+                    .filter_map(|n| self.pod(n))
+                    .collect();
                 ok(&json!({"apiVersion":"v1","kind":"PodList","metadata":{},"items":items}))
             }
             ["api", "v1", "namespaces", _, "pods", name] => {
@@ -277,7 +286,13 @@ impl State {
         }
     }
 
-    fn sandbox(&mut self, method: &Method, name: Option<&str>, body: &[u8]) -> Response<Body> {
+    fn sandbox(
+        &mut self,
+        method: &Method,
+        namespace: &str,
+        name: Option<&str>,
+        body: &[u8],
+    ) -> Response<Body> {
         match (method.clone(), name) {
             (Method::GET, Some(name)) => match self.sandboxes.get(name) {
                 Some(sandbox) => ok(sandbox),
@@ -289,6 +304,7 @@ impl State {
                     return status(500, "InternalError");
                 }
                 let mut sandbox: Value = serde_json::from_slice(body).unwrap();
+                sandbox["metadata"]["namespace"] = namespace.into();
                 let name = sandbox["metadata"]["name"].as_str().unwrap().to_string();
                 if self.sandboxes.contains_key(&name) {
                     return status(409, "AlreadyExists");

crates/buzz-sandbox-reaper/Cargo.toml

@@ -56,6 +56,7 @@ uuid = { workspace = true }
 rustls = { version = "0.23", default-features = false, features = ["ring", "std"] }
 
 [dev-dependencies]
+tempfile = "3"
 buzz-backend-kubernetes = { path = "../buzz-backend-kubernetes", features = ["test-utils"] }
 # The relay provider's end-to-end tests run against this crate's fake relay
 # and the manager in process (serve::relay_provider_tests).

crates/buzz-sandbox-reaper/src/manager.rs

@@ -456,6 +456,12 @@ impl Manager {
         self.client.clone()
     }
 
+    /// Sharing expands task invocation only when the matching harness rollout
+    /// is enabled. Owner lifecycle operations retain their separate checks.
+    pub(crate) fn sharing_enabled(&self) -> bool {
+        self.config.shared_tasks
+    }
+
     /// Reuse mobile start validation without manufacturing an owner control event.
     pub(crate) async fn start_mention(
         &self,
@@ -467,11 +473,26 @@ impl Manager {
         let Some(developer) = self.config.developer(&owner) else {
             return refusal(ErrorCode::UnauthorizedOwner, "owner is not configured");
         };
-        let start = match Start::admit(
-            serde_json::json!({"agent_pubkey": agent, "relay_url": relay, "trigger": event}),
-            owner,
-            Timestamp::now().as_secs(),
-        ) {
+        let admitted = if event.pubkey == owner {
+            Start::admit(
+                serde_json::json!({"agent_pubkey": agent, "relay_url": relay, "trigger": event}),
+                owner,
+                Timestamp::now().as_secs(),
+            )
+        } else {
+            let request = crate::shared_tasks::Request {
+                relay_url: relay.into(),
+                trigger: event.clone(),
+            };
+            match self
+                .authorize_shared(owner, agent, &developer.namespace, &request)
+                .await
+            {
+                Ok(task) => Start::shared(task, relay, event),
+                Err(error) => Err(error),
+            }
+        };
+        let start = match admitted {
             Ok(start) => start,
             Err(error) => return error,
         };
@@ -819,29 +840,49 @@ impl Manager {
         namespace: &str,
         request: &crate::shared_tasks::Request,
     ) -> serde_json::Value {
+        match self
+            .authorize_shared(owner, agent, namespace, request)
+            .await
+        {
+            Ok(task) => {
+                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) => error,
+        }
+    }
+
+    async fn authorize_shared(
+        &self,
+        owner: PublicKey,
+        agent: PublicKey,
+        namespace: &str,
+        request: &crate::shared_tasks::Request,
+    ) -> Result<buzz_sdk::sharing::policy::Task, serde_json::Value> {
         use buzz_sdk::sharing::policy::Error as PolicyError;
         if !self.config.shared_tasks {
-            return serde_json::json!({"ok":false,"code":"shared_task_denied","error":"shared tasks are not enabled on this manager"});
+            return Err(
+                serde_json::json!({"ok":false,"code":"shared_task_denied","error":"shared tasks are not enabled on this manager"}),
+            );
         }
         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"}),
+                Err(PolicyError::Denied(reason))=>return Err(serde_json::json!({"ok":false,"code":"shared_task_denied","error":reason})),
+                Err(PolicyError::Unavailable)=>return Err(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"});
+                return Err(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}),
+                Ok(())=>Ok(task),
                 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()})
+                    Err(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"}))
+        result.unwrap_or_else(|_|Err(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

crates/buzz-sandbox-reaper/src/mentions.rs

@@ -2,6 +2,7 @@
 //! Clients need no control protocol. Agent-authenticated subscriptions retain
 //! the relay's channel access checks; credentials never leave the manager.
 
+mod channels;
 mod journal;
 mod registry;
 pub(crate) use registry::Identity;

crates/buzz-sandbox-reaper/src/mentions/channels.rs

@@ -0,0 +1,75 @@
+//! Bound the wake subscription to the addressed agent's current channels.
+//! On Buzz, an unscoped live subscription routes global events only.
+
+use super::{registry::Identity, Error, IO_TIMEOUT};
+use buzz_ws_client::{NostrWsConnection, RelayMessage};
+use serde_json::json;
+use std::collections::BTreeSet;
+use uuid::Uuid;
+
+const MAX_CHANNELS: usize = 64;
+
+pub(super) async fn read(
+    connection: &mut NostrWsConnection,
+    identity: &Identity,
+) -> Result<Vec<Uuid>, Error> {
+    tokio::time::timeout(IO_TIMEOUT, async {
+        // Kind 39002 is relay-authored channel membership metadata. Its
+        // coordinate is d, not h. No manager membership is needed to read it.
+        connection
+            .send_raw(&json!(["REQ","mention-channels",{
+                "kinds":[39002],"#p":[identity.keys.public_key()],"limit":MAX_CHANNELS+1
+            }]))
+            .await
+            .map_err(|_| Error::Relay)?;
+        let mut channels = BTreeSet::new();
+        for _ in 0..MAX_CHANNELS + 2 {
+            match connection
+                .next_event(IO_TIMEOUT)
+                .await
+                .map_err(|_| Error::Relay)?
+            {
+                RelayMessage::Event {
+                    subscription_id,
+                    event,
+                } if subscription_id == "mention-channels" => {
+                    if event.kind.as_u16() != 39002
+                        || event.verify().is_err()
+                        || !event.tags.iter().any(|tag| {
+                            let t = tag.as_slice();
+                            t.first().map(String::as_str) == Some("p")
+                                && t.get(1) == Some(&identity.keys.public_key().to_hex())
+                        })
+                    {
+                        return Err(Error::Relay);
+                    }
+                    let d = buzz_sdk::sharing::task::single_tag(&event, "d", 2)
+                        .map_err(|_| Error::Relay)?
+                        .ok_or(Error::Relay)?;
+                    let channel = Uuid::parse_str(&d[1]).map_err(|_| Error::Relay)?;
+                    if channel.to_string() != d[1] {
+                        return Err(Error::Relay);
+                    }
+                    channels.insert(channel);
+                    if channels.len() > MAX_CHANNELS {
+                        return Err(Error::Capacity);
+                    }
+                }
+                RelayMessage::Eose { subscription_id } if subscription_id == "mention-channels" => {
+                    connection
+                        .send_raw(&json!(["CLOSE", "mention-channels"]))
+                        .await
+                        .map_err(|_| Error::Relay)?;
+                    return Ok(channels.into_iter().collect());
+                }
+                RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => {
+                    return Err(Error::Relay)
+                }
+                _ => {}
+            }
+        }
+        Err(Error::Capacity)
+    })
+    .await
+    .map_err(|_| Error::Deadline)?
+}

crates/buzz-sandbox-reaper/src/mentions/tests.rs

@@ -4,6 +4,7 @@
 mod live_relay;
 mod relay;
 mod revoked;
+mod shared_live;
 
 use std::sync::Arc;
 use std::time::Duration;
@@ -153,6 +154,43 @@ fn completed(cluster: &sandbox_fake::Shared, event: &Event) -> bool {
     inbox(cluster)["completed"].get(event.id.to_hex()).is_some()
 }
 
+#[tokio::test]
+async fn requested_history_close_ack_keeps_the_live_listener_running() {
+    let relay = relay::Relay::start().await;
+    let cluster = sandbox_fake::Shared::default();
+    let identity = register(&cluster, &relay.url).await;
+    let mut journal = Journal::acquire(
+        sandbox_fake::client(&cluster),
+        &identity,
+        "close-ack",
+        Timestamp::now().as_secs(),
+    )
+    .await
+    .unwrap();
+    let manager = manager(&cluster);
+    let mut listener = tokio::spawn(async move {
+        worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await
+    });
+    relay.wait_subscription("closed:mentions-history").await;
+    assert!(
+        tokio::time::timeout(Duration::from_millis(100), &mut listener)
+            .await
+            .is_err(),
+        "an acknowledgement of our CLOSE must not terminate live delivery"
+    );
+    let mut phone = NostrWsConnection::connect_authenticated(&relay.url, &keys("33"), None)
+        .await
+        .unwrap();
+    let event = message("@agent after history closed", None);
+    assert!(phone.send_event(event.clone()).await.unwrap().accepted);
+    wait_for(|| completed(&cluster, &event)).await;
+    assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1);
+    assert_eq!(relay.count("mentions-live"), 1);
+    listener.abort();
+    let _ = listener.await;
+    relay.stop().await;
+}
+
 #[tokio::test]
 async fn durable_pending_launch_survives_restart_and_storage_failure_prevents_deploy() {
     let relay = relay::Relay::start().await;

crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs

@@ -39,7 +39,14 @@ impl Relay {
             ])
             .sign_with_keys(&keys("33"))
             .unwrap();
-        let events = Arc::new(Mutex::new(vec![metadata, listing]));
+        let members = EventBuilder::new(Kind::Custom(39002), "")
+            .tags([
+                Tag::parse(["d", CHANNEL]).unwrap(),
+                Tag::public_key(keys("22").public_key()),
+            ])
+            .sign_with_keys(&keys("11"))
+            .unwrap();
+        let events = Arc::new(Mutex::new(vec![metadata, listing, members]));
         let cancel = CancellationToken::new();
         let lose_next_ack = Arc::new(AtomicBool::new(false));
         let (subscriptions, receiver) = watch::channel(BTreeMap::new());
@@ -177,7 +184,11 @@ async fn connection(
                         ws.send(frame(json!(["EOSE", id]))).await?;
                         subscriptions.send_modify(|counts| { *counts.entry(id).or_default() += 1; });
                     }
-                    "CLOSE" => { filters.remove(input[1].as_str().unwrap()); }
+                    "CLOSE" => {
+                        filters.remove(input[1].as_str().unwrap());
+                        ws.send(frame(json!(["CLOSED",input[1],""]))).await?;
+                        subscriptions.send_modify(|counts| { *counts.entry(format!("closed:{}",input[1].as_str().unwrap())).or_default() += 1; });
+                    }
                     "EVENT" => {
                         let event: Event = serde_json::from_value(input[1].clone())?;
                         let accepted = Some(event.pubkey) == author && event.verify().is_ok()
@@ -197,6 +208,7 @@ async fn connection(
                 let event = event?;
                 if event.kind.as_u16() != 24210 && !event.tags.iter().any(|t| t.as_slice() == ["h", CHANNEL]) { continue; }
                 for (id, filter) in &filters {
+                    if event.kind.as_u16()!=24210 && filter.get("#h").is_none() {continue;}
                     if matches(filter, &event) { ws.send(frame(json!(["EVENT",id,event]))).await?; }
                 }
             }

crates/buzz-sandbox-reaper/src/mentions/tests/shared_agent.py

@@ -0,0 +1,48 @@
+"""Bounded ACP test adapter. Real Buzz CLI and harness perform all routing."""
+import ctypes
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+if sys.platform == "linux":
+    ctypes.CDLL(None).prctl(1, 9)  # Do not survive a failed test's harness.
+
+
+def send(value):
+    print(json.dumps(value), flush=True)
+
+
+for line in sys.stdin:
+    message = json.loads(line)
+    if "id" not in message:
+        continue
+    method = message.get("method")
+    result = {}
+    if method == "initialize":
+        result = {"protocolVersion": 1, "agentCapabilities": {},
+                  "agentInfo": {"name": "shared-fixture", "version": "1"}}
+    elif method == "session/new":
+        result = {"sessionId": "shared-fixture"}
+    elif method == "session/prompt":
+        marker = Path("model-turns")
+        count = int(marker.read_text()) if marker.exists() else 0
+        marker.write_text(str(count + 1))
+        if os.environ.get("SHARED_FIXTURE_TARGET") and count == 0:
+            subprocess.run([os.environ["SHARED_FIXTURE_CLI"], "agents", "handoff",
+                            os.environ["SHARED_FIXTURE_TARGET"], "--from",
+                            os.environ["SHARED_FIXTURE_FROM"], "--message",
+                            "shared fixture handoff"], check=True, timeout=15,
+                           stdout=subprocess.DEVNULL)
+        subprocess.run([os.environ["SHARED_FIXTURE_CLI"], "messages", "send",
+                        "--channel", os.environ["SHARED_FIXTURE_CHANNEL"],
+                        "--reply-to", os.environ["SHARED_FIXTURE_FROM"],
+                        "--content", "shared fixture completed"], check=True,
+                       timeout=15, stdout=subprocess.DEVNULL)
+        send({"jsonrpc": "2.0", "method": "session/update", "params": {
+            "sessionId": "shared-fixture", "update": {
+                "sessionUpdate": "agent_message_chunk",
+                "content": {"type": "text", "text": "shared fixture completed"}}}})
+        result = {"stopReason": "end_turn"}
+    send({"jsonrpc": "2.0", "id": message["id"], "result": result})