diff --git a/.github/workflows/voltage-agent-images.yml b/.github/workflows/voltage-agent-images.yml index 68d1e3e3800..455651cea17 100644 --- a/.github/workflows/voltage-agent-images.yml +++ b/.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" diff --git a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs index c268a8c6cff..97377a24420 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs +++ b/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 { diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index 788f1ed8f47..e465564e7a7 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/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; diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index 96dac11553a..f3a4bec686e 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/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 { + 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, diff --git a/crates/buzz-sandbox-reaper/src/mentions.rs b/crates/buzz-sandbox-reaper/src/mentions.rs new file mode 100644 index 00000000000..bb343091f4d --- /dev/null +++ b/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, relay: &str, stop: CancellationToken) { + let permits = Arc::new(Semaphore::new(4)); + let mut workers: BTreeMap)> = 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; diff --git a/crates/buzz-sandbox-reaper/src/mentions/journal.rs b/crates/buzz-sandbox-reaper/src/mentions/journal.rs new file mode 100644 index 00000000000..fbc05aded3e --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/journal.rs @@ -0,0 +1,197 @@ +//! One CAS-protected inbox and lease per registered identity and community. + +use std::collections::BTreeMap; + +use k8s_openapi::api::core::v1::Secret; +use kube::{api::PostParams, Api}; +use nostr::{Event, Timestamp}; +use serde::{Deserialize, Serialize}; + +use super::{registry::Identity, Error}; + +const LABEL: &str = "buzz.voltage.cloud/mention-inbox"; +const DATA: &str = "inbox.json"; +const MAX_BYTES: usize = 512 * 1024; +const MAX_RECEIPTS: usize = 4096; +pub(super) const LEASE_SECONDS: u64 = 360; +pub(super) const OVERLAP_SECONDS: u64 = 120; + +#[derive(Deserialize, Serialize)] +pub(super) struct Pending { + pub event: Event, + pub attempts: u8, + pub retry_at: u64, + pub notice: Option, +} + +#[derive(Deserialize, Serialize)] +pub(super) struct Notice { + pub event: Event, + pub attempts: u8, +} + +#[derive(Deserialize, Serialize)] +pub(super) struct State { + version: u32, + holder: String, + lease_until: u64, + /// Events older than first activation never trigger historical launches. + pub activated: u64, + /// Advance only after a complete historical subscription, never per event. + pub since: u64, + pub completed: BTreeMap, + pub pending: Option, + /// Undeliverable replies must not block other channels' launches. + #[serde(default)] + pub outbox: Vec, +} + +pub(super) struct Journal { + api: Api, + secret: Secret, + holder: String, + pub state: State, +} + +impl Journal { + pub async fn acquire( + client: kube::Client, + identity: &Identity, + holder: &str, + activated: u64, + ) -> Result { + let api: Api = Api::namespaced(client, &identity.namespace); + let name = identity + .name + .replacen("buzz-mobile-launch-", "buzz-mention-inbox-", 1); + let prior = api.get_opt(&name).await.map_err(|_| Error::Storage)?; + let now = Timestamp::now().as_secs(); + let (secret, state) = 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::Storage); + } + let bytes = &secret + .data + .as_ref() + .and_then(|d| d.get(DATA)) + .ok_or(Error::Storage)? + .0; + if bytes.len() > MAX_BYTES { + return Err(Error::Capacity); + } + let state: State = serde_json::from_slice(bytes).map_err(|_| Error::Storage)?; + if state.version != 1 || state.completed.len() > MAX_RECEIPTS || state.outbox.len() > 32 + { + return Err(Error::Storage); + } + if state.holder != holder && state.lease_until > now { + return Err(Error::Lease); + } + (secret, state) + } else { + let mut secret = Secret { + type_: Some("Opaque".into()), + ..Default::default() + }; + secret.metadata.name = Some(name); + secret.metadata.namespace = Some(identity.namespace.clone()); + secret.metadata.labels = Some(BTreeMap::from([(LABEL.into(), "v1".into())])); + ( + secret, + State { + version: 1, + holder: holder.into(), + lease_until: 0, + activated, + since: activated, + completed: BTreeMap::new(), + pending: None, + outbox: Vec::new(), + }, + ) + }; + let mut journal = Self { + api, + secret, + holder: holder.into(), + state, + }; + journal.save().await?; + Ok(journal) + } + + /// CAS fences a previous holder even if it completes after its lease expires. + pub async fn save(&mut self) -> Result<(), Error> { + self.state.holder.clone_from(&self.holder); + self.state.lease_until = Timestamp::now().as_secs() + LEASE_SECONDS; + let text = serde_json::to_string(&self.state).map_err(|_| Error::Storage)?; + if text.len() > MAX_BYTES + || self.state.completed.len() > MAX_RECEIPTS + || self.state.outbox.len() > 32 + { + return Err(Error::Capacity); + } + self.secret.data = None; + self.secret.string_data = Some(BTreeMap::from([(DATA.into(), text)])); + let name = self.secret.metadata.name.as_deref().ok_or(Error::Storage)?; + self.secret = if self.secret.metadata.resource_version.is_some() { + self.api + .replace(name, &PostParams::default(), &self.secret) + .await + } else { + self.api.create(&PostParams::default(), &self.secret).await + } + .map_err(|e| match e { + kube::Error::Api(e) if e.code == 409 => Error::Lease, + _ => Error::Storage, + })?; + Ok(()) + } + + pub async fn enqueue(&mut self, event: Event) -> Result { + if self.state.completed.contains_key(&event.id.to_hex()) { + return Ok(false); + } + if self.state.pending.is_some() || self.state.completed.len() >= MAX_RECEIPTS { + return Err(Error::Capacity); + } + self.state.pending = Some(Pending { + event, + attempts: 0, + retry_at: 0, + notice: None, + }); + self.save().await?; + Ok(true) + } + + pub async fn complete(&mut self) -> Result<(), Error> { + let pending = self.state.pending.take().ok_or(Error::Storage)?; + if let Some(event) = pending.notice { + self.state.outbox.push(Notice { event, attempts: 0 }); + } + self.state.completed.insert( + pending.event.id.to_hex(), + pending.event.created_at.as_secs(), + ); + self.save().await + } + + pub async fn checkpoint(&mut self, connected_at: u64) -> Result<(), Error> { + self.state.since = self + .state + .since + .max(connected_at.saturating_sub(OVERLAP_SECONDS)); + self.state + .completed + .retain(|_, timestamp| *timestamp >= self.state.since); + self.save().await + } +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/registry.rs b/crates/buzz-sandbox-reaper/src/mentions/registry.rs new file mode 100644 index 00000000000..90b5fc0db7c --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/registry.rs @@ -0,0 +1,145 @@ +//! Only desktop-registered identities on the configured relay may be watched. + +use std::collections::BTreeMap; + +use k8s_openapi::api::core::v1::Secret; +use kube::{api::ListParams, Api}; +use nostr::{Keys, PublicKey, Tag, Timestamp}; + +use super::{Error, MAX_AGENTS}; +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 name: String, + pub namespace: String, + pub revision: String, + pub registered_at: u64, + pub owner: PublicKey, + pub keys: Keys, + pub auth: Tag, + pub relay: String, +} + +impl Identity { + pub fn read( + secret: Secret, + namespace: &str, + owner: PublicKey, + relay: &str, + ) -> Result { + let text = secret + .data + .as_ref() + .and_then(|data| data.get(DATA_KEY)) + .ok_or(Error::Registration)?; + let saved: Registration = + serde_json::from_slice(&text.0).map_err(|_| Error::Registration)?; + let key = saved.payload["private_key_nsec"] + .as_str() + .ok_or(Error::Registration)?; + let keys = Keys::parse(key).map_err(|_| Error::Registration)?; + let auth = saved.payload["auth_tag"] + .as_str() + .ok_or(Error::Registration)?; + let expected_relay = relay_endpoint(relay).map_err(|_| Error::Registration)?; + if saved.version != 1 + || saved.owner != owner + || saved.agent != keys.public_key() + || saved.relay != expected_relay + || relay_endpoint( + saved.payload["relay_url"] + .as_str() + .ok_or(Error::Registration)?, + ) + .map_err(|_| Error::Registration)? + != expected_relay + || buzz_sdk::nip_oa::verify_auth_tag_for_auth_event( + auth, + &saved.agent, + Timestamp::now().as_secs(), + ) + .map_err(|_| Error::Registration)? + != owner + || secret.metadata.namespace.as_deref() != Some(namespace) + || secret + .metadata + .labels + .as_ref() + .and_then(|labels| labels.get(LABEL)) + .map(String::as_str) + != Some("v1") + { + return Err(Error::Registration); + } + let name = Registration::name(owner, saved.agent, &expected_relay); + if secret.metadata.name.as_deref() != Some(&name) { + return Err(Error::Registration); + } + Ok(Self { + name, + namespace: namespace.into(), + owner, + keys, + revision: secret + .metadata + .resource_version + .ok_or(Error::Registration)?, + registered_at: secret + .metadata + .creation_timestamp + .as_ref() + .and_then(|time| u64::try_from(time.0.timestamp()).ok()) + .ok_or(Error::Registration)?, + auth: buzz_sdk::nip_oa::parse_auth_tag(auth).map_err(|_| Error::Registration)?, + relay: expected_relay, + }) + } +} + +pub(super) async fn discover( + manager: &Manager, + relay: &str, +) -> Result, Error> { + let mut found = BTreeMap::new(); + for (owner, developer) in manager.mention_bindings() { + let api: Api = Api::namespaced(manager.cluster(), &developer.namespace); + let list = api + .list( + &ListParams::default() + .labels(&format!("{LABEL}=v1")) + .limit((MAX_AGENTS + 1) as u32), + ) + .await + .map_err(|_| Error::Storage)?; + if list + .metadata + .continue_ + .as_ref() + .is_some_and(|v| !v.is_empty()) + || found.len() + list.items.len() > MAX_AGENTS + { + return Err(Error::Capacity); + } + for secret in list.items { + // Other community registrations are normal; never connect to their URLs. + let raw = secret + .data + .as_ref() + .and_then(|d| d.get(DATA_KEY)) + .ok_or(Error::Registration)?; + let saved: Registration = + serde_json::from_slice(&raw.0).map_err(|_| Error::Registration)?; + if saved.relay != relay_endpoint(relay).map_err(|_| Error::Registration)? { + continue; + } + let identity = Identity::read(secret, &developer.namespace, owner, relay)?; + found.insert( + format!("{}/{}", identity.namespace, identity.name), + identity, + ); + } + } + Ok(found) +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests.rs b/crates/buzz-sandbox-reaper/src/mentions/tests.rs new file mode 100644 index 00000000000..ccae768337b --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests.rs @@ -0,0 +1,621 @@ +//! A black-box client sends only ordinary signed chat events. A real WebSocket +//! drives the production supervisor/worker and provider against a fake cluster. + +mod live_relay; +mod relay; +mod revoked; + +use std::sync::Arc; +use std::time::Duration; + +use buzz_backend_kubernetes::sandbox_fake; +use buzz_ws_client::NostrWsConnection; +use nostr::{nips::nip19::ToBech32, Event, EventBuilder, Keys, Kind, Tag, Timestamp}; +use serde_json::{json, Value}; +use tokio::sync::Semaphore; +use tokio_util::sync::CancellationToken; + +use super::{journal::Journal, registry::Identity, *}; +use crate::manager::{Manager, ManagerConfig}; +use crate::mobile_launch::Registration; + +const CHANNEL: &str = "0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10"; +const NS: &str = "buzz-agent-staging-dev"; +const BOUND: Duration = Duration::from_secs(20); + +fn keys(byte: &str) -> Keys { + Keys::parse(&byte.repeat(32)).unwrap() +} + +fn manager(cluster: &sandbox_fake::Shared) -> Arc { + let config = ManagerConfig::parse(&json!({"version":1,"checkpoint_bucket":"fixture-checkpoints", "developers":{"dev":{ + "owner_pubkey":keys("33").public_key(), "namespace":NS, + "profile_secret_arn":"arn:aws:secretsmanager:us-west-2:000000000000:secret:voltage/staging/agents/dev/profile-AbCdEf" + }}}).to_string()).unwrap(); + Arc::new( + Manager::new( + keys("11"), + config, + sandbox_fake::client(cluster), + crate::enroll::tests::offline(Default::default()), + ) + .with_sandbox_defaults( + json!({"image":format!("example.invalid/agent@sha256:{}", "a".repeat(64)), + "pod_options":{"active_deadline_seconds":3600}}) + .as_object() + .unwrap() + .clone(), + ), + ) +} + +async fn register(cluster: &sandbox_fake::Shared, relay: &str) -> Identity { + let owner = keys("33"); + let agent = keys("22"); + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap(); + Registration::new(owner.public_key(), agent.public_key(), json!({ + "relay_url":relay, "private_key_nsec":agent.secret_key().to_bech32().unwrap(), "auth_tag":auth, + "launch":{"command":"codex-acp","args":[],"env":{}, + "policy_env":{"BUZZ_ACP_SESSION_POLICY":"thread"},"owner_pubkey":owner.public_key()} + })).unwrap().save(sandbox_fake::client(cluster), NS).await.unwrap(); + registry::discover(&manager(cluster), relay) + .await + .unwrap() + .pop_first() + .unwrap() + .1 +} + +fn message(content: &str, root: Option<&str>) -> Event { + let mut tags = vec![ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + ]; + if let Some(root) = root { + tags.push(Tag::parse(["e", root, "", "reply"]).unwrap()); + } + EventBuilder::new(Kind::Custom(9), content) + .tags(tags) + .sign_with_keys(&keys("33")) + .unwrap() +} + +async fn wait_for(mut condition: impl FnMut() -> bool) { + tokio::time::timeout(BOUND, async { + while !condition() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn unchanged_client_starts_new_threads_and_reuses_replies_without_control_requests() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + register(&cluster, &relay.url).await; + let stop = CancellationToken::new(); + let server = { + let manager = manager(&cluster); + let stop = stop.clone(); + let url = relay.url.clone(); + tokio::spawn(async move { + crate::serve::run_services(manager, &url, None, true, stop.cancelled()).await + }) + }; + relay.wait_subscription("mentions-history").await; + let mut phone = NostrWsConnection::connect_authenticated(&relay.url, &keys("33"), None) + .await + .unwrap(); + let first = message("@agent first task", None); + assert!(phone.send_event(first.clone()).await.unwrap().accepted); + wait_for(|| cluster.lock().unwrap().sandboxes.len() == 1).await; + let reply = message("@agent continue", Some(&first.id.to_hex())); + assert!(phone.send_event(reply.clone()).await.unwrap().accepted); + assert!(phone.send_event(first.clone()).await.unwrap().accepted); + let second = message("@agent another task", None); + assert!(phone.send_event(second.clone()).await.unwrap().accepted); + wait_for(|| cluster.lock().unwrap().sandboxes.len() == 2).await; + // Wait for the durable receipt, not merely the provider's first write. + wait_for(|| completed(&cluster, &second)).await; + assert!(completed(&cluster, &reply)); + assert!(relay + .events + .lock() + .unwrap() + .iter() + .all(|e| e.kind.as_u16() != 24210)); + stop.cancel(); + tokio::time::timeout(BOUND, server).await.unwrap().unwrap(); + relay.stop().await; +} + +fn inbox(cluster: &sandbox_fake::Shared) -> Value { + use base64::Engine; + cluster + .lock() + .unwrap() + .secrets + .values() + .find_map(|secret| { + secret["data"]["inbox.json"].as_str().map(|value| { + let bytes = base64::engine::general_purpose::STANDARD + .decode(value) + .unwrap(); + serde_json::from_slice(&bytes).unwrap() + }) + }) + .unwrap_or(Value::Null) +} + +fn completed(cluster: &sandbox_fake::Shared, event: &Event) -> bool { + inbox(cluster)["completed"].get(event.id.to_hex()).is_some() +} + +#[tokio::test] +async fn durable_pending_launch_survives_restart_and_storage_failure_prevents_deploy() { + 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, + "same-process", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let event = message("@agent persist first", None); + relay.events.lock().unwrap().push(event.clone()); + cluster.lock().unwrap().fail_secret_updates = 1; + assert_eq!( + tokio::time::timeout( + BOUND, + worker::session( + &manager(&cluster), + &identity, + &mut journal, + &Semaphore::new(1) + ) + ) + .await + .unwrap(), + Err(Error::Storage) + ); + assert!(cluster.lock().unwrap().sandboxes.is_empty()); + let mut restored = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "same-process", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + assert!(restored.state.pending.is_none()); + restored.enqueue(event.clone()).await.unwrap(); + drop(restored); + let mut restored = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "same-process", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let manager = manager(&cluster); + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut restored, &Semaphore::new(1)).await + }); + wait_for(|| completed(&cluster, &event)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + task.abort(); + let _ = task.await; + relay.stop().await; +} + +#[tokio::test] +async fn leases_and_resource_versions_fence_competing_workers() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let mut old = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "one", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + assert!(matches!( + Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "two", + Timestamp::now().as_secs() + ) + .await, + Err(Error::Lease) + )); + // Same-holder recreation is how the supervisor resumes an aborted worker. + let mut next = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "one", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + next.enqueue(message("@agent next", None)).await.unwrap(); + assert_eq!( + old.enqueue(message("@agent stale", None)).await, + Err(Error::Lease) + ); + relay.stop().await; +} + +#[tokio::test] +async fn catchup_handles_disconnect_and_duplicate_history_without_starting_old_messages() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let journal = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let old = EventBuilder::new(Kind::Custom(9), "@agent before activation") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + ]) + .custom_created_at(Timestamp::from(journal.state.activated - 10)) + .sign_with_keys(&keys("33")) + .unwrap(); + let event = message("@agent while disconnected", None); + relay.events.lock().unwrap().extend([old, event.clone()]); + for _ in 0..2 { + let previous = relay.count("mentions-history"); + let mut journal = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let identity = register(&cluster, &relay.url).await; + let manager = manager(&cluster); + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }); + relay.wait_after("mentions-history", previous).await; + wait_for(|| completed(&cluster, &event)).await; + task.abort(); + let _ = task.await; + } + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + relay.stop().await; +} + +#[tokio::test] +async fn tampered_mentions_and_display_references_never_launch() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let now = Timestamp::now().as_secs(); + let mut journal = Journal::acquire(sandbox_fake::client(&cluster), &identity, "test", now - 20) + .await + .unwrap(); + let mut tampered = message("@agent forged", None); + tampered.content = "tampered after signing".into(); + let reference = EventBuilder::new(Kind::Custom(9), "about an agent") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::parse(["mention", &keys("22").public_key().to_hex()]).unwrap(), + ]) + .sign_with_keys(&keys("33")) + .unwrap(); + let other_owner = EventBuilder::new(Kind::Custom(9), "@agent unauthorized") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + ]) + .sign_with_keys(&keys("44")) + .unwrap(); + let valid = EventBuilder::new(Kind::Custom(9), "@agent valid") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + ]) + .custom_created_at(Timestamp::from(now - 5)) + .sign_with_keys(&keys("33")) + .unwrap(); + relay + .events + .lock() + .unwrap() + .extend([tampered, reference, other_owner, valid.clone()]); + let manager = manager(&cluster); + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }); + wait_for(|| completed(&cluster, &valid)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + task.abort(); + let _ = task.await; + relay.stop().await; +} + +#[tokio::test] +async fn registration_binds_owner_relay_key_and_attestation() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let secret: k8s_openapi::api::core::v1::Secret = + serde_json::from_value(cluster.lock().unwrap().secrets[&identity.name].clone()).unwrap(); + assert!(Identity::read(secret.clone(), NS, keys("44").public_key(), &relay.url).is_err()); + assert!(Identity::read( + secret.clone(), + NS, + keys("33").public_key(), + "wss://another.community" + ) + .is_err()); + for field in ["private_key_nsec", "auth_tag"] { + let mut corrupt = secret.clone(); + let data = &mut corrupt + .data + .as_mut() + .unwrap() + .get_mut("launch.json") + .unwrap() + .0; + let mut value: Value = serde_json::from_slice(data).unwrap(); + value["payload"][field] = json!("invalid"); + *data = serde_json::to_vec(&value).unwrap(); + assert!(Identity::read(corrupt, NS, keys("33").public_key(), &relay.url).is_err()); + } + relay.stop().await; +} + +#[tokio::test] +async fn ended_session_posts_an_ordinary_reply_without_recovering_the_sandbox() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let manager = manager(&cluster); + let event = message("@agent original task", None); + assert_eq!( + manager + .start_mention( + identity.owner, + identity.keys.public_key(), + &identity.relay, + &event + ) + .await["ok"], + true + ); + { + let mut state = cluster.lock().unwrap(); + for sandbox in state.sandboxes.values_mut() { + sandbox["metadata"]["annotations"][buzz_backend_kubernetes::lifecycle::LIFECYCLE] = + json!("ended"); + } + state.pods.clear(); + } + let mut journal = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let retry = message("@agent continue ended session", Some(&event.id.to_hex())); + relay.events.lock().unwrap().push(retry.clone()); + journal.enqueue(retry.clone()).await.unwrap(); + relay + .lose_next_ack + .store(true, std::sync::atomic::Ordering::SeqCst); + let first_identity = register(&cluster, &relay.url).await; + let first_manager = Arc::clone(&manager); + let previous = relay.count("mentions-history"); + let first = tokio::spawn(async move { + worker::session( + &first_manager, + &first_identity, + &mut journal, + &Semaphore::new(1), + ) + .await + }); + relay.wait_after("mentions-history", previous).await; + assert!(completed(&cluster, &retry)); + first.abort(); + let _ = first.await; + let mut journal = Journal::acquire( + sandbox_fake::client(&cluster), + &identity, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + assert!(journal.state.pending.is_none()); + assert_eq!(journal.state.outbox.len(), 1); + let notice_id = journal.state.outbox[0].event.id; + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }); + wait_for(|| { + inbox(&cluster)["outbox"] + .as_array() + .is_some_and(|v| v.is_empty()) + }) + .await; + let notices: Vec<_> = relay + .events + .lock() + .unwrap() + .iter() + .filter(|e| e.pubkey == keys("22").public_key()) + .cloned() + .collect(); + assert_eq!(notices.len(), 1); + assert_eq!(notices[0].id, notice_id); + assert!(notices[0].content.contains("Recover it from desktop")); + assert_eq!(notices[0].kind.as_u16(), 9); + assert!(notices[0] + .tags + .iter() + .any(|tag| tag.as_slice() == ["e", &event.id.to_hex(), "", "root"])); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + assert!(cluster.lock().unwrap().pods.is_empty()); + task.abort(); + let _ = task.await; + relay.stop().await; +} + +#[tokio::test] +async fn dm_metadata_produces_a_channel_instruction_without_starting_a_thread_sandbox() { + 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, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let event = message("@agent direct message", None); + let dm = EventBuilder::new(Kind::Custom(39000), "") + .tags([ + Tag::parse(["d", CHANNEL]).unwrap(), + Tag::parse(["t", "dm"]).unwrap(), + ]) + .sign_with_keys(&keys("11")) + .unwrap(); + { + let mut events = relay.events.lock().unwrap(); + events.retain(|event| event.kind.as_u16() == 30177); + events.extend([dm, event.clone()]); + } + journal.enqueue(event.clone()).await.unwrap(); + let manager = manager(&cluster); + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }); + wait_for(|| { + completed(&cluster, &event) + && inbox(&cluster)["outbox"] + .as_array() + .is_some_and(|v| v.is_empty()) + }) + .await; + assert!(cluster.lock().unwrap().sandboxes.is_empty()); + assert!(relay + .events + .lock() + .unwrap() + .iter() + .any(|event| event.pubkey == keys("22").public_key() + && event.content.contains("requires a channel thread"))); + task.abort(); + let _ = task.await; + relay.stop().await; +} + +#[tokio::test] +async fn saturated_history_never_advances_its_cursor_past_unread_events() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let now = Timestamp::now().as_secs(); + let mut journal = + Journal::acquire(sandbox_fake::client(&cluster), &identity, "test", now - 300) + .await + .unwrap(); + let before = journal.state.since; + // Invalid channel bindings cannot launch, but still consume the relay's + // history limit. Every event has a distinct signed ID at the same second. + let events: Vec<_> = (0..1000) + .map(|i| { + EventBuilder::new(Kind::Custom(9), format!("{i}")) + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + ]) + .custom_created_at(Timestamp::from(now - 5)) + .sign_with_keys(&keys("33")) + .unwrap() + }) + .collect(); + relay.events.lock().unwrap().extend(events); + assert_eq!( + tokio::time::timeout( + BOUND, + worker::session( + &manager(&cluster), + &identity, + &mut journal, + &Semaphore::new(1) + ) + ) + .await + .unwrap(), + Err(Error::Capacity) + ); + let restored = Journal::acquire(sandbox_fake::client(&cluster), &identity, "test", now) + .await + .unwrap(); + assert_eq!(restored.state.since, before); + assert!(cluster.lock().unwrap().sandboxes.is_empty()); + relay.stop().await; +} + +#[tokio::test] +async fn undeliverable_notice_does_not_block_a_new_channel_launch() { + 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, + "test", + Timestamp::now().as_secs(), + ) + .await + .unwrap(); + let notice = EventBuilder::new(Kind::Custom(9), "startup failed in a removed channel") + .tags([Tag::parse(["h", "87dc5d01-5fb3-407e-936a-dc1d94ab192e"]).unwrap()]) + .sign_with_keys(&keys("22")) + .unwrap(); + journal.state.outbox.push(journal::Notice { + event: notice, + attempts: 0, + }); + journal.save().await.unwrap(); + let event = message("@agent new task despite failed notice", None); + relay.events.lock().unwrap().push(event.clone()); + journal.enqueue(event.clone()).await.unwrap(); + let manager = manager(&cluster); + let task = tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }); + relay.wait_subscription("mentions-history").await; + assert!(completed(&cluster, &event)); + assert_eq!(inbox(&cluster)["outbox"].as_array().unwrap().len(), 1); + assert_eq!(inbox(&cluster)["outbox"][0]["attempts"], 1); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + task.abort(); + let _ = task.await; + relay.stop().await; +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/live_relay.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/live_relay.rs new file mode 100644 index 00000000000..528f42d1dd0 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/live_relay.rs @@ -0,0 +1,118 @@ +//! Run with BUZZ_MENTION_TEST_RELAY pointing at an isolated local Buzz relay. +//! Uses real relay authentication, channel membership and subscription filters. + +use super::*; + +#[tokio::test] +#[ignore = "requires an isolated local relay with NIP-OA enabled"] +async fn unchanged_client_starts_agents_through_real_relay_permissions() { + let relay = std::env::var("BUZZ_MENTION_TEST_RELAY") + .expect("set BUZZ_MENTION_TEST_RELAY to an isolated local relay"); + let url = url::Url::parse(&relay).unwrap(); + assert_eq!(url.scheme(), "ws"); + assert!(matches!(url.host_str(), Some("127.0.0.1" | "localhost"))); + tokio::time::timeout(Duration::from_secs(60), exercise(&relay)) + .await + .expect("real-relay mention workflow timed out"); +} + +async fn exercise(relay: &str) { + let owner = keys("33"); + let agent = keys("22"); + let mut phone = NostrWsConnection::connect_authenticated(relay, &owner, None) + .await + .unwrap(); + let channel = uuid::Uuid::new_v4(); + for builder in [ + EventBuilder::new(Kind::Custom(30177), "{}").tags([ + Tag::parse(["d", &agent.public_key().to_hex()]).unwrap(), + Tag::parse(["runtime", "voltage.remote-workspace/sandbox"]).unwrap(), + ]), + buzz_sdk::build_create_channel( + channel, + &format!("phone-mention-{channel}"), + Some(buzz_sdk::Visibility::Private), + Some(buzz_sdk::ChannelKind::Stream), + None, + None, + ) + .unwrap(), + buzz_sdk::build_add_member(channel, &agent.public_key().to_hex(), None).unwrap(), + ] { + let result = phone + .send_event(builder.sign_with_keys(&owner).unwrap()) + .await + .unwrap(); + assert!(result.accepted, "channel setup refused: {}", result.message); + } + + let cluster = sandbox_fake::Shared::default(); + register(&cluster, relay).await; + let stop = CancellationToken::new(); + let server = { + let manager = manager(&cluster); + let stop = stop.clone(); + let relay = relay.to_owned(); + tokio::spawn(async move { + crate::serve::run_services(manager, &relay, None, true, stop.cancelled()).await; + }) + }; + let mention = |text: &str, root: Option<&Event>| { + let mut tags = vec![ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::public_key(agent.public_key()), + ]; + if let Some(root) = root { + tags.push(Tag::parse(["e", &root.id.to_hex(), "", "reply"]).unwrap()); + } + EventBuilder::new(Kind::Custom(9), text) + .tags(tags) + .sign_with_keys(&owner) + .unwrap() + }; + let first = mention("@agent phone task", None); + let result = phone.send_event(first.clone()).await.unwrap(); + assert!(result.accepted, "phone message refused: {}", result.message); + wait_for(|| completed(&cluster, &first)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + + let reply = mention("@agent phone follow-up", Some(&first)); + assert!(phone.send_event(reply.clone()).await.unwrap().accepted); + wait_for(|| completed(&cluster, &reply)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + + let second = mention("@agent new phone task", None); + assert!(phone.send_event(second.clone()).await.unwrap().accepted); + wait_for(|| completed(&cluster, &second)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 2); + assert!(inbox(&cluster)["outbox"].as_array().unwrap().is_empty()); + + let local = EventBuilder::new(Kind::Custom(30177), "{}") + .tags([ + Tag::parse(["d", &agent.public_key().to_hex()]).unwrap(), + Tag::parse(["runtime", "local/agent"]).unwrap(), + ]) + .custom_created_at(Timestamp::from(Timestamp::now().as_secs() + 1)) + .sign_with_keys(&owner) + .unwrap(); + let result = phone.send_event(local).await.unwrap(); + assert!( + result.accepted, + "runtime update refused: {}", + result.message + ); + let after_retype = mention("@agent now run locally", None); + assert!( + phone + .send_event(after_retype.clone()) + .await + .unwrap() + .accepted + ); + wait_for(|| completed(&cluster, &after_retype)).await; + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 2); + assert!(inbox(&cluster)["outbox"].as_array().unwrap().is_empty()); + + stop.cancel(); + tokio::time::timeout(BOUND, server).await.unwrap().unwrap(); +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs new file mode 100644 index 00000000000..f84c17ddf39 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs @@ -0,0 +1,205 @@ +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use futures_util::{SinkExt, StreamExt}; +use nostr::{Event, EventBuilder, Kind, Tag}; +use serde_json::{json, Value}; +use tokio::sync::{broadcast, watch}; +use tokio::task::{JoinHandle, JoinSet}; +use tokio_tungstenite::tungstenite::Message; +use tokio_util::sync::CancellationToken; + +use super::{keys, BOUND, CHANNEL}; + +pub(super) struct Relay { + pub url: String, + pub events: Arc>>, + pub lose_next_ack: Arc, + subscriptions: watch::Receiver>, + cancel: CancellationToken, + task: JoinHandle<()>, +} + +impl Relay { + pub async fn start() -> Self { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let metadata = EventBuilder::new(Kind::Custom(39000), "") + .tags([ + Tag::parse(["d", CHANNEL]).unwrap(), + Tag::parse(["t", "stream"]).unwrap(), + ]) + .sign_with_keys(&keys("11")) + .unwrap(); + let listing = EventBuilder::new(Kind::Custom(30177), "{}") + .tags([ + Tag::parse(["d", &keys("22").public_key().to_hex()]).unwrap(), + Tag::parse(["runtime", "voltage.remote-workspace/sandbox"]).unwrap(), + ]) + .sign_with_keys(&keys("33")) + .unwrap(); + let events = Arc::new(Mutex::new(vec![metadata, listing])); + let cancel = CancellationToken::new(); + let lose_next_ack = Arc::new(AtomicBool::new(false)); + let (subscriptions, receiver) = watch::channel(BTreeMap::new()); + let task = { + let events = events.clone(); + let cancel = cancel.clone(); + let lose_next_ack = lose_next_ack.clone(); + tokio::spawn(async move { + let mut tasks = JoinSet::new(); + let (hub, _) = broadcast::channel(256); + loop { + tokio::select! { + () = cancel.cancelled() => break, + Some(_) = tasks.join_next() => {}, + socket = listener.accept() => { + let (socket, _) = socket.unwrap(); + let events = events.clone(); let hub = hub.clone(); let subscriptions = subscriptions.clone(); + let lose_next_ack = lose_next_ack.clone(); + tasks.spawn(async move { let _ = connection(socket, events, hub, subscriptions, lose_next_ack).await; }); + } + } + } + tasks.shutdown().await; + }) + }; + Self { + url, + events, + lose_next_ack, + subscriptions: receiver, + cancel, + task, + } + } + + pub fn count(&self, id: &str) -> usize { + self.subscriptions.borrow().get(id).copied().unwrap_or(0) + } + + pub async fn wait_subscription(&self, id: &str) { + self.wait_after(id, 0).await; + } + + pub async fn wait_after(&self, id: &str, previous: usize) { + let mut receiver = self.subscriptions.clone(); + tokio::time::timeout( + BOUND, + receiver.wait_for(|v| v.get(id).copied().unwrap_or(0) > previous), + ) + .await + .unwrap() + .unwrap(); + } + + pub async fn stop(self) { + self.cancel.cancel(); + self.task.await.unwrap(); + } +} + +fn matches(filter: &Value, event: &Event) -> bool { + let contains = |name: &str, value: Value| { + filter[name] + .as_array() + .is_none_or(|values| values.contains(&value)) + }; + contains("kinds", json!(event.kind.as_u16())) + && contains("authors", json!(event.pubkey)) + && contains("ids", json!(event.id)) + && ["h", "d"].into_iter().all(|key| { + filter[format!("#{key}")].as_array().is_none_or(|values| { + event.tags.iter().any(|tag| { + tag.as_slice().first().map(String::as_str) == Some(key) + && tag + .as_slice() + .get(1) + .is_some_and(|value| values.contains(&json!(value))) + }) + }) + }) + && filter["#p"].as_array().is_none_or(|values| { + event.tags.iter().any(|tag| { + let tag = tag.as_slice(); + tag.first().map(String::as_str) == Some("p") + && tag.get(1).is_some_and(|key| values.contains(&json!(key))) + }) + }) + && filter["since"] + .as_u64() + .is_none_or(|since| event.created_at.as_secs() >= since) + && filter["until"] + .as_u64() + .is_none_or(|until| event.created_at.as_secs() <= until) +} + +async fn connection( + socket: tokio::net::TcpStream, + stored: Arc>>, + hub: broadcast::Sender, + subscriptions: watch::Sender>, + lose_next_ack: Arc, +) -> Result<(), Box> { + let mut ws = tokio_tungstenite::accept_async(socket).await?; + let frame = |value: Value| Message::Text(value.to_string().into()); + ws.send(frame(json!(["AUTH", "challenge"]))).await?; + let mut events = hub.subscribe(); + let mut author = None; + let mut filters = BTreeMap::::new(); + loop { + tokio::select! { + input = ws.next() => { + let Some(Ok(Message::Text(text))) = input else { return Ok(()) }; + let input: Value = serde_json::from_str(&text)?; + match input[0].as_str().unwrap() { + "AUTH" => { + let event: Event = serde_json::from_value(input[1].clone())?; + event.verify()?; + if event.pubkey == keys("22").public_key() { + let auth = event.tags.iter().find(|t| t.as_slice()[0] == "auth").unwrap(); + assert_eq!(buzz_sdk::nip_oa::verify_auth_tag(&serde_json::to_string(auth.as_slice())?, &event.pubkey)?, keys("33").public_key()); + } + author = Some(event.pubkey); + ws.send(frame(json!(["OK", event.id, true, ""]))).await?; + } + "REQ" => { + assert!(author.is_some()); + let id = input[1].as_str().unwrap().to_string(); + let filter = input[2].clone(); + let allowed = |event: &Event| matches!(event.kind.as_u16(), 24210 | 30177) || event.tags.iter().any(|tag| tag.as_slice() == ["h", CHANNEL] || tag.as_slice() == ["d", CHANNEL]); + let mut history: Vec<_> = stored.lock().unwrap().iter().filter(|e| matches(&filter,e) && allowed(e)).cloned().collect(); + history.sort_by_key(|e| std::cmp::Reverse((e.created_at, e.id))); + history.truncate(filter["limit"].as_u64().unwrap_or(1000) as usize); + filters.insert(id.clone(), filter); + for event in history { ws.send(frame(json!(["EVENT", id, event]))).await?; } + 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()); } + "EVENT" => { + let event: Event = serde_json::from_value(input[1].clone())?; + let accepted = Some(event.pubkey) == author && event.verify().is_ok() + && (event.kind.as_u16() == 24210 || event.tags.iter().any(|tag| tag.as_slice() == ["h",CHANNEL])); + if accepted { + let mut history = stored.lock().unwrap(); + if !history.iter().any(|e| e.id == event.id) { history.push(event.clone()); } + let _ = hub.send(event.clone()); + } + if accepted && lose_next_ack.swap(false, Ordering::SeqCst) { return Ok(()); } + ws.send(frame(json!(["OK",event.id,accepted,""]))).await?; + } + _ => panic!("unexpected client protocol"), + } + } + event = events.recv() => { + 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 matches(filter, &event) { ws.send(frame(json!(["EVENT",id,event]))).await?; } + } + } + } + } +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/revoked.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/revoked.rs new file mode 100644 index 00000000000..e719b683ced --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/revoked.rs @@ -0,0 +1,51 @@ +use super::*; + +#[tokio::test] +async fn deleted_retyped_and_ambiguous_listings_do_not_start_saved_registrations() { + for runtime in [None, Some("local/agent"), Some("ambiguous")] { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + let identity = register(&cluster, &relay.url).await; + let trigger = message("@agent use the current runtime", None); + { + let mut events = relay.events.lock().unwrap(); + events.retain(|event| event.kind.as_u16() != 30177); + if let Some(runtime) = runtime { + let mut tags = vec![ + Tag::parse(["d", &keys("22").public_key().to_hex()]).unwrap(), + Tag::parse(["runtime", runtime]).unwrap(), + ]; + if runtime == "ambiguous" { + tags.push(Tag::parse(["runtime", "voltage.remote-workspace/sandbox"]).unwrap()); + } + events.push( + EventBuilder::new(Kind::Custom(30177), "{}") + .tags(tags) + .sign_with_keys(&keys("33")) + .unwrap(), + ); + } + events.push(trigger.clone()); + } + let mut journal = Journal::acquire(sandbox_fake::client(&cluster), &identity, "worker", 0) + .await + .unwrap(); + // A trigger already persisted before revocation must be rechecked too. + journal.enqueue(trigger.clone()).await.unwrap(); + let task = { + let manager = manager(&cluster); + tokio::spawn(async move { + worker::session(&manager, &identity, &mut journal, &Semaphore::new(1)).await + }) + }; + wait_for(|| completed(&cluster, &trigger)).await; + assert!( + cluster.lock().unwrap().sandboxes.is_empty(), + "stale runtime {runtime:?} launched" + ); + assert!(inbox(&cluster)["outbox"].as_array().unwrap().is_empty()); + task.abort(); + let _ = task.await; + relay.stop().await; + } +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/worker.rs b/crates/buzz-sandbox-reaper/src/mentions/worker.rs new file mode 100644 index 00000000000..4653fe26786 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/worker.rs @@ -0,0 +1,453 @@ +use std::sync::Arc; +use std::time::Duration; + +use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; +use nostr::{Event, EventBuilder, Kind, Tag, Timestamp}; +use serde_json::json; +use tokio::sync::Semaphore; + +use super::journal::Journal; +use super::registry::Identity; +use super::{Error, IO_TIMEOUT}; +use crate::manager::Manager; +use crate::mobile_launch::Start; + +const PAGE: usize = 1000; +const MAX_PAGES: usize = 16; +const TURN_TIMEOUT: Duration = Duration::from_secs(180); +const RECYCLE_AFTER: Duration = Duration::from_secs(60); +const MAX_ATTEMPTS: u8 = 3; +const MAX_EVENT_BYTES: usize = 60_000; + +enum LaunchPermission { + Allowed, + UnsupportedConversation, + NoLongerRemote, +} + +// A worker must renew its fenced journal before its longest external effect +// can outlive the lease. IO includes delivery revalidation and result storage. +const _: () = + assert!(super::journal::LEASE_SECONDS > TURN_TIMEOUT.as_secs() + 3 * IO_TIMEOUT.as_secs()); + +pub(super) async fn run( + manager: Arc, + identity: Identity, + holder: &str, + activated: u64, + permits: Arc, +) { + let mut delay = Duration::from_secs(1); + loop { + let acquired = tokio::time::timeout( + IO_TIMEOUT, + Journal::acquire( + manager.cluster(), + &identity, + holder, + activated.max(identity.registered_at), + ), + ) + .await; + let result = match acquired { + Ok(Ok(mut journal)) => session(&manager, &identity, &mut journal, &permits).await, + Ok(Err(error)) => Err(error), + Err(_) => Err(Error::Deadline), + }; + match result { + Ok(()) => delay = Duration::from_secs(1), + Err(error) => { + tracing::warn!(%error, agent = %identity.keys.public_key(), "mention listener interrupted; durable inbox retained"); + // Saturation and corrupt registrations require operator repair. + // Discovery retries them at its bounded interval, never in a hot loop. + if matches!(error, Error::Capacity | Error::Registration) { + return; + } + delay = (delay * 2).min(Duration::from_secs(60)); + } + } + tokio::time::sleep(delay).await; + } +} + +async fn connect(identity: &Identity) -> Result { + tokio::time::timeout( + IO_TIMEOUT, + NostrWsConnection::connect_authenticated( + &identity.relay, + &identity.keys, + Some(&identity.auth), + ), + ) + .await + .map_err(|_| Error::Deadline)? + .map_err(|_| Error::Relay) +} + +async fn send(connection: &mut NostrWsConnection, value: serde_json::Value) -> Result<(), Error> { + tokio::time::timeout(IO_TIMEOUT, connection.send_raw(&value)) + .await + .map_err(|_| Error::Deadline)? + .map_err(|_| Error::Relay) +} + +/// Live delivery is installed before paged history. The cursor advances only +/// after every page reaches EOSE; a truncated second fails without skipping it. +pub(super) async fn session( + manager: &Manager, + identity: &Identity, + journal: &mut Journal, + permits: &Semaphore, +) -> Result<(), Error> { + drain(manager, identity, journal, permits).await?; + let mut connection = connect(identity).await?; + let connected_at = Timestamp::now().as_secs(); + let started = tokio::time::Instant::now(); + let filter = json!({"kinds": [9, 40002], "authors": [identity.owner], "#p": [identity.keys.public_key()]}); + let mut live = filter.clone(); + live["limit"] = json!(0); + send(&mut connection, json!(["REQ", "mentions-live", live])).await?; + let mut history_started = false; + let mut history_done = false; + let mut count = 0; + let mut oldest = connected_at; + let mut until = connected_at; + let mut pages = 0; + let mut progress = tokio::time::Instant::now(); + loop { + if history_done && started.elapsed() >= RECYCLE_AFTER { + return Ok(()); + } + if !history_done && progress.elapsed() > IO_TIMEOUT { + return Err(Error::Deadline); + } + let message = match connection.next_event(Duration::from_secs(5)).await { + Ok(message) => message, + Err(WsClientError::Timeout) => { + continue; + } + Err(_) => return Err(Error::Relay), + }; + match message { + RelayMessage::Event { + subscription_id, + event, + } if matches!( + subscription_id.as_str(), + "mentions-live" | "mentions-history" + ) => + { + if subscription_id == "mentions-history" { + count += 1; + if count > PAGE { + return Err(Error::Capacity); + } + oldest = oldest.min(event.created_at.as_secs()); + } + if eligible(identity, journal, &event) { + let persisted = tokio::time::timeout(IO_TIMEOUT, journal.enqueue(*event)) + .await + .map_err(|_| Error::Deadline)??; + if persisted { + drain(manager, identity, journal, permits).await?; + } + } + if subscription_id == "mentions-history" { + progress = tokio::time::Instant::now(); + } + } + RelayMessage::Eose { subscription_id } + if subscription_id == "mentions-live" && !history_started => + { + history_started = true; + history(&mut connection, filter.clone(), journal.state.since, until).await?; + progress = tokio::time::Instant::now(); + } + RelayMessage::Eose { subscription_id } + if subscription_id == "mentions-history" && !history_done => + { + pages += 1; + if count == PAGE { + // Inclusive boundary preserves all events sharing a timestamp. + // If a whole second fills a page, NIP-01 cannot paginate it. + if oldest >= until || pages >= MAX_PAGES { + return Err(Error::Capacity); + } + until = oldest; + count = 0; + history(&mut connection, filter.clone(), journal.state.since, until).await?; + } else { + send(&mut connection, json!(["CLOSE", "mentions-history"])).await?; + tokio::time::timeout(IO_TIMEOUT, journal.checkpoint(connected_at)) + .await + .map_err(|_| Error::Deadline)??; + history_done = true; + } + progress = tokio::time::Instant::now(); + } + RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(Error::Relay), + _ => {} + } + } +} + +async fn history( + connection: &mut NostrWsConnection, + mut filter: serde_json::Value, + since: u64, + until: u64, +) -> Result<(), Error> { + filter["since"] = json!(since); + filter["until"] = json!(until); + filter["limit"] = json!(PAGE); + send(connection, json!(["REQ", "mentions-history", filter])).await +} + +fn eligible(identity: &Identity, journal: &Journal, event: &Event) -> bool { + if event.created_at.as_secs() < journal.state.activated + || serde_json::to_vec(event).map_or(true, |v| v.len() > MAX_EVENT_BYTES) + { + return false; + } + Start::admit(json!({"agent_pubkey": identity.keys.public_key(), "relay_url": identity.relay, "trigger": event}), + identity.owner, Timestamp::now().as_secs()).is_ok() +} + +/// The journal survives all cancellation points, including a successful deploy +/// whose result could not be saved. The provider converges on the same thread. +async fn drain( + manager: &Manager, + identity: &Identity, + journal: &mut Journal, + permits: &Semaphore, +) -> Result<(), Error> { + while journal.state.pending.is_some() { + let pending = journal.state.pending.as_ref().ok_or(Error::Storage)?; + if !eligible(identity, journal, &pending.event) { + return Err(Error::Registration); + } + if pending.notice.is_some() { + // Atomically retain the reply in its own outbox and finish this + // launch. Lost channel access must not block unrelated threads. + tokio::time::timeout(IO_TIMEOUT, journal.complete()) + .await + .map_err(|_| Error::Deadline)??; + continue; + } + let wait = pending + .retry_at + .saturating_sub(Timestamp::now().as_secs()) + .min(60); + tokio::time::sleep(Duration::from_secs(wait)).await; + let _permit = tokio::time::timeout(Duration::from_secs(60), permits.acquire()) + .await + .map_err(|_| Error::Deadline)? + .map_err(|_| Error::Deadline)?; + let pending = journal.state.pending.as_mut().ok_or(Error::Storage)?; + // Persist the attempt before any external effect; a crash consumes it. + if pending.attempts >= MAX_ATTEMPTS { + pending.notice = Some(failure_notice(identity, &pending.event, "retry_exhausted")?); + tokio::time::timeout(IO_TIMEOUT, journal.save()) + .await + .map_err(|_| Error::Deadline)??; + continue; + } + pending.attempts += 1; + let event = pending.event.clone(); + tokio::time::timeout(IO_TIMEOUT, journal.save()) + .await + .map_err(|_| Error::Deadline)??; + let result = match channel_launch_allowed(identity, &event).await { + Ok(LaunchPermission::Allowed) => tokio::time::timeout( + TURN_TIMEOUT, + manager.start_mention( + identity.owner, + identity.keys.public_key(), + &identity.relay, + &event, + ), + ) + .await + .unwrap_or_else(|_| json!({"ok": false, "code": "launch_timeout"})), + Ok(LaunchPermission::UnsupportedConversation) => { + json!({"ok": false, "code": "unsupported_conversation"}) + } + // Deletion or retyping revokes automatic starts. Let the current + // runtime handle the message without injecting a remote reply. + Ok(LaunchPermission::NoLongerRemote) => json!({"ok": true}), + Err(_) => json!({"ok": false, "code": "delivery_unconfirmed"}), + }; + if result["ok"] == true { + tokio::time::timeout(IO_TIMEOUT, journal.complete()) + .await + .map_err(|_| Error::Deadline)??; + } else { + let pending = journal.state.pending.as_mut().ok_or(Error::Storage)?; + let code = result["code"].as_str().unwrap_or("launch_failed"); + if matches!( + code, + "session_ended" + | "session_ending" + | "recovery_interrupted" + | "config_differs" + | "launch_not_registered" + | "unauthorized_owner" + | "invalid_start" + | "unsupported_conversation" + ) || pending.attempts >= MAX_ATTEMPTS + { + pending.notice = Some(failure_notice(identity, &event, code)?); + } else { + pending.retry_at = Timestamp::now().as_secs() + 5 * u64::from(pending.attempts); + } + tokio::time::timeout(IO_TIMEOUT, journal.save()) + .await + .map_err(|_| Error::Deadline)??; + } + } + flush_notices(identity, journal).await +} + +/// Bounded delivery attempts retain the exact signed event, including after a +/// lost acknowledgement. Exhausted replies remain available to operators. +async fn flush_notices(identity: &Identity, journal: &mut Journal) -> Result<(), Error> { + let ids: Vec<_> = journal + .state + .outbox + .iter() + .filter(|notice| notice.attempts < MAX_ATTEMPTS) + .take(4) + .map(|notice| notice.event.id) + .collect(); + for id in ids { + let notice = journal + .state + .outbox + .iter_mut() + .find(|notice| notice.event.id == id) + .ok_or(Error::Storage)?; + if notice.event.pubkey != identity.keys.public_key() || notice.event.verify().is_err() { + return Err(Error::Storage); + } + notice.attempts += 1; + let event = notice.event.clone(); + tokio::time::timeout(IO_TIMEOUT, journal.save()) + .await + .map_err(|_| Error::Deadline)??; + // No subscriptions: waiting for OK cannot buffer another inbox's traffic. + let published = tokio::time::timeout(IO_TIMEOUT, async { + let mut publisher = connect(identity).await?; + let ok = publisher + .send_event(event) + .await + .map_err(|_| Error::Relay)?; + if ok.accepted { + Ok(()) + } else { + Err(Error::Relay) + } + }) + .await + .unwrap_or(Err(Error::Deadline)); + match published { + Ok(()) => { + journal.state.outbox.retain(|notice| notice.event.id != id); + tokio::time::timeout(IO_TIMEOUT, journal.save()) + .await + .map_err(|_| Error::Deadline)??; + } + Err(error) => { + tracing::warn!(%error, event_id = %id, "startup reply retained in durable outbox") + } + } + } + Ok(()) +} + +fn failure_notice(identity: &Identity, event: &Event, code: &str) -> Result { + let start = Start::admit(json!({"agent_pubkey": identity.keys.public_key(), "relay_url": identity.relay, "trigger": event}), + identity.owner, Timestamp::now().as_secs()).map_err(|_| Error::Registration)?; + let text = match code { + "session_ended" | "session_ending" | "recovery_interrupted" => "This thread's agent session ended. Recover it from desktop, or mention me in a new thread.", + "config_differs" => "This thread has different agent settings. Continue its existing session or mention me in a new thread.", + "launch_not_registered" => "Start this agent once from desktop to register its launch settings, then mention me again.", + "unsupported_conversation" => "Automatic remote startup requires a channel thread. Mention me in a channel, or manage this conversation from desktop.", + _ => "Agent startup could not be confirmed. Mention me again to retry.", + }; + let (channel, root) = start.conversation(); + let tags = [ + vec!["h", channel], + vec!["e", root, "", "root"], + vec!["e", &event.id.to_hex(), "", "reply"], + ] + .into_iter() + .map(Tag::parse) + .collect::, _>>() + .map_err(|_| Error::Registration)?; + EventBuilder::new(Kind::Custom(9), text) + .tags(tags) + .sign_with_keys(&identity.keys) + .map_err(|_| Error::Registration) +} + +/// Recheck delivery under the agent's current relay permissions, including +/// restored intents. DMs use conversation-scoped ACP sessions and cannot run +/// in this provider's immutable thread binding. +async fn channel_launch_allowed( + identity: &Identity, + trigger: &Event, +) -> Result { + tokio::time::timeout(IO_TIMEOUT, async { + let start = Start::admit(json!({"agent_pubkey":identity.keys.public_key(), "relay_url":identity.relay, "trigger":trigger}), + identity.owner, Timestamp::now().as_secs()).map_err(|_| Error::Registration)?; + let (channel, _) = start.conversation(); + let mut connection = connect(identity).await?; + send(&mut connection, json!(["REQ", "mention-trigger", {"kinds":[9,40002],"ids":[trigger.id],"#h":[channel],"limit":1}])).await?; + send(&mut connection, json!(["REQ", "mention-channel", {"kinds":[39000],"#d":[channel],"limit":1}])).await?; + // Query the current coordinate without a runtime filter: an old remote + // record must not survive a newer local record or coordinate deletion. + let agent = identity.keys.public_key().to_hex(); + send(&mut connection, json!(["REQ", "mention-listing", {"kinds":[30177],"authors":[identity.owner],"#d":[agent],"limit":1}])).await?; + let (mut message_seen, mut metadata, mut trigger_done, mut channel_done, mut listing_done, mut remote) = (false, None, false, false, false, false); + for _ in 0..16 { + match connection.next_event(IO_TIMEOUT).await.map_err(|_| Error::Relay)? { + RelayMessage::Event { subscription_id, event } if subscription_id == "mention-trigger" => { + message_seen |= event.id == trigger.id && event.verify().is_ok(); + } + RelayMessage::Event { subscription_id, event } if subscription_id == "mention-channel" => { + if event.kind.as_u16() != 39000 || event.verify().is_err() + || !event.tags.iter().any(|tag| tag.as_slice() == ["d", channel]) { return Err(Error::Relay); } + metadata = Some(!event.tags.iter().any(|tag| { + let tag = tag.as_slice(); + tag.first().map(String::as_str) == Some("hidden") || tag == ["t", "dm"] || tag == ["archived", "true"] + })); + } + RelayMessage::Event { subscription_id, event } if subscription_id == "mention-listing" => { + if event.kind.as_u16() != 30177 || event.pubkey != identity.owner || event.verify().is_err() { + return Err(Error::Relay); + } + let coordinates: Vec<_> = event.tags.iter().filter(|tag| tag.as_slice().first().map(String::as_str) == Some("d")).collect(); + let runtimes: Vec<_> = event.tags.iter().filter(|tag| tag.as_slice().first().map(String::as_str) == Some("runtime")).collect(); + remote = coordinates.len() == 1 && coordinates[0].as_slice() == ["d", &agent] + && runtimes.len() == 1 && runtimes[0].as_slice() == ["runtime", "voltage.remote-workspace/sandbox"]; + } + RelayMessage::Eose { subscription_id } => { + trigger_done |= subscription_id == "mention-trigger"; + channel_done |= subscription_id == "mention-channel"; + listing_done |= subscription_id == "mention-listing"; + } + RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(Error::Relay), + _ => {}, + } + if trigger_done && channel_done && listing_done { + if !remote { return Ok(LaunchPermission::NoLongerRemote); } + return match (message_seen, metadata) { + (true, Some(true)) => Ok(LaunchPermission::Allowed), + (true, Some(false)) => Ok(LaunchPermission::UnsupportedConversation), + _ => Err(Error::Relay), + }; + } + } + Err(Error::Capacity) + }).await.map_err(|_| Error::Deadline)? +} diff --git a/crates/buzz-sandbox-reaper/src/mobile_launch.rs b/crates/buzz-sandbox-reaper/src/mobile_launch.rs index eeecf057ce5..f51ed979d83 100644 --- a/crates/buzz-sandbox-reaper/src/mobile_launch.rs +++ b/crates/buzz-sandbox-reaper/src/mobile_launch.rs @@ -13,8 +13,8 @@ use sha2::{Digest, Sha256}; use crate::manager::refusal; -const DATA_KEY: &str = "launch.json"; -const LABEL: &str = "buzz.voltage.cloud/mobile-launch"; +pub(crate) const DATA_KEY: &str = "launch.json"; +pub(crate) const LABEL: &str = "buzz.voltage.cloud/mobile-launch"; fn store_failed() -> Value { refusal( @@ -23,7 +23,7 @@ fn store_failed() -> Value { ) } -fn relay_endpoint(text: &str) -> Result { +pub(crate) fn relay_endpoint(text: &str) -> Result { let url = url::Url::parse(text).map_err(|_| invalid())?; if !matches!(url.scheme(), "wss" | "ws") || url.host_str().is_none() { return Err(invalid()); @@ -49,11 +49,11 @@ fn invalid() -> Value { #[derive(Deserialize, Serialize)] #[serde(deny_unknown_fields)] pub(crate) struct Registration { - version: u32, - owner: PublicKey, - agent: PublicKey, - relay: String, - payload: Value, + pub(crate) version: u32, + pub(crate) owner: PublicKey, + pub(crate) agent: PublicKey, + pub(crate) relay: String, + pub(crate) payload: Value, } impl Registration { @@ -77,7 +77,7 @@ impl Registration { }) } - fn name(owner: PublicKey, agent: PublicKey, relay: &str) -> String { + pub(crate) fn name(owner: PublicKey, agent: PublicKey, relay: &str) -> String { let hash = Sha256::digest(format!("{owner}\n{agent}\n{relay}").as_bytes()); let hex: String = hash[..20].iter().map(|b| format!("{b:02x}")).collect(); format!("buzz-mobile-launch-{hex}") @@ -210,6 +210,14 @@ impl Start { self.agent } + /// Verified channel and thread for an ordinary startup failure reply. + pub(crate) fn conversation(&self) -> (&str, &str) { + ( + self.scope["channel_id"].as_str().unwrap_or_default(), + self.scope["thread_root"].as_str().unwrap_or_default(), + ) + } + /// Read only the verified owner's namespace. Recheck all saved identity /// bindings and attestation before invoking the ordinary confined provider. pub async fn request( diff --git a/crates/buzz-sandbox-reaper/src/serve.rs b/crates/buzz-sandbox-reaper/src/serve.rs index e77a9180cbe..6736c928103 100644 --- a/crates/buzz-sandbox-reaper/src/serve.rs +++ b/crates/buzz-sandbox-reaper/src/serve.rs @@ -69,6 +69,10 @@ pub struct ServeArgs { /// `buzz-backend-relay` clients, which send none of them. #[arg(long, env = "BUZZ_SANDBOX_PROVIDER_CONFIG")] sandbox_provider_config: Option, + /// Start registered agents from ordinary owner mentions. Requires namespace + /// Secret list/get/create/update permissions; no mobile client changes. + #[arg(long, env = "BUZZ_SANDBOX_MENTION_LAUNCH", default_value_t = false)] + mention_launch: bool, /// GitHub REST API base for the enrollment login check. #[arg(long, env = "BUZZ_GITHUB_API_URL", default_value = enroll::DEFAULT_GITHUB_API_URL)] github_api_url: String, @@ -119,7 +123,14 @@ pub async fn serve(args: ServeArgs) -> Result<()> { if let Some(broker) = broker { let stop = CancellationToken::new(); let relay = async { - run(manager, &args.relay_url, auth_tag, stop.cancelled()).await; + run_services( + manager, + &args.relay_url, + auth_tag, + args.mention_launch, + stop.cancelled(), + ) + .await; stop.cancel(); }; let signal = async { @@ -131,11 +142,43 @@ pub async fn serve(args: ServeArgs) -> Result<()> { let (result, (), ()) = join!(broker.run(stop.clone()), relay, signal); result?; } else { - run(manager, &args.relay_url, auth_tag, shutdown()).await; + run_services( + manager, + &args.relay_url, + auth_tag, + args.mention_launch, + shutdown(), + ) + .await; } Ok(()) } +/// The mention listener shares the manager's lifetime, configuration and keys. +pub(crate) async fn run_services( + manager: Arc, + relay: &str, + auth: Option, + mention_launch: bool, + shutdown: impl Future, +) { + let stop = CancellationToken::new(); + let signal = async { + shutdown.await; + stop.cancel(); + }; + let mentions = async { + if mention_launch { + crate::mentions::run(Arc::clone(&manager), relay, stop.clone()).await; + } + }; + join!( + run(Arc::clone(&manager), relay, auth, stop.cancelled()), + mentions, + signal + ); +} + /// Never echoes the key or the file's content. fn load_keys(file: Option<&std::path::Path>) -> Result { let secret = match file { diff --git a/docs/mobile-remote-agent-launch.md b/docs/mobile-remote-agent-launch.md index 3e20e70daa7..323e805a694 100644 --- a/docs/mobile-remote-agent-launch.md +++ b/docs/mobile-remote-agent-launch.md @@ -1,5 +1,8 @@ # Start remote agents from mobile +For an unchanged App Store client, use [server mention launches](server-mention-launch.md). +The protocol below describes the optional modified-client path. + Mention an owned Remote workspace agent from mobile to start its session in that thread. The desktop can be closed after the agent has registered its launch settings with the manager. diff --git a/docs/server-mention-launch.md b/docs/server-mention-launch.md new file mode 100644 index 00000000000..24efcf28ef9 --- /dev/null +++ b/docs/server-mention-launch.md @@ -0,0 +1,138 @@ +# Start remote agents from an unchanged chat client + +Enable server mention launches to start registered Remote workspace agents from ordinary channel messages. +The phone sends its existing signed message. It does not send a launch request or receive agent credentials. + +```mermaid +sequenceDiagram + participant Phone as Existing chat client + participant Relay + participant Manager as Sandbox manager + participant Inbox as Kubernetes inbox Secret + participant Sandbox + Phone->>Relay: Signed channel message addressing an agent + Relay->>Manager: Deliver through agent-authenticated subscription + Manager->>Inbox: Save trigger and attempt under a lease + Manager->>Sandbox: Start or reuse this thread's session + Manager->>Inbox: Record completion + Sandbox->>Relay: Agent reply in the original thread + Relay->>Phone: Ordinary chat reply +``` + +## Contract + +Only configured owners can start their registered agents. +The manager verifies signatures, addressed public keys, channel tags, and thread tags. +It connects only to its configured relay, using each registered agent's key and owner attestation. +Relay membership controls which channel events the listener receives. +Before deployment, the manager confirms that the trigger remains readable and checks current channel metadata. +It also requires the owner's current kind 30177 listing to select `voltage.remote-workspace/sandbox`. +Deleting that listing or switching its runtime disables automatic starts, including pending work from an earlier connection. +Saved registrations alone do not override the current listing. +The owner private key stays on the client. + +The manager reuses the existing launch registration, provider confinement, and thread session identity. +Replies reuse that session. A new root message selects a separate session. +Ended sessions require desktop recovery or a new thread. +This feature supports channel messages of kinds 9 and 40002, with an addressed `p` tag. +Display-only `mention` tags, edits, DMs, and Solo session management are outside this launch contract. + +The feature is disabled by default. +Existing encrypted control requests and modified mobile clients remain supported. +Concurrent desktop and server launches converge through the provider's existing thread identity. + +## Durability and limits + +Each registration has a separate `buzz-mention-inbox-*` Secret. +It contains a replay cursor, recent completion receipts, and at most one pending signed trigger. +It also retains up to 32 undelivered startup replies separately from launch work. +Secret resource versions fence competing writers. A six-minute lease bounds takeover after a crashed manager. +Registration updates do not replace the inbox. +Registration removal stops its listener on the next successful discovery pass, within approximately 30 seconds. +Each launch also reloads the registration before invoking the provider. +Discovery failures suspend listeners until registry access recovers. + +The listener installs live delivery before fetching history. +History uses inclusive timestamp boundaries and pages of 1,000 events. +Only a complete history traversal advances the replay cursor. +Reconnects overlap that cursor by two minutes. +Messages signed before first activation or registration do not launch historical work. +During a disconnect, messages backdated before the replay cursor cannot be recovered through this timestamp-based subscription. +Connected live delivery can receive older timestamps after activation. +This limitation applies to unusually delayed offline sends; it does not remove their chat messages. + +The manager limits discovery to 64 registrations and concurrent launches to four per process. +Each launch attempt has a three-minute deadline and at most three persisted attempts. +Each inbox holds at most 4,096 recent receipts and 512 KiB of serialized state. +Triggers larger than 60,000 serialized bytes do not launch agents. +History stops after 16 pages or when one timestamp fills an entire page. +Capacity failures retain the cursor and pending work; they never skip a page to claim recovery. +Operators must resolve capacity failures before relying on further automatic starts. + +Startup failures produce ordinary replies signed by the agent. +The inbox saves the signed reply before publication and retains it until relay acceptance. +Lost acknowledgements retry the same event ID. +Each reply gets three delivery attempts; exhausted replies remain in the journal for operator investigation. +An undeliverable reply does not block launches in other channels. +A fresh owner mention creates a new attempt after a terminal failure. +The system never automatically recovers an explicitly ended session. + +## Rollout + +1. Grant the manager `get`, `list`, `create`, and `update` on Secrets in its configured developer namespaces. +2. Publish and promote the updated manager image through the deployment repository. +3. Enable `manager.mentionLaunch` in the CD chart after promoting the new manager image. + The chart sets `BUZZ_SANDBOX_MENTION_LAUNCH=true`; standalone managers can use `serve --mention-launch`. +4. Start each agent once from desktop after the registration-capable manager is deployed. +5. Use the same owner identity on the phone and desktop. +6. Add the agent to the test channel before mentioning it. +7. Complete the acceptance test below. + +The existing `buzz-mobile-launch-*` registrations retain launch credentials. +Inbox Secrets contain message content; restrict their access like other namespace Secrets. +Neither Secret belongs in logs, source control, or support transcripts. +This code change does not apply Terraform, change deployment settings, or publish an image. + +To roll back, disable `BUZZ_SANDBOX_MENTION_LAUNCH` and roll the manager deployment. +Keep registrations and inboxes so later re-enablement retains pending work and receipts. +Deleting an inbox discards its pending work and resets its history boundary. + +## Acceptance test + +Use the installed phone application without rebuilding it. + +1. Start a registered agent from desktop, then close desktop. +2. Wait for registry discovery, approximately 30 seconds. +3. From the phone, create a channel thread: `@Agent reply with phone launch confirmed`. +4. Confirm one Sandbox starts and the agent replies in that thread. +5. Mention it again inside the thread. Confirm the existing Sandbox is reused. +6. Mention it in another new thread. Confirm a separate Sandbox starts. +7. Restart the manager during startup. Confirm the original message remains singular and the launch eventually converges. +8. End a session from desktop. Mention it from the phone and confirm a recovery instruction appears. +9. Mention the agent from another owner's identity. Confirm no Sandbox starts for that message. + +Record the application build, community, thread IDs, and observed results. +Automated WebSocket tests exercise the manager with ordinary messages and an in-memory Kubernetes API. +The opt-in real-relay test also checks authentication, private-channel membership, message ingestion, and subscription delivery through Buzz itself. +Run it against an isolated local relay with `BUZZ_ALLOW_NIP_OA_AUTH=true`: + +```sh +BUZZ_MENTION_TEST_RELAY=ws://127.0.0.1:13039 cargo test --locked -p buzz-sandbox-reaper \ + unchanged_client_starts_agents_through_real_relay_permissions -- --ignored --nocapture +``` + +The test creates a private channel and synthetic identities, then checks new threads and same-thread replies. +It refuses non-loopback relay URLs. Kubernetes remains an in-memory API in this test. +These tests do not establish physical-phone, real-cluster, or model-runtime behavior. + +## Troubleshooting + +| Observation | Action | +|---|---| +| No launch registration | Start the agent once from desktop after upgrading the manager. | +| `registration_invalid` | Check the namespace, configured relay, saved identity, and owner attestation; register again from desktop. | +| `inbox_storage_failed` | Restore namespace Secret permissions or Kubernetes connectivity. Keep the inbox. | +| `inbox_lease_unavailable` | Check for another manager replica. After a crash, allow up to six minutes for takeover. | +| `mention_capacity_exceeded` | Inspect backlog size and registration count without printing Secret contents. Escalate before changing or deleting the journal. | +| `mention_relay_failed` | Check relay connectivity and the agent's channel membership. | +| Ended-session reply | Recover from desktop or begin a new channel thread. | diff --git a/voltage/remote-agents.md b/voltage/remote-agents.md index 995e2ea2fd7..56b5e9b61f8 100644 --- a/voltage/remote-agents.md +++ b/voltage/remote-agents.md @@ -10,6 +10,12 @@ Every commit an agent makes is signed with a key that is on your GitHub account, ## How it works +An unchanged phone app can trigger registered agents through [server mention launches](../docs/server-mention-launch.md). +This requires the manager feature to be enabled; no mobile rebuild is required. + +The diagrams and steps below describe the Voltage desktop app's explicit control-request path. +For the unchanged phone app, follow the server mention launch guide linked above. + ```mermaid flowchart LR subgraph laptop["Developer's machine"]