feat(agents): launch remote sessions from ordinary mentions

.github/workflows/voltage-agent-images.yml

@@ -237,6 +237,7 @@ jobs:
         run: |
           docker run --rm "$IMAGE" --help
           docker run --rm "$IMAGE" reconcile --help
+          docker run --rm "$IMAGE" serve --help | grep -- '--mention-launch'
       - name: Relay smoke test
         if: matrix.image == 'buzz-relay'
         run: bash voltage/images/test-relay-image.sh "$IMAGE"
@@ -304,7 +305,9 @@ jobs:
           tags: ${{ env.IMAGE }}
       - name: Reaper smoke test
         if: matrix.image == 'buzz-sandbox-reaper'
-        run: docker run --rm "$IMAGE" --help
+        run: |
+          docker run --rm "$IMAGE" --help
+          docker run --rm "$IMAGE" serve --help | grep -- '--mention-launch'
       - name: Relay smoke test
         if: matrix.image == 'buzz-relay'
         run: bash voltage/images/test-relay-image.sh "$IMAGE"

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

@@ -37,6 +37,8 @@ pub struct State {
     pub fail_secret_creates: usize,
     /// Store the Secret, then fail this many create responses with a 500.
     pub fail_secret_create_responses: usize,
+    /// Fail Secret updates before writing, for durable inbox recovery tests.
+    pub fail_secret_updates: usize,
     /// Fail Pod-owned credential Secret creates, leaving recovery journals intact.
     pub fail_binding_secret_creates: usize,
     /// Fail this many Sandbox creates with a 500.
@@ -186,6 +188,10 @@ impl State {
                 ok(&json!({"apiVersion":"v1","kind":"Status","status":"Success"}))
             }
             ["api", "v1", "namespaces", _, "secrets", name] if method == Method::PUT => {
+                if self.fail_secret_updates > 0 {
+                    self.fail_secret_updates -= 1;
+                    return status(500, "InternalError");
+                }
                 let Some(prior) = self.secrets.get(*name) else {
                     return status(404, "NotFound");
                 };
@@ -197,6 +203,8 @@ impl State {
                     return status(409, "Conflict");
                 }
                 secret.metadata.uid = prior["metadata"]["uid"].as_str().map(str::to_owned);
+                secret.metadata.creation_timestamp =
+                    serde_json::from_value(prior["metadata"]["creationTimestamp"].clone()).unwrap();
                 secret.metadata.resource_version = Some(self.next_rv());
                 secret.data.get_or_insert_with(Default::default).extend(
                     secret
@@ -214,6 +222,19 @@ impl State {
                 Some(secret) => ok(secret),
                 None => status(404, "NotFound"),
             },
+            ["api", "v1", "namespaces", namespace, "secrets"] if method == Method::GET => {
+                let items: Vec<_> = self
+                    .secrets
+                    .values()
+                    .filter(|secret| {
+                        secret["metadata"]["namespace"] == *namespace
+                            && secret["metadata"]["labels"]["buzz.voltage.cloud/mobile-launch"]
+                                == "v1"
+                    })
+                    .cloned()
+                    .collect();
+                ok(&json!({"apiVersion":"v1", "kind":"SecretList", "metadata":{}, "items":items}))
+            }
             ["api", "v1", "namespaces", _, "secrets"] if method == Method::POST => {
                 if self.fail_secret_creates > 0 {
                     self.fail_secret_creates -= 1;
@@ -241,6 +262,9 @@ impl State {
                 });
                 secret.metadata.resource_version = Some(self.next_rv());
                 secret.metadata.uid = Some(self.next_uid("secret"));
+                secret.metadata.creation_timestamp = Some(
+                    k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(chrono::Utc::now()),
+                );
                 let stored = serde_json::to_value(&secret).unwrap();
                 self.secrets.insert(name, stored.clone());
                 if self.fail_secret_create_responses > 0 {

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

@@ -14,6 +14,7 @@ mod config;
 mod enroll;
 mod evidence;
 mod manager;
+mod mentions;
 mod mobile_launch;
 mod plan;
 mod profile;

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

@@ -421,6 +421,53 @@ fn op_matches(op: Op, request: &Request) -> bool {
 }
 
 impl Manager {
+    /// Registry discovery uses only the manager's configured namespaces.
+    pub(crate) fn mention_bindings(&self) -> impl Iterator<Item = (PublicKey, &Developer)> {
+        self.config
+            .developers
+            .iter()
+            .map(|(owner, developer)| (*owner, developer))
+    }
+
+    /// Share the confined manager's existing Kubernetes client with its inbox.
+    pub(crate) fn cluster(&self) -> kube::Client {
+        self.client.clone()
+    }
+
+    /// Reuse mobile start validation without manufacturing an owner control event.
+    pub(crate) async fn start_mention(
+        &self,
+        owner: PublicKey,
+        agent: PublicKey,
+        relay: &str,
+        event: &Event,
+    ) -> serde_json::Value {
+        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(),
+        ) {
+            Ok(start) => start,
+            Err(error) => return error,
+        };
+        Work::Start {
+            start,
+            target: Target::Managed {
+                client: self.client.clone(),
+                namespace: developer.namespace.clone(),
+                agent,
+                owner,
+                defaults: self.defaults.clone(),
+                env: self.config.launch_env(developer),
+            },
+        }
+        .run(self)
+        .await
+    }
+
     pub fn new(
         keys: Keys,
         config: ManagerConfig,

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

@@ -0,0 +1,106 @@
+//! Launch registered remote workspaces from ordinary, owner-signed mentions.
+//! Clients need no control protocol. Agent-authenticated subscriptions retain
+//! the relay's channel access checks; credentials never leave the manager.
+
+mod journal;
+mod registry;
+mod worker;
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::Semaphore;
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+
+use crate::manager::Manager;
+
+const MAX_AGENTS: usize = 64;
+const DISCOVERY_INTERVAL: Duration = Duration::from_secs(30);
+pub(super) const IO_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Stable labels only: upstream errors can contain credentials or message text.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub(super) enum Error {
+    #[error("registration_invalid")]
+    Registration,
+    #[error("inbox_storage_failed")]
+    Storage,
+    #[error("inbox_lease_unavailable")]
+    Lease,
+    #[error("mention_capacity_exceeded")]
+    Capacity,
+    #[error("mention_relay_failed")]
+    Relay,
+    #[error("mention_deadline_exceeded")]
+    Deadline,
+}
+
+/// Reconcile the bounded set of agent listeners. Removing/changing a registry
+/// entry cancels and joins its old worker before starting a replacement.
+pub(crate) async fn run(manager: Arc<Manager>, relay: &str, stop: CancellationToken) {
+    let permits = Arc::new(Semaphore::new(4));
+    let mut workers: BTreeMap<String, (String, JoinHandle<()>)> = BTreeMap::new();
+    let holder = uuid::Uuid::new_v4().to_string();
+    let activated = nostr::Timestamp::now().as_secs();
+    loop {
+        let discovered = tokio::select! {
+            () = stop.cancelled() => break,
+            result = tokio::time::timeout(IO_TIMEOUT, registry::discover(&manager, relay)) => result.unwrap_or(Err(Error::Deadline)),
+        };
+        match discovered {
+            Ok(identities) => {
+                let retired: Vec<_> = workers
+                    .iter()
+                    .filter(|(name, (revision, handle))| {
+                        handle.is_finished()
+                            || identities
+                                .get(*name)
+                                .is_none_or(|identity| &identity.revision != revision)
+                    })
+                    .map(|(name, _)| name.clone())
+                    .collect();
+                for name in retired {
+                    if let Some((_, handle)) = workers.remove(&name) {
+                        handle.abort();
+                        let _ = handle.await;
+                    }
+                }
+                for (name, identity) in identities {
+                    if workers.contains_key(&name) {
+                        continue;
+                    }
+                    let revision = identity.revision.clone();
+                    let manager = Arc::clone(&manager);
+                    let permits = Arc::clone(&permits);
+                    let holder = holder.clone();
+                    let handle = tokio::spawn(async move {
+                        worker::run(manager, identity, &holder, activated, permits).await;
+                    });
+                    workers.insert(name, (revision, handle));
+                }
+            }
+            Err(error) => {
+                // Fail closed when revocations cannot be observed. Durable
+                // pending work and cursors remain available after recovery.
+                tracing::warn!(%error, "mention registry unavailable; suspending listeners");
+                for (_, (_, handle)) in std::mem::take(&mut workers) {
+                    handle.abort();
+                    let _ = handle.await;
+                }
+            }
+        }
+        tokio::select! {
+            () = stop.cancelled() => break,
+            () = tokio::time::sleep(DISCOVERY_INTERVAL) => {},
+        }
+    }
+    for (_, (_, handle)) in workers {
+        handle.abort();
+        let _ = handle.await;
+    }
+}
+
+#[cfg(test)]
+mod tests;