From 63bf1dcea4a0d69b54fd689d5b0d48bc22717686 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Mon, 26 Oct 2026 12:19:11 -0400 Subject: [PATCH] feat(mobile): launch remote agents from channel mentions Signed-off-by: Tee8z --- Cargo.lock | 1 + .../buzz-backend-kubernetes/src/provider.rs | 9 +- .../src/sandbox_fake.rs | 25 ++ crates/buzz-backend-kubernetes/src/wire.rs | 8 + crates/buzz-sandbox-reaper/Cargo.toml | 1 + crates/buzz-sandbox-reaper/src/main.rs | 1 + crates/buzz-sandbox-reaper/src/manager.rs | 83 ++++- .../buzz-sandbox-reaper/src/manager_tests.rs | 1 + .../buzz-sandbox-reaper/src/mobile_launch.rs | 284 +++++++++++++++ .../src/mobile_launch/tests.rs | 343 ++++++++++++++++++ .../buzz-sandbox-reaper/src/profile_tests.rs | 24 +- crates/buzz-sandbox-reaper/src/serve_tests.rs | 83 ++++- crates/buzz-session-control/src/lib.rs | 19 +- crates/buzz-session-control/src/tests.rs | 11 +- .../tests/fixtures/start.request.event.json | 1 + docs/mobile-remote-agent-launch.md | 98 +++++ docs/remote-agents.md | 26 +- mobile/lib/app.dart | 2 + .../channels/channel_detail_page.dart | 3 + .../channels/send_message_provider.dart | 40 +- .../features/channels/thread_detail_page.dart | 6 + mobile/lib/shared/agents/agent_launch.dart | 69 ++++ .../agents/mobile_agent_launch_queue.dart | 327 +++++++++++++++++ .../agents/remote_agent_launch_provider.dart | 134 +++++++ .../agents/remote_agent_launch_status.dart | 133 +++++++ .../agents/remote_agent_start_client.dart | 245 +++++++++++++ mobile/lib/shared/relay/nostr_models.dart | 6 + .../lib/shared/relay/signed_event_relay.dart | 4 +- .../test/features/agents/launch_fixture.dart | 223 ++++++++++++ .../agents/mobile_agent_launch_flow_test.dart | 312 ++++++++++++++++ .../mobile_agent_lookup_failure_test.dart | 72 ++++ .../remote_agent_launch_status_test.dart | 131 +++++++ .../remote_agent_start_client_test.dart | 127 +++++++ 33 files changed, 2814 insertions(+), 38 deletions(-) create mode 100644 crates/buzz-sandbox-reaper/src/mobile_launch.rs create mode 100644 crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs create mode 100644 crates/buzz-session-control/tests/fixtures/start.request.event.json create mode 100644 docs/mobile-remote-agent-launch.md create mode 100644 mobile/lib/shared/agents/agent_launch.dart create mode 100644 mobile/lib/shared/agents/mobile_agent_launch_queue.dart create mode 100644 mobile/lib/shared/agents/remote_agent_launch_provider.dart create mode 100644 mobile/lib/shared/agents/remote_agent_launch_status.dart create mode 100644 mobile/lib/shared/agents/remote_agent_start_client.dart create mode 100644 mobile/test/features/agents/launch_fixture.dart create mode 100644 mobile/test/features/agents/mobile_agent_launch_flow_test.dart create mode 100644 mobile/test/features/agents/mobile_agent_lookup_failure_test.dart create mode 100644 mobile/test/features/agents/remote_agent_launch_status_test.dart create mode 100644 mobile/test/features/agents/remote_agent_start_client_test.dart diff --git a/Cargo.lock b/Cargo.lock index 3d25d7086b6..13507ebf0ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2033,6 +2033,7 @@ dependencies = [ "tower 0.5.3", "tracing", "tracing-subscriber", + "url", "zeroize", ] diff --git a/crates/buzz-backend-kubernetes/src/provider.rs b/crates/buzz-backend-kubernetes/src/provider.rs index e485312d111..12f50ab26bd 100644 --- a/crates/buzz-backend-kubernetes/src/provider.rs +++ b/crates/buzz-backend-kubernetes/src/provider.rs @@ -62,12 +62,17 @@ pub fn check_managed_defaults( /// A provider that forwards the request (`buzz-backend-relay`) runs them /// first, so a request that would be refused never leaves the machine. pub fn check_sandbox_request(request: &wire::DeployRequest) -> Result<(), wire::ProviderError> { - sandbox::Options::parse(&request.provider_config)? - .ok_or("the session manager starts Sandbox sessions only")?; + check_sandbox_scope(&request.provider_config)?; identity::check(request, true)?; Ok(()) } +/// Validate a concrete channel and thread scope before accepting a managed start. +pub fn check_sandbox_scope(config: &serde_json::Value) -> Result<(), String> { + sandbox::Options::parse(config)?.ok_or("the session manager starts Sandbox sessions only")?; + Ok(()) +} + impl Target { /// Overwrite what the managed target decides. A no-op for kubeconfig. fn confine(&self, request: &mut wire::DeployRequest) -> Result<(), wire::ProviderError> { diff --git a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs index 042d53ed885..c268a8c6cff 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs @@ -185,6 +185,31 @@ impl State { self.secrets.remove(*name); ok(&json!({"apiVersion":"v1","kind":"Status","status":"Success"})) } + ["api", "v1", "namespaces", _, "secrets", name] if method == Method::PUT => { + let Some(prior) = self.secrets.get(*name) else { + return status(404, "NotFound"); + }; + let mut secret: Secret = serde_json::from_slice(body).unwrap(); + if prior["immutable"] == true + || secret.metadata.resource_version.as_deref() + != prior["metadata"]["resourceVersion"].as_str() + { + return status(409, "Conflict"); + } + secret.metadata.uid = prior["metadata"]["uid"].as_str().map(str::to_owned); + secret.metadata.resource_version = Some(self.next_rv()); + secret.data.get_or_insert_with(Default::default).extend( + secret + .string_data + .take() + .unwrap_or_default() + .into_iter() + .map(|(key, value)| (key, ByteString(value.into_bytes()))), + ); + let stored = serde_json::to_value(secret).unwrap(); + self.secrets.insert((*name).into(), stored.clone()); + ok(&stored) + } ["api", "v1", "namespaces", _, "secrets", name] => match self.secrets.get(*name) { Some(secret) => ok(secret), None => status(404, "NotFound"), diff --git a/crates/buzz-backend-kubernetes/src/wire.rs b/crates/buzz-backend-kubernetes/src/wire.rs index eede47fd7e6..3a9a084d9de 100644 --- a/crates/buzz-backend-kubernetes/src/wire.rs +++ b/crates/buzz-backend-kubernetes/src/wire.rs @@ -159,6 +159,8 @@ pub struct ErrorResponse { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ErrorCode { + /// A desktop launch cannot register its relay endpoint for mobile starts. + InvalidLaunchRegistration, /// The session is a tombstone; only an explicit recovery starts another. SessionEnded, /// A stop is draining; the session ends once its Pod is gone. @@ -242,6 +244,12 @@ pub enum ErrorCode { GithubLoginMismatch, /// The profile could not be written to the developer's secret. ProfileStoreFailed, + /// No saved launch exists for this owner, community and agent. + LaunchNotRegistered, + /// A saved launch could not be read or durably updated. + LaunchStoreFailed, + /// An owner-start payload is not a signed mention of the requested agent. + InvalidStart, /// Client-side: the manager did not accept the request within 30 s. ManagerUnavailable, /// Client-side: the manager accepted the request but did not finish it diff --git a/crates/buzz-sandbox-reaper/Cargo.toml b/crates/buzz-sandbox-reaper/Cargo.toml index 533092c2f9a..4d2b4ad5e9c 100644 --- a/crates/buzz-sandbox-reaper/Cargo.toml +++ b/crates/buzz-sandbox-reaper/Cargo.toml @@ -44,6 +44,7 @@ aws-sdk-secretsmanager = { version = "1", default-features = false, features = [ reqwest = { workspace = true } toml = "1.0" zeroize = { workspace = true } +url = { workspace = true } # Same reason as buzz-backend-kubernetes: the process-level CryptoProvider is # installed explicitly because the workspace unifies ring and aws-lc-rs. rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index 4833cffa34a..6d7823e24cc 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/crates/buzz-sandbox-reaper/src/main.rs @@ -13,6 +13,7 @@ mod config; mod enroll; mod evidence; mod manager; +mod mobile_launch; mod plan; mod profile; mod reconcile; diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index 5bb232017b8..112f8be9408 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/crates/buzz-sandbox-reaper/src/manager.rs @@ -19,6 +19,7 @@ use sha2::{Digest, Sha256}; use crate::config::{is_dns_label, is_owner, CONFIG_VERSION}; use crate::enroll::{self, Enrollment}; +use crate::mobile_launch::{Registration, Start}; /// How long received event ids and finished responses are kept. pub const REPLAY_WINDOW_SECS: u64 = 600; @@ -253,7 +254,13 @@ pub struct Job { /// What a job runs. pub enum Work { /// A provider protocol request, against the owner's namespace. - Provider { request: Request, target: Target }, + Provider { + request: Request, + target: Target, + registration: Option, + }, + /// An owner-authorized start using a registered launch identity. + Start { start: Start, target: Target }, /// Steps 5 and 6 of enrollment: the merge, the GitHub login check and /// the write. Enroll { @@ -268,8 +275,41 @@ impl Work { /// Run to the `done` response. pub async fn run(self, manager: &Manager) -> serde_json::Value { match self { - Work::Provider { request, target } => { - response_value(&provider::run(request, &target).await) + Work::Provider { + request, + target, + registration, + } => { + let response = response_value(&provider::run(request, &target).await); + if response["ok"] == true { + if let ( + Some(registration), + Target::Managed { + client, namespace, .. + }, + ) = (registration, &target) + { + if let Err(error) = registration.save(client.clone(), namespace).await { + return error; + } + } + } + response + } + Work::Start { start, target } => { + let Target::Managed { + client, namespace, .. + } = &target + else { + return refusal( + ErrorCode::InvalidStart, + "mobile starts require a managed target", + ); + }; + match start.request(client.clone(), namespace).await { + Ok(request) => response_value(&provider::run(request, &target).await), + Err(error) => error, + } } Work::Enroll { developer, profile } => { manager.enrollment.run(&developer, profile).await @@ -444,11 +484,42 @@ impl Manager { Work::Profile { developer: developer.clone(), } + } else if op == Op::Start { + let start = match Start::admit( + opened.request.payload.clone(), + owner, + opened.created_at.as_secs(), + ) { + Ok(start) => start, + Err(response) => return answer(response), + }; + let target = Target::Managed { + client: self.client.clone(), + namespace: developer.namespace.clone(), + agent: start.agent(), + owner, + defaults: self.defaults.clone(), + env: self.config.launch_env(developer), + }; + Work::Start { start, target } } else { + let registration = if op == Op::Deploy { + match Registration::new( + owner, + opened.signer, + opened.request.payload["agent"].clone(), + ) { + Ok(registration) => Some(registration), + Err(response) => return answer(response), + } + } else { + None + }; match self.payload(&opened) { // 6. The namespace comes from the verified owner, never the request. Ok(request) => Work::Provider { request, + registration, target: Target::Managed { client: self.client.clone(), namespace: developer.namespace.clone(), @@ -563,7 +634,7 @@ impl Manager { return Admission::Ignore { event: event.id, error, - } + }; } }; Admission::Done { @@ -579,12 +650,12 @@ impl Manager { /// Steps 2 and 7: the verified owner. fn owner(&self, opened: &OpenedRequest) -> Result { - if matches!(opened.request.op, Op::Enroll | Op::Profile) { + if matches!(opened.request.op, Op::Enroll | Op::Profile | Op::Start) { // The profile is owner-level: an agent key's attestation is not enough. if opened.auth_tag.is_some() { return Err(( ErrorCode::SignerMismatch, - "enroll and profile must be signed by the owner key itself", + "enroll, profile and start must be signed by the owner key itself", )); } return Ok(opened.signer); diff --git a/crates/buzz-sandbox-reaper/src/manager_tests.rs b/crates/buzz-sandbox-reaper/src/manager_tests.rs index fc810d7bb7d..de5d2bc883d 100644 --- a/crates/buzz-sandbox-reaper/src/manager_tests.rs +++ b/crates/buzz-sandbox-reaper/src/manager_tests.rs @@ -154,6 +154,7 @@ async fn an_attested_fresh_deploy_starts_in_the_owners_namespace() { env, .. }, + .. } = job.work else { panic!("not a managed deploy"); diff --git a/crates/buzz-sandbox-reaper/src/mobile_launch.rs b/crates/buzz-sandbox-reaper/src/mobile_launch.rs new file mode 100644 index 00000000000..eeecf057ce5 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mobile_launch.rs @@ -0,0 +1,284 @@ +//! Owner-authorized mobile starts. Launch credentials stay in a namespaced +//! Secret on the manager's substrate, never on the phone or in relay history. + +use std::collections::BTreeMap; + +use buzz_backend_kubernetes::{provider, wire}; +use k8s_openapi::api::core::v1::Secret; +use kube::{api::PostParams, Api}; +use nostr::{Event, Keys, PublicKey}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use crate::manager::refusal; + +const DATA_KEY: &str = "launch.json"; +const LABEL: &str = "buzz.voltage.cloud/mobile-launch"; + +fn store_failed() -> Value { + refusal( + wire::ErrorCode::LaunchStoreFailed, + "the saved mobile launch could not be read or updated; retry", + ) +} + +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()); + } + // Desktop supports relay paths and queries. Preserve the full endpoint + // in the community binding; distinct paths on one host must not collide. + if url.path() == "/" && url.query().is_none() && url.fragment().is_none() { + Ok(url.as_str().trim_end_matches('/').to_owned()) + } else { + Ok(url.to_string()) + } +} + +fn invalid() -> Value { + refusal( + wire::ErrorCode::InvalidStart, + "start requires the owner's signed channel message addressing this agent", + ) +} + +/// A launch snapshot registered by a successful, agent-signed deploy. +/// Deliberately has no Debug implementation: the payload contains credentials. +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Registration { + version: u32, + owner: PublicKey, + agent: PublicKey, + relay: String, + payload: Value, +} + +impl Registration { + /// Bind a desktop launch snapshot to its verified owner, agent and relay. + pub fn new(owner: PublicKey, agent: PublicKey, payload: Value) -> Result { + let relay = payload["relay_url"] + .as_str() + .and_then(|text| relay_endpoint(text).ok()) + .ok_or_else(|| { + refusal( + wire::ErrorCode::InvalidLaunchRegistration, + "desktop launch registration requires a valid ws:// or wss:// relay URL", + ) + })?; + Ok(Self { + version: 1, + owner, + agent, + relay, + payload, + }) + } + + 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}") + } + + /// One atomic create or resourceVersion-checked replace. A conflict is + /// bounded; a failed save is returned instead of claiming full success. + pub async fn save(&self, client: kube::Client, namespace: &str) -> Result<(), Value> { + let api: Api = Api::namespaced(client, namespace); + let name = Self::name(self.owner, self.agent, &self.relay); + let text = serde_json::to_string(self).map_err(|_| store_failed())?; + for _ in 0..3 { + let prior = api.get_opt(&name).await.map_err(|_| store_failed())?; + if let Some(prior) = &prior { + if prior + .metadata + .labels + .as_ref() + .and_then(|l| l.get(LABEL)) + .map(String::as_str) + != Some("v1") + { + return Err(store_failed()); + } + } + let mut secret = Secret { + string_data: Some(BTreeMap::from([(DATA_KEY.into(), text.clone())])), + type_: Some("Opaque".into()), + ..Default::default() + }; + secret.metadata.name = Some(name.clone()); + secret.metadata.namespace = Some(namespace.into()); + secret.metadata.labels = Some(BTreeMap::from([(LABEL.into(), "v1".into())])); + secret.metadata.resource_version = prior + .as_ref() + .and_then(|s| s.metadata.resource_version.clone()); + let saved = if prior.is_some() { + api.replace(&name, &PostParams::default(), &secret).await + } else { + api.create(&PostParams::default(), &secret).await + }; + match saved { + Ok(_) => return Ok(()), + Err(kube::Error::Api(error)) if error.code == 409 => {} + Err(_) => return Err(store_failed()), + } + } + Err(store_failed()) + } +} + +/// A verified owner mention, with its immutable channel and thread binding. +pub(crate) struct Start { + owner: PublicKey, + agent: PublicKey, + relay: String, + scope: Value, + replay_floor: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct StartPayload { + agent_pubkey: PublicKey, + relay_url: String, + trigger: Event, +} + +impl Start { + /// Verify the owner's signed mention and derive its immutable thread scope. + pub fn admit(payload: Value, owner: PublicKey, request_time: u64) -> Result { + let payload: StartPayload = serde_json::from_value(payload).map_err(|_| invalid())?; + let event = payload.trigger; + let tags: Vec<&[String]> = event.tags.iter().map(|tag| tag.as_slice()).collect(); + let channels: Vec<_> = tags + .iter() + .filter(|t| t.first().map(String::as_str) == Some("h")) + .collect(); + let addressed = tags.iter().any(|t| { + t.first().map(String::as_str) == Some("p") + && t.get(1).map(String::as_str) == Some(payload.agent_pubkey.to_hex().as_str()) + }); + if event.verify().is_err() + || event.pubkey != owner + || payload.agent_pubkey == owner + || !matches!(event.kind.as_u16(), 9 | 40002) + || event.created_at.as_secs() > request_time.saturating_add(60) + || !addressed + || channels.len() != 1 + || channels[0].len() != 2 + { + return Err(invalid()); + } + let marked = |marker: &str| -> Vec<&str> { + tags.iter() + .filter(|t| { + t.first().map(String::as_str) == Some("e") + && t.get(3).map(String::as_str) == Some(marker) + }) + .filter_map(|t| t.get(1).map(String::as_str)) + .collect() + }; + let roots = marked("root"); + let replies = marked("reply"); + let unmarked = tags.iter().any(|tag| { + tag.first().map(String::as_str) == Some("e") + && !matches!( + tag.get(3).map(String::as_str), + Some("root" | "reply" | "mention") + ) + }); + if roots.len() > 1 || replies.len() > 1 || unmarked { + return Err(invalid()); + } + let id = event.id.to_hex(); + let root = roots.first().or(replies.first()).copied().unwrap_or(&id); + let scope = json!({"channel_id": channels[0][1], "thread_root": root}); + provider::check_sandbox_scope(&json!({"sandbox": scope})).map_err(|_| invalid())?; + Ok(Self { + owner, + agent: payload.agent_pubkey, + relay: relay_endpoint(&payload.relay_url)?, + scope, + replay_floor: event.created_at.as_secs(), + }) + } + + /// The addressed agent, used to confine the managed provider target. + pub fn agent(&self) -> PublicKey { + self.agent + } + + /// Read only the verified owner's namespace. Recheck all saved identity + /// bindings and attestation before invoking the ordinary confined provider. + pub async fn request( + self, + client: kube::Client, + namespace: &str, + ) -> Result { + let api: Api = Api::namespaced(client, namespace); + let name = Registration::name(self.owner, self.agent, &self.relay); + let secret = api.get_opt(&name).await.map_err(|_| store_failed())?.ok_or_else(|| { + refusal(wire::ErrorCode::LaunchNotRegistered, + "start this agent once from desktop after the manager upgrade to register mobile launches") + })?; + let data = secret + .data + .as_ref() + .and_then(|d| d.get(DATA_KEY)) + .ok_or_else(store_failed)?; + if secret + .metadata + .labels + .as_ref() + .and_then(|labels| labels.get(LABEL)) + .map(String::as_str) + != Some("v1") + { + return Err(store_failed()); + } + let saved: Registration = serde_json::from_slice(&data.0).map_err(|_| store_failed())?; + let key = saved.payload["private_key_nsec"] + .as_str() + .ok_or_else(store_failed)?; + if saved.version != 1 + || saved.owner != self.owner + || saved.agent != self.agent + || saved.relay != self.relay + || Keys::parse(key).map(|keys| keys.public_key()).ok() != Some(self.agent) + || relay_endpoint( + saved.payload["relay_url"] + .as_str() + .ok_or_else(store_failed)?, + )? != self.relay + { + return Err(store_failed()); + } + let mut payload = json!({"op": "deploy", "agent": saved.payload, + "provider_config": {"sandbox": self.scope, + "identity_policy": {"agent_pubkey": self.agent, "owner_pubkey": self.owner}}}); + // Replay is invocation state. Never retain an earlier thread's floor. + for tier in ["env", "policy_env"] { + if let Some(env) = payload + .get_mut("agent") + .and_then(|a| a.get_mut("launch")) + .and_then(|launch| launch.get_mut(tier)) + .and_then(Value::as_object_mut) + { + env.retain(|key, _| !key.eq_ignore_ascii_case("BUZZ_ACP_REPLAY_FLOOR")); + } + } + payload["agent"]["launch"]["env"]["BUZZ_ACP_REPLAY_FLOOR"] = + self.replay_floor.to_string().into(); + let request = provider::parse(payload).map_err(|_| store_failed())?; + let wire::Request::Deploy(deploy) = &request else { + return Err(store_failed()); + }; + provider::check_sandbox_request(deploy).map_err(|_| store_failed())?; + Ok(request) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs b/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs new file mode 100644 index 00000000000..a44b7066416 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs @@ -0,0 +1,343 @@ +use super::*; +use buzz_backend_kubernetes::sandbox_fake; +use nostr::{nips::nip19::ToBech32, EventBuilder, Kind, Tag, Timestamp}; +use std::sync::{Arc, Mutex}; + +fn owner() -> Keys { + Keys::parse(&"33".repeat(32)).unwrap() +} +fn agent() -> Keys { + Keys::parse(&"22".repeat(32)).unwrap() +} +const RELAY: &str = "wss://relay.example.invalid"; +const CHANNEL: &str = "0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10"; +const NAMESPACE: &str = "buzz-agent-staging-dev"; + +fn launch() -> Value { + json!({"relay_url": RELAY, "private_key_nsec": agent().secret_key().to_bech32().unwrap(), + "auth_tag": buzz_sdk::nip_oa::compute_auth_tag(&owner(), &agent().public_key(), "").unwrap(), + "launch": {"command": "codex-acp", "args": [], "env": {"BUZZ_ACP_REPLAY_FLOOR":"1"}, + "policy_env": {"BUZZ_ACP_SESSION_POLICY":"thread"}, "owner_pubkey": owner().public_key()}}) +} + +fn mention(tags: Vec) -> Event { + EventBuilder::new(Kind::Custom(9), "@agent investigate this") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(agent().public_key()), + ]) + .tags(tags) + .custom_created_at(Timestamp::from(1_800_000_000)) + .sign_with_keys(&owner()) + .unwrap() +} + +fn start(event: &Event) -> Start { + Start::admit( + json!({"agent_pubkey":agent().public_key(),"relay_url":RELAY,"trigger":event}), + owner().public_key(), + event.created_at.as_secs(), + ) + .unwrap() +} + +#[test] +fn binds_the_exact_signed_root_direct_reply_and_nested_reply() { + let root = mention(vec![]); + assert_eq!(start(&root).scope["thread_root"], root.id.to_hex()); + let root_id = "a".repeat(64); + let direct = mention(vec![Tag::parse(["e", &root_id, "", "reply"]).unwrap()]); + assert_eq!(start(&direct).scope["thread_root"], root_id); + let nested = mention(vec![ + Tag::parse(["e", &root_id, "", "root"]).unwrap(), + Tag::parse(["e", &"b".repeat(64), "", "reply"]).unwrap(), + ]); + assert_eq!(start(&nested).scope["thread_root"], root_id); + assert_eq!(start(&nested).scope["channel_id"], CHANNEL); +} + +#[test] +fn rejects_forged_wrong_owner_unaddressed_ambiguous_and_future_triggers() { + let event = mention(vec![]); + let value = json!({"agent_pubkey":agent().public_key(),"relay_url":RELAY,"trigger":event}); + let mut cases = vec![]; + let mut wrong_owner = value.clone(); + wrong_owner["trigger"] = json!(EventBuilder::new(Kind::Custom(9), "forged") + .sign_with_keys(&agent()) + .unwrap()); + cases.push(wrong_owner); + let mut tampered = value.clone(); + tampered["trigger"]["content"] = "tampered".into(); + cases.push(tampered); + let mut wrong_agent = value.clone(); + wrong_agent["agent_pubkey"] = json!(Keys::generate().public_key()); + cases.push(wrong_agent); + let mut extra = value.clone(); + extra["namespace"] = "victim".into(); + cases.push(extra); + let duplicate = mention(vec![Tag::parse(["h", CHANNEL]).unwrap()]); + cases.push(json!({"agent_pubkey":agent().public_key(),"relay_url":RELAY,"trigger":duplicate})); + for value in cases { + assert!(Start::admit(value, owner().public_key(), event.created_at.as_secs()).is_err()); + } + assert!(Start::admit(value, owner().public_key(), event.created_at.as_secs() - 61).is_err()); +} + +#[tokio::test] +async fn durable_registration_survives_manager_restart_and_rebinds_each_thread() { + let state: sandbox_fake::Shared = Arc::new(Mutex::new(Default::default())); + let registration = + Registration::new(owner().public_key(), agent().public_key(), launch()).unwrap(); + registration + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + let event = mention(vec![]); + // A fresh client and Start know nothing about the desktop or prior process. + let wire::Request::Deploy(request) = start(&event) + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap() + else { + panic!() + }; + assert_eq!( + request.agent.private_key_nsec, + agent().secret_key().to_bech32().unwrap() + ); + assert_eq!( + request.provider_config["sandbox"]["thread_root"], + event.id.to_hex() + ); + assert_eq!( + request.agent.launch.unwrap().env["BUZZ_ACP_REPLAY_FLOOR"], + event.created_at.to_string() + ); + let root = "b".repeat(64); + let reply = mention(vec![Tag::parse(["e", &root, "", "reply"]).unwrap()]); + let wire::Request::Deploy(request) = start(&reply) + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap() + else { + panic!() + }; + assert_eq!(request.provider_config["sandbox"]["thread_root"], root); + assert_eq!(state.lock().unwrap().secrets.len(), 1); +} + +#[tokio::test] +async fn missing_registration_and_tampered_saved_identity_fail_closed() { + let state: sandbox_fake::Shared = Arc::new(Mutex::new(Default::default())); + let event = mention(vec![]); + let error = start(&event) + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap_err(); + assert_eq!(error["code"], "launch_not_registered"); + let mut registration = + Registration::new(owner().public_key(), agent().public_key(), launch()).unwrap(); + registration.payload["private_key_nsec"] = owner().secret_key().to_bech32().unwrap().into(); + registration + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + let error = start(&event) + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap_err(); + assert_eq!(error["code"], "launch_store_failed"); + assert!(!error.to_string().contains("nsec")); + assert!(state.lock().unwrap().sandboxes.is_empty()); +} + +#[tokio::test] +async fn successful_desktop_deploys_refresh_settings_atomically_and_report_save_failure() { + let state: sandbox_fake::Shared = Arc::new(Mutex::new(Default::default())); + let mut registration = + Registration::new(owner().public_key(), agent().public_key(), launch()).unwrap(); + registration + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + registration.payload["launch"]["env"]["BUZZ_ACP_MODEL"] = "new-model".into(); + registration + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + assert_eq!(state.lock().unwrap().secrets.len(), 1); + let wire::Request::Deploy(request) = start(&mention(vec![])) + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap() + else { + panic!() + }; + assert_eq!( + request.agent.launch.unwrap().env["BUZZ_ACP_MODEL"], + "new-model" + ); + let new_state: sandbox_fake::Shared = Arc::new(Mutex::new(sandbox_fake::State { + fail_secret_creates: 1, + ..Default::default() + })); + assert_eq!( + registration + .save(sandbox_fake::client(&new_state), NAMESPACE) + .await + .unwrap_err()["code"], + "launch_store_failed" + ); +} + +#[tokio::test] +async fn another_community_or_owner_cannot_read_the_registered_identity() { + let state: sandbox_fake::Shared = Arc::new(Mutex::new(Default::default())); + Registration::new(owner().public_key(), agent().public_key(), launch()) + .unwrap() + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + let mut other = start(&mention(vec![])); + other.relay = "wss://other.example.invalid".into(); + assert_eq!( + other + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap_err()["code"], + "launch_not_registered" + ); + let mut other = start(&mention(vec![])); + other.owner = Keys::generate().public_key(); + assert_eq!( + other + .request(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap_err()["code"], + "launch_not_registered" + ); + assert!(state.lock().unwrap().sandboxes.is_empty()); +} + +#[tokio::test] +async fn manager_admits_only_the_owner_key_and_confines_mobile_starts() { + use crate::manager::{Admission, Manager, ManagerConfig, Work}; + use buzz_session_control::{build_request, Op, Request as Envelope}; + let manager_keys = Keys::parse(&"11".repeat(32)).unwrap(); + let manager = Manager::new(manager_keys.clone(), ManagerConfig::parse(&json!({ + "version":1,"checkpoint_bucket":"fixture-checkpoints","developers":{"dev":{ + "owner_pubkey":owner().public_key(),"namespace":NAMESPACE, + "profile_secret_arn":"arn:aws:secretsmanager:us-west-2:000000000000:secret:voltage/staging/agents/dev/profile-AbCdEf"}}}).to_string()).unwrap(), + sandbox_fake::client(&Default::default()), crate::enroll::tests::offline(Default::default())); + let event = mention(vec![]); + let payload = json!({"agent_pubkey":agent().public_key(),"relay_url":RELAY,"trigger":event}); + let request = Envelope::new(Op::Start, "mobile-start", payload); + let sent = build_request( + &owner(), + &manager_keys.public_key(), + None, + &request, + event.created_at, + ) + .unwrap(); + let Admission::Start { job, .. } = manager.admit(&sent, event.created_at) else { + panic!() + }; + let Work::Start { + target: + provider::Target::Managed { + namespace, + owner: target_owner, + agent: target_agent, + .. + }, + .. + } = job.work + else { + panic!() + }; + assert_eq!(namespace, NAMESPACE); + assert_eq!(target_owner, owner().public_key()); + assert_eq!(target_agent, agent().public_key()); + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner(), &agent().public_key(), "").unwrap(); + let agent_signed = build_request( + &agent(), + &manager_keys.public_key(), + Some(&auth), + &request, + event.created_at, + ) + .unwrap(); + let Admission::Done { response, .. } = manager.admit(&agent_signed, event.created_at) else { + panic!() + }; + assert_eq!(response["code"], "signer_mismatch"); +} + +#[test] +fn opens_a_real_dart_signed_and_nip44_encrypted_start_fixture() { + let event: Event = serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../buzz-session-control/tests/fixtures/start.request.event.json" + ))) + .unwrap(); + let manager = Keys::parse(&"11".repeat(32)).unwrap(); + let opened = buzz_session_control::open_request(&event, &manager).unwrap(); + assert_eq!(opened.request.op, buzz_session_control::Op::Start); + assert_eq!(opened.signer, owner().public_key()); + assert!(opened.auth_tag.is_none()); + let start = Start::admit( + opened.request.payload, + opened.signer, + opened.created_at.as_secs(), + ) + .unwrap(); + assert_eq!(start.agent, agent().public_key()); + assert_eq!(start.replay_floor, 1_800_000_000); +} + +#[tokio::test] +async fn desktop_relay_paths_and_queries_register_without_merging_communities() { + let relay = "wss://relay.example.invalid/community-a?network=one"; + let state: sandbox_fake::Shared = Arc::new(Mutex::new(Default::default())); + let mut payload = launch(); + payload["relay_url"] = relay.into(); + Registration::new(owner().public_key(), agent().public_key(), payload) + .unwrap() + .save(sandbox_fake::client(&state), NAMESPACE) + .await + .unwrap(); + for endpoint in [ + relay, + "wss://relay.example.invalid/community-b?network=one", + "wss://relay.example.invalid/community-a?network=two", + ] { + let event = mention(vec![]); + let start = Start::admit( + json!({"agent_pubkey":agent().public_key(),"relay_url":endpoint,"trigger":event}), + owner().public_key(), + event.created_at.as_secs(), + ) + .unwrap(); + let result = start.request(sandbox_fake::client(&state), NAMESPACE).await; + if endpoint == relay { + let wire::Request::Deploy(request) = result.unwrap() else { + panic!() + }; + assert_eq!(request.agent.relay_url, relay); + } else { + assert_eq!(result.unwrap_err()["code"], "launch_not_registered"); + } + } +} + +#[test] +fn invalid_desktop_registration_has_a_deploy_specific_refusal_without_url_secrets() { + let mut payload = launch(); + payload["relay_url"] = "https://relay.example.invalid?token=private".into(); + let Err(error) = Registration::new(owner().public_key(), agent().public_key(), payload) else { + panic!() + }; + assert_eq!(error["code"], "invalid_launch_registration"); + assert!(!error.to_string().contains("private")); +} diff --git a/crates/buzz-sandbox-reaper/src/profile_tests.rs b/crates/buzz-sandbox-reaper/src/profile_tests.rs index c003d29f446..e19e1607d09 100644 --- a/crates/buzz-sandbox-reaper/src/profile_tests.rs +++ b/crates/buzz-sandbox-reaper/src/profile_tests.rs @@ -232,10 +232,26 @@ fn a_partial_profile_may_omit_fields_but_not_add_them() { partial["git"] = json!({"email": "ada@example.com"}); assert_eq!(check_partial(&partial), Ok(())); for (pointer, bad, message) in [ - ("/models", json!({"codex_access_token": 1}), "Unsupported model credential type"), - ("/models", json!({"gemini_api_key": "x"}), "Unsupported model credential type"), - ("/git", json!({"signing": "x"}), "Profile fields are incomplete or unsupported"), - ("/github", json!({"token": "ghp_x"}), "A GitHub token needs its login"), + ( + "/models", + json!({"codex_access_token": 1}), + "Unsupported model credential type", + ), + ( + "/models", + json!({"gemini_api_key": "x"}), + "Unsupported model credential type", + ), + ( + "/git", + json!({"signing": "x"}), + "Profile fields are incomplete or unsupported", + ), + ( + "/github", + json!({"token": "ghp_x"}), + "A GitHub token needs its login", + ), ("/version", json!(2), "Unsupported profile version"), ] { let mut partial = base(); diff --git a/crates/buzz-sandbox-reaper/src/serve_tests.rs b/crates/buzz-sandbox-reaper/src/serve_tests.rs index 28882cfcbd0..3388794b42c 100644 --- a/crates/buzz-sandbox-reaper/src/serve_tests.rs +++ b/crates/buzz-sandbox-reaper/src/serve_tests.rs @@ -225,12 +225,23 @@ async fn deploy_and_stop_run_end_to_end_over_the_relay() { ) .unwrap(); let cluster = sandbox_fake::Shared::default(); - let manager = Arc::new(Manager::new( - manager_keys.clone(), - config, - sandbox_fake::client(&cluster), - crate::enroll::tests::offline(Default::default()), - )); + let manager = Arc::new( + Manager::new( + manager_keys.clone(), + config, + sandbox_fake::client(&cluster), + crate::enroll::tests::offline(Default::default()), + ) + .with_sandbox_defaults( + json!({ + "image": format!("example.invalid/sprig@sha256:{}", "a".repeat(64)), + "pod_options": {"active_deadline_seconds": 3600} + }) + .as_object() + .unwrap() + .clone(), + ), + ); let mut relay = FakeRelay::start().await; let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); let url = relay.url.clone(); @@ -244,6 +255,28 @@ async fn deploy_and_stop_run_end_to_end_over_the_relay() { let mut agent = Agent::connect(&mut relay, agent_keys.clone(), manager_keys.public_key()).await; let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent_keys.public_key(), "").unwrap(); + let trigger = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@agent investigate") + .tags([ + Tag::parse(["h", "0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10"]).unwrap(), + Tag::public_key(agent_keys.public_key()), + ]) + .sign_with_keys(&owner) + .unwrap(); + let before_setup = Envelope::new( + Op::Start, + "phone-before-setup", + json!({"agent_pubkey":agent_keys.public_key(), + "relay_url":"wss://relay.example.invalid", "trigger":trigger}), + ); + let mut setup_phone = + Agent::connect(&mut relay, owner.clone(), manager_keys.public_key()).await; + let refused = setup_phone.request(&before_setup, None).await; + assert_eq!( + refused.last().unwrap().response.as_ref().unwrap()["code"], + "launch_not_registered" + ); + drop(setup_phone); + let deploy = Envelope::new( Op::Deploy, "deploy-1", @@ -265,6 +298,40 @@ async fn deploy_and_stop_run_end_to_end_over_the_relay() { } } + // The owner can start a new thread after the desktop/agent connection + // closes: only the saved registration and owner's key are needed. + drop(agent); + let mut phone = Agent::connect(&mut relay, owner.clone(), manager_keys.public_key()).await; + // Completed failures stay cached even after desktop setup fixes their cause. + // Explicit Retry needs a new operation ID; the signed trigger stays intact. + let cached = phone.request(&before_setup, None).await; + assert_eq!( + cached.last().unwrap().response.as_ref().unwrap()["code"], + "launch_not_registered" + ); + let start = Envelope::new(Op::Start, "phone-1", before_setup.payload.clone()); + let mobile = phone.request(&start, None).await; + let mobile_response = mobile.last().unwrap().response.as_ref().unwrap(); + assert_eq!(mobile_response["ok"], true, "{mobile_response}"); + assert_ne!(mobile_response["agent_id"], agent_id); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 2); + let duplicate = phone.request(&start, None).await; + assert_eq!( + duplicate.last().unwrap().response.as_ref().unwrap(), + mobile_response + ); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 2); + // Even a new operation ID converges to the existing thread Sandbox; + // correctness does not depend only on the manager's short-lived cache. + let retry = Envelope::new(Op::Start, "phone-2", start.payload.clone()); + let retried = phone.request(&retry, None).await; + assert_eq!( + retried.last().unwrap().response.as_ref().unwrap(), + mobile_response + ); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 2); + + let mut agent = Agent::connect(&mut relay, agent_keys.clone(), manager_keys.public_key()).await; let stop_request = Envelope::new( Op::Stop, "stop-1", @@ -302,11 +369,13 @@ async fn deploy_and_stop_run_end_to_end_over_the_relay() { ); let sent = relay.published_by(&manager_keys.public_key()); - assert_eq!(sent.len(), 5); + assert_eq!(sent.len(), 13); for event in sent { let json = serde_json::to_string(&event).unwrap(); for needle in [ "unauthorized_owner", + "launch_not_registered", + "phone-before-setup", "ending", "agent_id", "deploy-1", diff --git a/crates/buzz-session-control/src/lib.rs b/crates/buzz-session-control/src/lib.rs index ceb7650f73a..f7c6a94ee9b 100644 --- a/crates/buzz-session-control/src/lib.rs +++ b/crates/buzz-session-control/src/lib.rs @@ -54,6 +54,8 @@ pub enum Op { /// Describe the developer's stored profile, naming no value. Signed by /// the owner key itself. Profile, + /// Start an already registered agent for an owner-signed channel mention. + Start, } impl Op { @@ -65,13 +67,21 @@ impl Op { Op::Stop => "stop", Op::Enroll => "enroll", Op::Profile => "profile", + Op::Start => "start", } } fn parse(value: &str) -> Option { - [Op::Info, Op::Deploy, Op::Stop, Op::Enroll, Op::Profile] - .into_iter() - .find(|op| op.as_str() == value) + [ + Op::Info, + Op::Deploy, + Op::Stop, + Op::Enroll, + Op::Profile, + Op::Start, + ] + .into_iter() + .find(|op| op.as_str() == value) } } @@ -306,7 +316,8 @@ pub fn check_created_at(created_at: Timestamp, now: Timestamp) -> Result<(), Wir /// Encrypt `request` to `manager` and sign it with `signer`. /// /// `auth_tag` is the owner's NIP-OA attestation JSON for the signing agent -/// key (`BUZZ_AUTH_TAG`); `enroll` is signed by the owner and passes `None`. +/// key (`BUZZ_AUTH_TAG`); `enroll`, `profile`, and `start` are signed by the +/// owner and pass `None`. pub fn build_request( signer: &Keys, manager: &PublicKey, diff --git a/crates/buzz-session-control/src/tests.rs b/crates/buzz-session-control/src/tests.rs index 734f96c903a..2d1f8f3dbcf 100644 --- a/crates/buzz-session-control/src/tests.rs +++ b/crates/buzz-session-control/src/tests.rs @@ -170,9 +170,16 @@ fn golden_events_open_to_their_envelopes() { #[test] fn envelopes_round_trip_for_every_op_and_status() { - for op in [Op::Info, Op::Deploy, Op::Stop, Op::Enroll, Op::Profile] { + for op in [ + Op::Info, + Op::Deploy, + Op::Stop, + Op::Enroll, + Op::Profile, + Op::Start, + ] { let request = Request::new(op, "r-1", json!({"op": op.as_str()})); - let owner_signed = matches!(op, Op::Enroll | Op::Profile); + let owner_signed = matches!(op, Op::Enroll | Op::Profile | Op::Start); let auth = (!owner_signed).then(auth_tag); let event = build_request( &agent(), diff --git a/crates/buzz-session-control/tests/fixtures/start.request.event.json b/crates/buzz-session-control/tests/fixtures/start.request.event.json new file mode 100644 index 00000000000..df093781af7 --- /dev/null +++ b/crates/buzz-session-control/tests/fixtures/start.request.event.json @@ -0,0 +1 @@ +{"id":"9fa7217cd73a5b85978841e9252bab810a19f60da2b8228faed51df2ca21dd2f","pubkey":"3c72addb4fdf09af94f0c94d7fe92a386a7e70cf8a1d85916386bb2535c7b1b1","created_at":1800000000,"kind":24210,"tags":[["p","4f355bdcb7cc0af728ef3cceb9615d90684bb5b2ca5f859ab0f0b704075871aa"],["expiration","1800000120"]],"content":"At3b/ya4vXsbz0rSFZxwt9zOZWKYyZqcwRu8YWRCeZNmK2GyCgPUn22aWU0e7F4SXSNvz7FrRJW8w2gmBhCu7BSEXv/aiQ5viyMTqA55kdQ0L+EMgRNwmm+mERdXgFMePOGRGCq9lMhrOCpz8mJAWDoYXPrAYQKmepRmhk9FqgjRr1dZNfMQtpMZNoiy5BQ1r993JIUMeHIwIa2ky7kAJ06R6Hf93/NZZuoSeoOeafpHUup3M7VVKlEH6ctaQTg12zEOgbkFFPkDXPI6EfqXKUp+SHFYj41LFefOa5f9yk2XOyL3ga5ACXWjzu6rKBzIbmNHJMQ2C2/V56mfiuoungmVRG9CuR9jrwpc/TwEmDecLYrH5W+DhEJ3ytCrawdXF1JRzZ1fCNidVG75B9e0g50GGn0jCk92Ayno9JuxbSILvuVvajnDmAyMkbAWuZZoNfvYre7QIdRkP91gg/X7VnxJrgCo5Atbgv2pc6warGcS2lfKKs1HvrHsy6uFS02SmLfG0NdhjeOf4My6wChJqB6dv989Z6rQICiGCH4cXFR/2bDN/HgoDSkRn58BhtRXMEcSZyly6xTgumER7YsBajSXplw/J6XljjH8srRRn5kWvatpRJDRu9i/r8WrQYKKmpqcw6hPfyAkhzNHHszorSv7/wBGCH2nKwjvx0oQs3vngCa2ZKxzNP2f2ar9wMR/Slzf97oebycEf+fM5SJEj/rggHOWtdtPb02HzrFlDicm09FHVgMhvIsScJfVr9l5VMsUv3xm+symcgY2XkpbU8Ct6xK/sYEnVmS9zFE7kTqsjlzg/BqW+Ya/bEvvoLX7QmCRNoHooYC6M3HnBQN6doIn4cYaahp7xdrYkci2hhxmR1QlCMTVf3/njK00hPq/RkEuCQmiARZmX7OE+PhTQOydq1xOuF8PF5yZHmkl396kHspBDhQ9l+gWnKq/R7TgajqclBfComm5OgvWjYu99u/WOIY8btnGraXoAKqpjslfC6DvYggd8FEm2Wci0Ni4OoTPURFQgDd/I6YeGJbuo4MVWa49dCWWHxEAZX0r0OZlyahMQlDITTPXf9A4TQqjDaS+4DtPScLyG7cfi2wmnj38BQ==","sig":"2895a3dec0707ca6c0beff97c8e34b13c7be042ca6004bc2e0fca4193baa82a301a0353000c3813db86ea128f0e8a93effbac1e6e39927650aebb401ecdd494a"} diff --git a/docs/mobile-remote-agent-launch.md b/docs/mobile-remote-agent-launch.md new file mode 100644 index 00000000000..3e20e70daa7 --- /dev/null +++ b/docs/mobile-remote-agent-launch.md @@ -0,0 +1,98 @@ +# Start remote agents from mobile + +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. + +## Behavior + +```mermaid +sequenceDiagram + participant Phone + participant Relay + participant Manager + participant Sandbox + Phone->>Phone: Save startup intent + Phone->>Relay: Publish signed channel message + Relay-->>Phone: Confirm message acceptance + Phone->>Relay: Encrypted, owner-signed start request + Relay->>Manager: Deliver kind 24210 request + Manager->>Manager: Verify owner, mention, and saved identity + Manager->>Sandbox: Start the message's thread session + Manager-->>Phone: Encrypted startup result through relay +``` + +The manager derives the channel, thread root, and replay floor from the original signed message. +The harness replays the triggering message after startup. +An existing session keeps its current workspace and configuration. +An ended session requires explicit recovery from desktop, or a new thread. + +The phone shows startup progress above the composer. +A startup failure leaves the accepted message in place and offers **Retry** and **Dismiss**. +Retry preserves the launch intent and message, with a new operation ID to escape cached refusals. +The provider converges on the same thread session; retry never republishes the message. +Interrupted attempts retain their operation ID when resumed after restart. +Pending intents survive app restarts and remain scoped to the community and owner. +The complete relay endpoint, including its path and query, separates communities sharing a host. +On restart, an unconfirmed message must be found on the relay before its agent can start. +Each account and community retain at most 32 intents, with at most four starts in flight. + +Only owner-signed kind 30177 records with runtime `voltage.remote-workspace/sandbox` enable automatic starts. +Other owners' agents, local agents, and display-only `mention` references do not request startup. +If eligibility lookup fails, the message still sends and a dismissible notice explains that startup was skipped. +Malformed or unverified listings cannot authorize startup. +Once an eligible agent is confirmed, its startup intent must save successfully before the message sends. +The manager key matches the desktop default; `BUZZ_SANDBOX_MANAGER_PUBKEY` can override it at build time. + +## Credentials and authorization + +The phone signs `start` as the owner and encrypts it to the manager with NIP-44 v2. +The request contains the agent pubkey, community relay URL, and original signed channel message. +It carries no agent private key, model credential, namespace, image, or launch override. + +The manager accepts `start` only from a configured owner's key, without an agent attestation. +It verifies the message signature and author, addressed agent, channel, and canonical thread root. +The verified owner selects the developer namespace. +The manager rechecks the saved agent key and current owner attestation before invoking the ordinary confined provider. +Replies must verify under the manager key and match both request ID and request event ID. +The existing freshness, expiration, replay, concurrency, and idempotency limits also apply to `start`. + +A successful agent-signed desktop deploy saves its agent payload in a mutable Kubernetes Secret named `buzz-mobile-launch-*`. +The registration binds the owner, community, and agent, and retains the last successful desktop launch settings. +The manager replaces it with a resourceVersion check and at most three conflict retries. +A failed save returns an error, even if the Sandbox already started; retry reconciles that same session. + +Registrations contain agent identity and launch credentials and survive Sandbox cleanup. +Cluster operators who can read namespace Secrets can read these credentials, as with existing launch Secrets. +Agent service accounts receive no Secret permissions from this feature. +Deleting the registration disables mobile starts until a subsequent desktop deploy registers it again. +Credential payloads are never logged or returned to the phone. +Serde and the Kubernetes client own the plaintext copies needed to persist and deploy a registration. + +## Rollout + +1. Apply the reviewed staging Terraform change that adds `update` on Secrets for the Sandbox manager. +2. Publish and promote the manager image containing this change through the CD repository. +3. Start each existing agent once from desktop to register its identity and settings. +4. Install a mobile build containing this change. +5. Complete the phone test below before accepting the rollout. + +Older managers return an unsupported-operation error that mobile displays as an upgrade requirement. +Unregistered agents display the one-time desktop setup instruction. +This change does not deploy a manager, sync ArgoCD, or change production permissions. + +## Phone acceptance test + +Use an owned staging Remote workspace agent and a disposable test thread. + +1. Start the agent from the updated desktop once, then close desktop. +2. From mobile, post `@Agent reply with mobile launch confirmed` as a new thread. +3. Confirm startup progress appears and the agent replies in that exact thread. +4. Mention the agent again in a reply; confirm the existing thread continues. +5. Start another new thread; confirm it receives a separate session. +6. Interrupt the connection during startup, then restore it and use **Retry**. +7. Confirm the original message appears once and startup either completes or shows an actionable error. +8. Restart mobile with a pending launch; confirm its intent remains available. +9. Switch communities during startup; confirm no launch or notice crosses the community boundary. +10. Mention another owner's agent and confirm mobile sends no owner-start request for it. + +Record the phone platform, build, and observed result in the PR. diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 2414a305923..bdea3330758 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1857,12 +1857,12 @@ subscribers and never stores it. No stored fallback kind exists. | Event | Signer | Clear-text tags | Encrypted content (NIP-44 v2) | | --- | --- | --- | --- | -| Request | agent key (`private_key_nsec`); owner key for `enroll` and `profile` | `["p", ]`, the owner's NIP-OA `auth` tag (not on `enroll` or `profile`), `expiration` | `{"v":1,"op","request_id","payload"}`, to the manager | +| Request | agent key (`private_key_nsec`); owner key for `enroll`, `profile`, and `start` | `["p", ]`, the owner's NIP-OA `auth` tag (agent-signed operations only), `expiration` | `{"v":1,"op","request_id","payload"}`, to the manager | | Reply | manager key | `["p", ]`, `["e", ]`, `expiration` | `{"v":1,"request_id","status","response"}`, to the requester | -- `op` is `info`, `deploy`, `stop`, `enroll` or `profile`. Recovery is a `deploy` with +- `op` is `info`, `deploy`, `stop`, `enroll`, `profile` or `start`. Recovery is a `deploy` with `provider_config.sandbox.recovery`, as in the stdin protocol. -- `payload` is the provider protocol request, the same JSON as the stdin +- For `info`, `deploy`, and `stop`, `payload` is the provider protocol request, the same JSON as the stdin request, including `op`. `response` is the provider protocol response, including its `code` on a refusal. - `status` is `accepted`, then optionally `progress`, then `done`. Only @@ -1916,20 +1916,28 @@ authorizes each request in this order: 1. The event id and signature verify, and the event passes the parser checks above. Otherwise the manager ignores the event: it cannot address a reply. -2. The `auth` tag verifies for the signer at the event's `created_at` +2. For agent-signed operations, the `auth` tag verifies for the signer at the event's `created_at` (`verify_auth_tag_for_auth_event`), and its owner is a configured developer. -3. The payload's `private_key_nsec` is the signer's key, and the payload's - `op` matches the envelope's. +3. Provider payloads bind `private_key_nsec` to the signer, and their `op` matches the envelope's. + A `start` payload instead proves an owner-signed channel mention addressing its requested agent. 4. `created_at` is within 60 s of the manager's clock, and the event has not expired. 5. The manager has not received the event id in the last 10 minutes. 6. The namespace is the verified owner's developer namespace from the configuration file. It replaces `provider_config.namespace`, and the verified agent and owner replace `identity_policy`. The manager runs Sandbox sessions only. -7. `enroll` and `profile` are signed by the owner key itself, without an +7. `enroll`, `profile`, and `start` are signed by the owner key itself, without an `auth` tag. +**Mobile start.** `start` is an owner-authorized operation that uses a previously registered desktop launch. +Its payload is `{agent_pubkey, relay_url, trigger}`, where `trigger` is the complete owner-signed channel message addressing the agent. +The manager derives the thread scope and replay floor from that message, then uses the ordinary confined Sandbox deploy path. +A successful desktop deploy retains its launch payload in an owner-, community-, and agent-bound `buzz-mobile-launch-*` Secret. +These registrations survive individual Sandbox cleanup and require manager Secret `update` permission. +The phone receives no agent credentials or launch overrides. +See [mobile remote-agent launch](mobile-remote-agent-launch.md) for retention, authorization, rollout, and acceptance tests. + Then the manager replies `accepted` and runs the request in-process with its own in-cluster credentials (`provider::run`), and replies `done` with the provider response. A retry is a new event with the same `request_id`: while @@ -1948,6 +1956,10 @@ Refusals are `done` replies with one of these codes | `stale_request` | `created_at` is more than 60 s from the manager's clock, or the event has expired. | fix the clock, then retry | | `replayed_request` | The manager has already received this event. | retry with a new event | | `request_id_reused` | The `request_id` belongs to a different request. | use a new `request_id` | +| `launch_not_registered` | No saved launch exists for this owner, community and agent. | start the agent once from the updated desktop | +| `launch_store_failed` | The saved launch could not be read, verified, or updated. | retry; ask an operator to check manager Secret access | +| `invalid_launch_registration` | The desktop launch does not name a valid WebSocket relay endpoint. | use a valid `ws://` or `wss://` relay URL | +| `invalid_start` | The payload is not a valid owner-signed channel mention of the requested agent. | send a new signed mention from the owning account | | `manager_busy` | The manager is at its limit: 16 running requests, 1,024 tracked requests, 10,000 recent event ids, or 8 events for one request. | retry later | | `unsupported_op` | An unknown operation. | update Buzz | | `unsupported_version` | An envelope version other than `1`. | update Buzz | diff --git a/mobile/lib/app.dart b/mobile/lib/app.dart index 03c73bb9155..d5e5bac9b13 100644 --- a/mobile/lib/app.dart +++ b/mobile/lib/app.dart @@ -12,6 +12,7 @@ import 'package:uuid/uuid.dart'; import 'features/age_gate/age_restriction_page.dart'; import 'features/age_gate/age_signal_provider.dart'; +import 'shared/agents/remote_agent_launch_provider.dart'; import 'features/activity/activity_provider.dart'; import 'features/activity/inbox_local_state_provider.dart'; import 'features/activity/inbox_read_state.dart'; @@ -353,6 +354,7 @@ class App extends HookConsumerWidget { ref.watch(relaySessionProvider); ref.watch(observerRelayProvider); ref.watch(appLifecycleProvider); + ref.watch(remoteAgentLaunchesProvider); ref.watch(userStatusCacheProvider); if (ref.watch(activeCommunityProvider).value?.pushNotificationsEnabled == true && diff --git a/mobile/lib/features/channels/channel_detail_page.dart b/mobile/lib/features/channels/channel_detail_page.dart index 5549d6af635..2ea5ad93fb6 100644 --- a/mobile/lib/features/channels/channel_detail_page.dart +++ b/mobile/lib/features/channels/channel_detail_page.dart @@ -1,4 +1,6 @@ import 'dart:async'; + +import '../../shared/agents/remote_agent_launch_status.dart'; import 'dart:math' show cos, max, min, pi; import 'dart:ui' show ImageFilter, lerpDouble; @@ -832,6 +834,7 @@ class ChannelDetailPage extends HookConsumerWidget { entries: typingEntries, ), ), + RemoteAgentLaunchStatus(channelId: channel.id), ComposeBar( channelId: channel.id, focusNode: composerFocusNode, diff --git a/mobile/lib/features/channels/send_message_provider.dart b/mobile/lib/features/channels/send_message_provider.dart index e97c9198f29..8a74afe661f 100644 --- a/mobile/lib/features/channels/send_message_provider.dart +++ b/mobile/lib/features/channels/send_message_provider.dart @@ -1,5 +1,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; +import '../../shared/agents/remote_agent_launch_provider.dart'; + import '../../shared/mentions/mention_tags.dart'; import '../../shared/relay/relay.dart'; import '../channels/channel_management_provider.dart'; @@ -22,6 +24,9 @@ class SendMessage { final void Function(String channelId, String eventId) _completeLocalMessage; final void Function(String channelId, String eventId) _removeLocalMessage; final bool Function()? _isDeliveryValid; + final Future Function(NostrEvent)? _prepareAgentLaunch; + final Future Function(NostrEvent)? _acceptAgentLaunch; + final Future Function(NostrEvent)? _failAgentLaunch; SendMessage({ required SignedEventRelay signedEventRelay, @@ -35,6 +40,9 @@ class SendMessage { completeLocalMessage, required void Function(String channelId, String eventId) removeLocalMessage, bool Function()? isDeliveryValid, + Future Function(NostrEvent)? prepareAgentLaunch, + Future Function(NostrEvent)? acceptAgentLaunch, + Future Function(NostrEvent)? failAgentLaunch, }) : _signedEventRelay = signedEventRelay, _fetchMembers = fetchMembers, _readUserCache = readUserCache, @@ -43,7 +51,10 @@ class SendMessage { markLocalMessageForAnimation ?? ((_, _) {}), _completeLocalMessage = completeLocalMessage, _removeLocalMessage = removeLocalMessage, - _isDeliveryValid = isDeliveryValid; + _isDeliveryValid = isDeliveryValid, + _prepareAgentLaunch = prepareAgentLaunch, + _acceptAgentLaunch = acceptAgentLaunch, + _failAgentLaunch = failAgentLaunch; /// Send a text message to a channel. /// @@ -119,19 +130,30 @@ class SendMessage { kind: EventKind.streamMessage, content: content, tags: tags, - onSigned: (event) { + onSigned: (event) async { localMessage = event; _markLocalMessageForAnimation(channelId, event.id); _addLocalMessage(channelId, event); + await _prepareAgentLaunch?.call(event); + _ensureDeliveryValid(); }, ); final event = localMessage; - if (event != null) _completeLocalMessage(channelId, event.id); + if (event != null && _isDeliveryValid?.call() != false) { + _completeLocalMessage(channelId, event.id); + } } catch (_) { final event = localMessage; - if (event != null) _removeLocalMessage(channelId, event.id); + if (event != null) { + if (_isDeliveryValid?.call() != false) { + _removeLocalMessage(channelId, event.id); + } + await _failAgentLaunch?.call(event); + } rethrow; } + final event = localMessage; + if (event != null) await _acceptAgentLaunch?.call(event); } /// Resolve every identity that is actually a current member of this DM. @@ -258,6 +280,7 @@ class SendMessage { final sendMessageProvider = Provider((ref) { final config = ref.watch(relayConfigProvider); + final launches = ref.watch(remoteAgentLaunchesProvider.notifier).queue; return SendMessage( signedEventRelay: SignedEventRelay( session: ref.read(relaySessionProvider.notifier), @@ -266,6 +289,15 @@ final sendMessageProvider = Provider((ref) { fetchMembers: (channelId) => ref.read(channelMembersProvider(channelId).future), readUserCache: () => ref.read(userCacheProvider), + prepareAgentLaunch: (event) async { + await launches?.prepare(event); + }, + acceptAgentLaunch: (event) async { + await launches?.accepted(event); + }, + failAgentLaunch: (event) async { + await launches?.failed(event); + }, addLocalMessage: (channelId, event) => ref .read(channelMessagesProvider(channelId).notifier) .addLocalMessage(event), diff --git a/mobile/lib/features/channels/thread_detail_page.dart b/mobile/lib/features/channels/thread_detail_page.dart index eaaa8cdac6c..d89f46f5d96 100644 --- a/mobile/lib/features/channels/thread_detail_page.dart +++ b/mobile/lib/features/channels/thread_detail_page.dart @@ -1,4 +1,6 @@ import 'dart:async'; + +import '../../shared/agents/remote_agent_launch_status.dart'; import 'dart:math' as math; import 'package:flutter/foundation.dart' show ValueListenable; @@ -937,6 +939,10 @@ class ThreadDetailPage extends HookConsumerWidget { channelId: channelId, entries: threadTyping, ), + RemoteAgentLaunchStatus( + channelId: channelId, + threadRoot: effectiveRootId, + ), ComposeBar( channelId: channelId, focusNode: composerFocusNode, diff --git a/mobile/lib/shared/agents/agent_launch.dart b/mobile/lib/shared/agents/agent_launch.dart new file mode 100644 index 00000000000..faf4f5b2878 --- /dev/null +++ b/mobile/lib/shared/agents/agent_launch.dart @@ -0,0 +1,69 @@ +import '../relay/relay.dart'; + +String _threadRoot(NostrEvent message) { + for (final marker in ['root', 'reply']) { + for (final tag in message.tags) { + if (tag.length >= 4 && tag[0] == 'e' && tag[3] == marker) return tag[1]; + } + } + return message.id; +} + +/// A transient, community-scoped notice; it never authorizes an agent launch. +class AgentLaunchNotice { + final NostrEvent message; + final String text; + const AgentLaunchNotice(this.message, this.text); + String get threadRoot => _threadRoot(message); +} + +/// A durable startup intent. It contains a signed message, never agent secrets. +class AgentLaunch { + final String agent; + final String name; + final NostrEvent message; + final bool accepted; + + /// A persisted generation for explicit retries; restarts reuse this value. + final int attempt; + final String? error; + + const AgentLaunch({ + required this.agent, + required this.name, + required this.message, + this.accepted = false, + this.attempt = 0, + this.error, + }); + + String get id => '${message.id}:$agent'; + String get threadRoot => _threadRoot(message); + + AgentLaunch copyWith({bool? accepted, int? attempt, String? error}) => + AgentLaunch( + agent: agent, + name: name, + message: message, + accepted: accepted ?? this.accepted, + attempt: attempt ?? this.attempt, + error: error, + ); + + Map toJson() => { + 'agent': agent, + 'name': name, + 'message': message.toJson(), + 'accepted': accepted, + 'attempt': attempt, + 'error': error, + }; + factory AgentLaunch.fromJson(Map value) => AgentLaunch( + agent: value['agent'] as String, + name: value['name'] as String, + message: NostrEvent.fromJson(value['message'] as Map), + accepted: value['accepted'] as bool, + attempt: (value['attempt'] as int?) ?? 0, + error: value['error'] as String?, + ); +} diff --git a/mobile/lib/shared/agents/mobile_agent_launch_queue.dart b/mobile/lib/shared/agents/mobile_agent_launch_queue.dart new file mode 100644 index 00000000000..ba6a5ae84e0 --- /dev/null +++ b/mobile/lib/shared/agents/mobile_agent_launch_queue.dart @@ -0,0 +1,327 @@ +import 'dart:async'; +import 'dart:convert'; + +import '../relay/relay.dart'; +import 'agent_launch.dart'; +import 'remote_agent_start_client.dart'; + +/// Persists startup intents before message publication. Retries launch the +/// original signed message, never republish it or borrow another thread. +class MobileAgentLaunchQueue { + final String owner; + final RelaySessionNotifier session; + final Future Function() load; + final Future Function(String) save; + final Future Function(String agent, NostrEvent message, int attempt) + start; + final bool Function() isCurrent; + final bool Function() isConnected; + final void Function(List) onChanged; + final void Function(AgentLaunchNotice)? onLookupSkipped; + final Map _entries = {}; + final Set _running = {}; + final Set _publishing = {}; + Future _tail = Future.value(); + late final Future ready = _restore(); + static const maxEntries = 32; + static const maxConcurrent = 4; + + MobileAgentLaunchQueue({ + required this.owner, + required this.session, + required this.load, + required this.save, + required this.start, + required this.isCurrent, + required this.isConnected, + required this.onChanged, + this.onLookupSkipped, + }); + + Future _restore() async { + final text = await load(); + if (!isCurrent()) return; + if (text != null) { + final values = jsonDecode(text) as List; + if (values.length > maxEntries) { + throw const FormatException('Too many saved agent launches'); + } + for (final value in values) { + final entry = AgentLaunch.fromJson(value as Map); + if (entry.attempt < 0) { + throw const FormatException('Invalid launch attempt'); + } + verifyAgentEvent(entry.message); + if (entry.message.pubkey != owner) { + throw const FormatException( + 'Saved launch belongs to another account', + ); + } + _entries[entry.id] = entry; + } + } + _notify(); + } + + void _notify() { + if (isCurrent()) onChanged(List.unmodifiable(_entries.values)); + } + + Future _change(void Function(Map) change) { + final result = _tail.then((_) async { + if (!isCurrent()) throw StateError('The active community changed'); + final next = Map.from(_entries); + change(next); + await save(jsonEncode(next.values.map((e) => e.toJson()).toList())); + if (!isCurrent()) return; + _entries + ..clear() + ..addAll(next); + _notify(); + }); + _tail = result.then((_) {}, onError: (Object _, StackTrace _) {}); + return result; + } + + /// Only owner-signed Remote workspace listings may request startup. A + /// profile, bot role or another owner's listing cannot authorize a launch. + Future prepare(NostrEvent message) async { + if (message.pubkey != owner) throw StateError('The active account changed'); + final recipients = message.tags + .where((t) => t.length >= 2 && t[0] == 'p') + .map((t) => t[1]) + .toSet(); + if (recipients.isEmpty) return; + List records; + try { + records = await session + .fetchHistory( + NostrFilter( + kinds: const [EventKind.managedAgent], + authors: [owner], + tags: {'#d': recipients.toList()}, + limit: recipients.length, + ), + timeout: const Duration(seconds: 2), + ) + .timeout(const Duration(seconds: 2)); + } catch (_) { + if (!isCurrent()) throw StateError('The active community changed'); + _lookupSkipped(message); + return; + } + if (!isCurrent()) throw StateError('The active community changed'); + final newest = {}; + for (final record in records) { + if (record.pubkey != owner || record.kind != EventKind.managedAgent) { + continue; + } + try { + verifyAgentEvent(record); + } catch (_) { + continue; + } + final agent = record.getTagValue('d'); + if (agent == null || !recipients.contains(agent)) continue; + final prior = newest[agent]; + if (prior == null || + record.createdAt > prior.createdAt || + (record.createdAt == prior.createdAt && + record.id.compareTo(prior.id) < 0)) { + newest[agent] = record; + } + } + final launches = []; + for (final record in newest.values) { + if (record.getTagValue('runtime') != 'voltage.remote-workspace/sandbox') { + continue; + } + Map body; + try { + body = jsonDecode(record.content) as Map; + } catch (_) { + _lookupSkipped(message); + continue; + } + if (utf8.encode(jsonEncode(message.toJson())).length > 60000) { + throw ArgumentError( + 'This message is too large to start an agent. Shorten it and try again.', + ); + } + final agent = record.getTagValue('d')!; + final name = body['name'] is String ? body['name'] as String : 'agent'; + launches.add(AgentLaunch(agent: agent, name: name, message: message)); + } + if (launches.isEmpty) return; + // Only a confirmed eligible launch depends on the durable journal. + await ready; + _publishing.add(message.id); + try { + await _change((entries) { + for (final launch in launches) { + entries[launch.id] = launch; + } + if (entries.length > maxEntries) { + throw StateError( + 'Retry or dismiss previous agent launches before sending another.', + ); + } + }); + } catch (_) { + _publishing.remove(message.id); + rethrow; + } + } + + void _lookupSkipped(NostrEvent message) { + if (isCurrent()) { + onLookupSkipped?.call( + AgentLaunchNotice( + message, + 'Agent startup was skipped because its settings could not be checked.', + ), + ); + } + } + + Future accepted(NostrEvent message) async { + try { + if (!message.tags.any((tag) => tag.isNotEmpty && tag[0] == 'p')) return; + if (!_entries.values.any((entry) => entry.message.id == message.id)) { + return; + } + await ready; + try { + await _change((entries) { + for (final entry in entries.values.toList()) { + if (entry.message.id == message.id) { + entries[entry.id] = entry.copyWith(accepted: true); + } + } + }); + } catch (_) { + // The pre-publication intent is already durable. A restart/retry checks + // relay acceptance instead of losing or republishing the sent message. + _showFailure( + message.id, + 'Message sent. Startup confirmation could not be saved; retry the agent launch.', + ); + } + } finally { + _publishing.remove(message.id); + } + resume(); + } + + Future failed(NostrEvent message) async { + try { + if (!message.tags.any((tag) => tag.isNotEmpty && tag[0] == 'p')) return; + if (!_entries.values.any((entry) => entry.message.id == message.id)) { + return; + } + await ready; + const error = + 'Message delivery was not confirmed. Retry checks whether it reached the relay.'; + try { + await _change((entries) { + for (final entry in entries.values.toList()) { + if (entry.message.id == message.id) { + entries[entry.id] = entry.copyWith(error: error); + } + } + }); + } catch (_) { + _showFailure(message.id, error); + } + } finally { + _publishing.remove(message.id); + } + } + + void _showFailure(String messageId, String error) { + if (!isCurrent()) return; + for (final entry in _entries.values.toList()) { + if (entry.message.id == messageId) { + _entries[entry.id] = entry.copyWith(error: error); + } + } + _notify(); + } + + /// Resume interrupted pending launches after reconnect or process restart. + /// Rejections remain visible until the user explicitly retries or dismisses. + void resume() { + if (!isCurrent() || !isConnected()) return; + for (final entry in _entries.values.toList()) { + if (_running.length >= maxConcurrent) break; + if (entry.error != null || + _running.contains(entry.id) || + _publishing.contains(entry.message.id)) { + continue; + } + _running.add(entry.id); + unawaited(_run(entry)); + } + } + + Future retry(String id) async { + await ready; + if (_running.contains(id)) return; + await _change((entries) { + final entry = entries[id]; + if (entry != null) { + entries[id] = entry.copyWith(attempt: entry.attempt + 1); + } + }); + resume(); + } + + Future dismiss(String id) async { + if (_running.contains(id)) return; + await _change((entries) => entries.remove(id)); + } + + Future _run(AgentLaunch entry) async { + try { + if (!entry.accepted) { + final events = await session.fetchHistory( + NostrFilter( + kinds: const [9, 40002], + ids: [entry.message.id], + limit: 1, + ), + ); + final accepted = events.any((event) { + verifyAgentEvent(event); + return event.id == entry.message.id && event.pubkey == owner; + }); + if (!accepted) { + throw const AgentStartException( + 'The message was not found on the relay. Send the message again to start the agent.', + ); + } + } + if (!isCurrent() || !isConnected()) return; + await start(entry.agent, entry.message, entry.attempt); + await _change((entries) => entries.remove(entry.id)); + } catch (error) { + if (!isCurrent()) return; + final message = error is AgentStartException + ? error.message + : 'Startup could not be confirmed. Retry the agent launch.'; + try { + await _change((entries) { + final current = entries[entry.id]; + if (current != null) { + entries[entry.id] = current.copyWith(error: message); + } + }); + } catch (_) { + _showFailure(entry.message.id, message); + } + } finally { + _running.remove(entry.id); + resume(); + } + } +} diff --git a/mobile/lib/shared/agents/remote_agent_launch_provider.dart b/mobile/lib/shared/agents/remote_agent_launch_provider.dart new file mode 100644 index 00000000000..bcd0c1679b1 --- /dev/null +++ b/mobile/lib/shared/agents/remote_agent_launch_provider.dart @@ -0,0 +1,134 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:pointycastle/digests/sha256.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +import '../relay/relay.dart'; +import 'agent_launch.dart'; +import 'mobile_agent_launch_queue.dart'; +import 'remote_agent_start_client.dart'; + +/// Include the full endpoint so communities sharing a host keep separate queues. +String remoteAgentLaunchStorageKey(RelayConfig config, String owner) { + final digest = SHA256Digest() + .process(Uint8List.fromList(utf8.encode(config.wsUrl))) + .map((byte) => byte.toRadixString(16).padLeft(2, '0')) + .join(); + return 'mobile-agent-launches:v1:$digest:$owner'; +} + +/// A dismissible notice that cannot prevent ordinary message publication. +class RemoteAgentLaunchNotice extends Notifier { + @override + AgentLaunchNotice? build() { + ref.watch(relayConfigProvider); + return null; + } + + /// Show the most recent lookup failure in its originating conversation. + void show(AgentLaunchNotice notice) => state = notice; + + /// Dismiss the transient notice. + void dismiss() => state = null; +} + +final remoteAgentLaunchNoticeProvider = + NotifierProvider( + RemoteAgentLaunchNotice.new, + ); + +class RemoteAgentLaunches extends Notifier>> { + MobileAgentLaunchQueue? queue; + + @override + AsyncValue> build() { + final config = ref.watch(relayConfigProvider); + final owner = pubkeyFromNsec(config.nsec); + if (owner == null) { + queue = null; + return const AsyncData([]); + } + final session = ref.read(relaySessionProvider.notifier); + var active = true; + bool current() => + active && + ref.mounted && + ref.read(relayConfigProvider).baseUrl == config.baseUrl && + ref.read(relayConfigProvider).nsec == config.nsec; + final client = RemoteAgentStartClient( + session: session, + config: config, + isCurrent: current, + ); + ref.onDispose(() { + active = false; + client.dispose(); + }); + final storageKey = remoteAgentLaunchStorageKey(config, owner); + final localQueue = MobileAgentLaunchQueue( + owner: owner, + session: session, + isCurrent: current, + isConnected: () => + ref.read(relaySessionProvider).status == SessionStatus.connected, + start: (agent, message, attempt) => + client.start(agent, message, attempt: attempt), + onLookupSkipped: (notice) { + if (current()) { + ref.read(remoteAgentLaunchNoticeProvider.notifier).show(notice); + } + }, + load: () async => + (await SharedPreferences.getInstance()).getString(storageKey), + save: (value) async { + final saved = await (await SharedPreferences.getInstance()).setString( + storageKey, + value, + ); + if (!saved) throw StateError('Agent launch history could not be saved'); + }, + onChanged: (entries) { + if (current()) state = AsyncData(entries); + }, + ); + queue = localQueue; + ref.listen(relaySessionProvider.select((s) => s.status), (_, status) { + if (status != SessionStatus.connected) client.cancelActive(); + if (status == SessionStatus.connected) { + unawaited( + localQueue.ready.then( + (_) { + if (current()) localQueue.resume(); + }, + onError: (Object e, StackTrace st) { + if (current()) state = AsyncError(e, st); + }, + ), + ); + } + }); + unawaited( + localQueue.ready.then( + (_) { + if (current() && + ref.read(relaySessionProvider).status == + SessionStatus.connected) { + localQueue.resume(); + } + }, + onError: (Object e, StackTrace st) { + if (current()) state = AsyncError(e, st); + }, + ), + ); + return const AsyncLoading(); + } +} + +final remoteAgentLaunchesProvider = + NotifierProvider>>( + RemoteAgentLaunches.new, + ); diff --git a/mobile/lib/shared/agents/remote_agent_launch_status.dart b/mobile/lib/shared/agents/remote_agent_launch_status.dart new file mode 100644 index 00000000000..6d5e866b342 --- /dev/null +++ b/mobile/lib/shared/agents/remote_agent_launch_status.dart @@ -0,0 +1,133 @@ +import 'package:flutter/material.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; + +import '../theme/grid.dart'; +import 'remote_agent_launch_provider.dart'; + +/// Startup status belongs to the channel/thread that requested it. Failed +/// starts offer retry independently of the already-sent message. +class RemoteAgentLaunchStatus extends ConsumerWidget { + final String channelId; + final String? threadRoot; + const RemoteAgentLaunchStatus({ + super.key, + required this.channelId, + this.threadRoot, + }); + + @override + Widget build(BuildContext context, WidgetRef ref) { + final launches = ref.watch(remoteAgentLaunchesProvider); + final notice = ref.watch(remoteAgentLaunchNoticeProvider); + final showNotice = + notice != null && + notice.message.channelId == channelId && + (threadRoot == null || notice.threadRoot == threadRoot); + if (launches.hasError) { + return ListTile( + title: const Text('Agent startup history could not be loaded.'), + trailing: TextButton( + onPressed: () => ref.invalidate(remoteAgentLaunchesProvider), + child: const Text('Retry'), + ), + ); + } + final entries = (launches.value ?? []) + .where( + (entry) => + entry.message.channelId == channelId && + (threadRoot == null || entry.threadRoot == threadRoot), + ) + .toList(); + if (entries.isEmpty && !showNotice) return const SizedBox.shrink(); + final queue = ref.read(remoteAgentLaunchesProvider.notifier).queue; + return ConstrainedBox( + constraints: const BoxConstraints(maxHeight: 180), + child: SingleChildScrollView( + child: Column( + mainAxisSize: MainAxisSize.min, + children: [ + if (showNotice) + ListTile( + leading: const Icon(Icons.info_outline), + title: Text(notice.text), + trailing: IconButton( + tooltip: 'Dismiss agent startup notice', + onPressed: () => ref + .read(remoteAgentLaunchNoticeProvider.notifier) + .dismiss(), + icon: const Icon(Icons.close), + ), + ), + for (final entry in entries) + Padding( + padding: const EdgeInsets.symmetric( + horizontal: Grid.twelve, + vertical: Grid.half, + ), + child: Row( + children: [ + if (entry.error == null) + const SizedBox( + width: 16, + height: 16, + child: CircularProgressIndicator(strokeWidth: 2), + ) + else + const Icon(Icons.error_outline, size: 18), + const SizedBox(width: Grid.xxs), + Expanded( + child: Text( + entry.error == null + ? 'Starting ${entry.name}…' + : '${entry.name}: ${entry.error}', + ), + ), + if (entry.error != null) ...[ + TextButton( + onPressed: () async { + try { + await queue?.retry(entry.id); + } catch (_) { + if (context.mounted) { + ScaffoldMessenger.of(context).showSnackBar( + const SnackBar( + content: Text( + 'The launch retry could not be saved. Try again.', + ), + ), + ); + } + } + }, + child: const Text('Retry'), + ), + IconButton( + tooltip: 'Dismiss startup for ${entry.name}', + onPressed: () async { + try { + await queue?.dismiss(entry.id); + } catch (_) { + if (context.mounted) { + ScaffoldMessenger.of(context).showSnackBar( + const SnackBar( + content: Text( + 'The startup notice could not be dismissed. Try again.', + ), + ), + ); + } + } + }, + icon: const Icon(Icons.close, size: 18), + ), + ], + ], + ), + ), + ], + ), + ), + ); + } +} diff --git a/mobile/lib/shared/agents/remote_agent_start_client.dart b/mobile/lib/shared/agents/remote_agent_start_client.dart new file mode 100644 index 00000000000..fc1a764a3bb --- /dev/null +++ b/mobile/lib/shared/agents/remote_agent_start_client.dart @@ -0,0 +1,245 @@ +import 'dart:async'; +import 'dart:convert'; + +import '../crypto/nip44.dart'; +import '../relay/relay.dart'; +import 'package:nostr/nostr.dart' as nostr; +import 'package:uuid/uuid.dart'; + +/// Same public manager identity used by desktop's Remote workspace type. +const remoteAgentManagerPubkey = String.fromEnvironment( + 'BUZZ_SANDBOX_MANAGER_PUBKEY', + defaultValue: + '1881c0c6186c81c00e607f9dcd9b6c655c49cd9a0227054e764a78e6829d31fd', +); + +class AgentStartException implements Exception { + final String message; + const AgentStartException(this.message); + @override + String toString() => message; +} + +/// Starts a registered agent over the existing encrypted session-control wire. +/// The phone signs as the owner; it never receives the agent's private key. +class RemoteAgentStartClient { + final RelaySessionNotifier session; + final RelayConfig config; + final String manager; + final bool Function() isCurrent; + final Set _cancellations = {}; + final Duration acceptTimeout; + final Duration finishTimeout; + + RemoteAgentStartClient({ + required this.session, + required this.config, + required this.isCurrent, + this.manager = remoteAgentManagerPubkey, + this.acceptTimeout = const Duration(seconds: 30), + this.finishTimeout = const Duration(minutes: 10), + }); + + void dispose() { + cancelActive(); + } + + /// Leave durable intents retryable when the relay connection is interrupted. + void cancelActive() { + for (final cancel in _cancellations.toList()) { + cancel(); + } + _cancellations.clear(); + } + + Future start( + String agent, + NostrEvent trigger, { + int attempt = 0, + }) async { + final nsec = config.nsec; + final owner = pubkeyFromNsec(nsec); + if (owner == null || !isCurrent() || trigger.pubkey != owner) { + throw const AgentStartException( + 'Agent launch cancelled because the account changed.', + ); + } + final key = getConversationKey( + nostr.Nip19.decode(payload: nsec!).data, + manager, + ); + final requestId = const Uuid().v5( + Namespace.url.value, + '${config.wsUrl}/$agent/${trigger.id}${attempt == 0 ? '' : '/$attempt'}', + ); + final done = Completer(); + final ready = Completer(); + String? requestEvent; + Timer? acceptTimer; + void fail(String message) { + if (!done.isCompleted) done.completeError(AgentStartException(message)); + } + + // Attach the error handler before subscribing, publishing or awaiting EOSE. + void cancel() => + fail('Startup confirmation was interrupted. Retry the agent launch.'); + _cancellations.add(cancel); + final result = done.future.timeout( + finishTimeout, + onTimeout: () => throw const AgentStartException( + 'The agent has not confirmed startup. Retry checks the same launch.', + ), + ); + result.ignore(); + void Function()? unsubscribe; + try { + unsubscribe = await session.subscribeWithStatus( + NostrFilter( + kinds: const [EventKind.sandboxSessionControl], + authors: [manager], + tags: { + '#p': [owner], + }, + limit: 0, + ), + (event) { + if (!isCurrent() || + done.isCompleted || + event.pubkey != manager || + event.kind != EventKind.sandboxSessionControl) { + return; + } + try { + verifyAgentEvent(event); + final tags = {}; + for (final tag in event.tags) { + if (tag.length != 2 || + !const ['p', 'e', 'expiration'].contains(tag[0]) || + tags.containsKey(tag[0])) { + return; + } + tags[tag[0]] = tag[1]; + } + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + if (tags['p'] != owner || + tags['e'] != requestEvent || + (now - event.createdAt).abs() > 60 || + (int.tryParse(tags['expiration'] ?? '') ?? 0) <= now) { + return; + } + final reply = + jsonDecode(nip44Decrypt(key, event.content)) + as Map; + if (reply['v'] != 1 || reply['request_id'] != requestId) return; + if (reply['status'] == 'accepted' || + reply['status'] == 'progress') { + acceptTimer?.cancel(); + } else if (reply['status'] == 'done') { + acceptTimer?.cancel(); + final response = reply['response']; + if (response is Map && response['ok'] == true) { + done.complete(); + } else { + fail( + _failureMessage(response is Map ? response['code'] : null), + ); + } + } + } catch (_) { + // Unrelated or malformed ciphertext never completes a launch. + } + }, + onClosed: (_) => fail( + 'The connection could not confirm startup. Retry the agent launch.', + ), + onStatusChanged: (status) { + if (status == RelaySubscriptionStatus.ready && !ready.isCompleted) { + ready.complete(); + } + if (status == RelaySubscriptionStatus.retrying && + requestEvent != null) { + fail('The connection was interrupted. Retry the agent launch.'); + } + }, + ); + await Future.any([ready.future.timeout(acceptTimeout), result]); + if (!isCurrent()) { + throw const AgentStartException( + 'Agent launch cancelled because the account changed.', + ); + } + if (done.isCompleted) return await result; + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + acceptTimer = Timer( + acceptTimeout, + () => fail('The agent service did not answer. Retry the agent launch.'), + ); + final published = SignedEventRelay(session: session, nsec: nsec).submit( + kind: EventKind.sandboxSessionControl, + createdAt: now, + tags: [ + ['p', manager], + ['expiration', '${now + 120}'], + ], + content: nip44Encrypt( + key, + jsonEncode({ + 'v': 1, + 'op': 'start', + 'request_id': requestId, + 'payload': { + 'agent_pubkey': agent, + 'relay_url': config.wsUrl, + 'trigger': trigger.toJson(), + }, + }), + ), + onSigned: (event) => requestEvent = event.id, + ); + // A verified manager completion also proves delivery when the relay's + // separate OK is delayed or lost. Still consume a later publish failure. + await Future.any([published.then((_) {}), result]); + await result; + } on AgentStartException { + rethrow; + } catch (_) { + throw const AgentStartException( + 'The connection could not confirm startup. Retry the agent launch.', + ); + } finally { + acceptTimer?.cancel(); + if (!ready.isCompleted) ready.complete(); + _cancellations.remove(cancel); + unsubscribe?.call(); + // End the total deadline even when subscribe/publish failed first. + if (!done.isCompleted) done.complete(); + } + } +} + +/// Verify received control events and owner listings before using their data. +void verifyAgentEvent(NostrEvent event) { + nostr.Event( + event.id, + event.pubkey, + event.createdAt, + event.kind, + event.tags, + event.content, + event.sig, + ); +} + +String _failureMessage(Object? code) => switch (code) { + 'launch_not_registered' => + 'Start this agent once from desktop after the server update to enable mobile launches.', + 'unsupported_op' => + 'The agent service needs an update before it can accept mobile launches.', + 'unauthorized_owner' || + 'signer_mismatch' => 'This account is not authorized to start the agent.', + 'session_ended' || 'session_ending' || 'recovery_interrupted' => + 'This thread’s session ended. Recover it from desktop, or start a new thread.', + 'config_differs' => + 'This thread is running with different settings. Continue there or start a new thread.', + _ => 'The agent could not start. Retry the agent launch.', +}; diff --git a/mobile/lib/shared/relay/nostr_models.dart b/mobile/lib/shared/relay/nostr_models.dart index c041e4399fa..b657500e5f6 100644 --- a/mobile/lib/shared/relay/nostr_models.dart +++ b/mobile/lib/shared/relay/nostr_models.dart @@ -24,8 +24,14 @@ abstract final class EventKind { static const typingIndicator = 20002; static const auth = 22242; static const agentObserverFrame = 24200; + + /// Encrypted requests and replies for the remote session manager. + static const sandboxSessionControl = 24210; static const huddleReaction = 24810; static const readState = 30078; + + /// Owner-signed managed-agent records, including their runtime type. + static const managedAgent = 30177; static const eventReminder = 30300; static const userStatus = 30315; static const dmVisibility = 30622; diff --git a/mobile/lib/shared/relay/signed_event_relay.dart b/mobile/lib/shared/relay/signed_event_relay.dart index b0106639eb7..1f7f3c2e0c5 100644 --- a/mobile/lib/shared/relay/signed_event_relay.dart +++ b/mobile/lib/shared/relay/signed_event_relay.dart @@ -34,7 +34,7 @@ class SignedEventRelay { required String content, required List> tags, int? createdAt, - void Function(NostrEvent event)? onSigned, + FutureOr Function(NostrEvent event)? onSigned, }) async { final nsec = _nsec; if (nsec == null || nsec.isEmpty) { @@ -56,7 +56,7 @@ class SignedEventRelay { ); final nostrEvent = NostrEvent.fromJson(event.toMap()); - onSigned?.call(nostrEvent); + await onSigned?.call(nostrEvent); return _session.publish(nostrEvent); } } diff --git a/mobile/test/features/agents/launch_fixture.dart b/mobile/test/features/agents/launch_fixture.dart new file mode 100644 index 00000000000..858e63ac005 --- /dev/null +++ b/mobile/test/features/agents/launch_fixture.dart @@ -0,0 +1,223 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:buzz/shared/agents/agent_launch.dart'; +import 'package:buzz/shared/agents/mobile_agent_launch_queue.dart'; +import 'package:buzz/shared/agents/remote_agent_start_client.dart'; +import 'package:buzz/features/channels/channel_management_provider.dart'; +import 'package:buzz/features/channels/send_message_provider.dart'; +import 'package:buzz/shared/crypto/nip44.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:nostr/nostr.dart' as nostr; + +const channelId = '11111111-1111-4111-8111-111111111111'; +final owner = nostr.Keys('33' * 32); +final agent = nostr.Keys('22' * 32); +final manager = nostr.Keys('11' * 32); + +NostrEvent signed( + String key, + int kind, + String content, + List> tags, { + int? createdAt, +}) => NostrEvent.fromJson( + nostr.Event.from( + kind: kind, + content: content, + tags: tags, + secretKey: key, + createdAt: createdAt, + verify: false, + ).toMap(), +); + +class LaunchRelay extends RelaySessionNotifier { + final published = []; + final requests = >[]; + List? listings; + Object? listingError; + void Function(NostrEvent)? listener; + Completer? messageAck; + Completer? controlAck; + void Function(NostrEvent)? beforePublish; + String? refusal; + bool cacheFailures = false; + final cachedFailures = >{}; + bool answer = true; + int unsubscribed = 0; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) async { + if (filter.kinds.contains(30177)) { + if (listingError case final error?) throw error; + return listings ?? + [ + signed('33' * 32, 30177, jsonEncode({'name': 'Bifrost'}), [ + ['d', agent.public], + ['runtime', 'voltage.remote-workspace/sandbox'], + ]), + ]; + } + return published + .where((event) => filter.ids?.contains(event.id) == true) + .toList(); + } + + @override + Future subscribeWithStatus( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String)? onClosed, + required void Function(RelaySubscriptionStatus) onStatusChanged, + }) async { + listener = onEvent; + onStatusChanged(RelaySubscriptionStatus.ready); + return () { + listener = null; + unsubscribed++; + }; + } + + @override + Future publish( + NostrEvent event, { + Duration timeout = const Duration(seconds: 8), + }) async { + beforePublish?.call(event); + published.add(event); + if (event.kind == 9) { + await messageAck?.future; + return event; + } + if (event.kind == 24210) { + final key = getConversationKey('11' * 32, owner.public); + final request = + jsonDecode(nip44Decrypt(key, event.content)) as Map; + requests.add(request); + if (answer) { + reply(event, request, 'accepted'); + final response = + cachedFailures[request['request_id']] ?? + (refusal == null + ? {'ok': true, 'agent_id': 'fixture-sandbox'} + : {'ok': false, 'code': refusal}); + if (cacheFailures && response['ok'] == false) { + cachedFailures[request['request_id'] as String] = response; + } + reply(event, request, 'done', response: response); + } + } + if (event.kind == 24210) await controlAck?.future; + return event; + } + + void reply( + NostrEvent requestEvent, + Map request, + String status, { + Map? response, + String? eventId, + String? requestId, + String? signingKey, + }) { + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + final content = nip44Encrypt( + getConversationKey('11' * 32, owner.public), + jsonEncode({ + 'v': 1, + 'request_id': requestId ?? request['request_id'], + 'status': status, + 'response': ?response, + }), + ); + listener?.call( + signed(signingKey ?? '11' * 32, 24210, content, [ + ['p', owner.public], + ['e', eventId ?? requestEvent.id], + ['expiration', '${now + 120}'], + ]), + ); + } +} + +class LaunchFixture { + final LaunchRelay relay = LaunchRelay(); + String? stored; + bool current = true; + bool connected = true; + bool denySave = false; + void Function()? onSave; + List entries = []; + final notices = []; + void Function(List)? onEntries; + late final client = RemoteAgentStartClient( + session: relay, + config: RelayConfig(baseUrl: 'https://relay.example', nsec: owner.nsec), + manager: manager.public, + isCurrent: () => current, + ); + late final queue = newQueue(); + + MobileAgentLaunchQueue newQueue() => MobileAgentLaunchQueue( + owner: owner.public, + session: relay, + load: () async => stored, + save: (value) async { + if (denySave) throw StateError('disk unavailable'); + stored = value; + onSave?.call(); + }, + start: (agent, message, attempt) => + client.start(agent, message, attempt: attempt), + isCurrent: () => current, + isConnected: () => connected, + onChanged: (value) { + entries = value; + onEntries?.call(value); + }, + onLookupSkipped: notices.add, + ); + + SendMessage sender({ + void Function(String, NostrEvent)? addLocal, + void Function(String, String)? removeLocal, + void Function(String, String)? completeLocal, + }) => SendMessage( + signedEventRelay: SignedEventRelay(session: relay, nsec: owner.nsec), + fetchMembers: (_) async => [ + ChannelMember( + pubkey: agent.public, + role: 'bot', + joinedAt: DateTime(2025), + ), + ], + readUserCache: () => const {}, + addLocalMessage: addLocal ?? (_, _) {}, + completeLocalMessage: completeLocal ?? (_, _) {}, + removeLocalMessage: removeLocal ?? (_, _) {}, + prepareAgentLaunch: queue.prepare, + acceptAgentLaunch: queue.accepted, + failAgentLaunch: queue.failed, + isDeliveryValid: () => current, + ); + + Future settle(bool Function() predicate) async { + for (var i = 0; i < 100; i++) { + if (predicate()) return; + await Future.delayed(const Duration(milliseconds: 2)); + } + throw StateError('Launch did not settle'); + } + + void dispose() { + current = false; + client.dispose(); + } +} diff --git a/mobile/test/features/agents/mobile_agent_launch_flow_test.dart b/mobile/test/features/agents/mobile_agent_launch_flow_test.dart new file mode 100644 index 00000000000..92e4fc4d0e5 --- /dev/null +++ b/mobile/test/features/agents/mobile_agent_launch_flow_test.dart @@ -0,0 +1,312 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; + +import 'launch_fixture.dart'; + +void main() { + test('resume skips a message while its publication is in flight', () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.messageAck = Completer(); + fixture.relay.beforePublish = (event) { + if (event.kind == 9) fixture.queue.resume(); + }; + final sending = fixture.sender()( + channelId: channelId, + content: '@Bifrost work', + mentionPubkeys: [agent.public], + ); + await fixture.settle(() => fixture.relay.published.isNotEmpty); + fixture.queue.resume(); + await Future.delayed(Duration.zero); + expect(fixture.relay.requests, isEmpty); + expect(fixture.entries.single.error, isNull); + fixture.relay.messageAck!.complete(); + await sending; + await fixture.settle(() => fixture.entries.isEmpty); + expect(fixture.relay.requests, hasLength(1)); + }); + + test( + 'oversized launch messages fail before publication instead of creating an unretryable intent', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + await expectLater( + fixture.sender()( + channelId: channelId, + content: 'x' * 60001, + mentionPubkeys: [agent.public], + ), + throwsArgumentError, + ); + expect(fixture.relay.published, isEmpty); + expect(fixture.entries, isEmpty); + }, + ); + + test('pending launches wait for relay reconnection', () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.connected = false; + await fixture.sender()( + channelId: channelId, + content: '@Bifrost work', + mentionPubkeys: [agent.public], + ); + expect(fixture.relay.requests, isEmpty); + fixture.connected = true; + fixture.queue.resume(); + await fixture.settle(() => fixture.entries.isEmpty); + expect(fixture.relay.requests, hasLength(1)); + }); + + test( + 'a mobile mention persists its intent before publishing and launches its exact signed root', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.messageAck = Completer(); + fixture.relay.beforePublish = (event) { + if (event.kind == 9) { + final saved = jsonDecode(fixture.stored!) as List; + expect(saved.single['message']['id'], event.id); + expect(fixture.relay.requests, isEmpty); + } + }; + final send = fixture.sender()( + channelId: channelId, + content: '@Bifrost investigate', + mentionPubkeys: [agent.public], + ); + await fixture.settle(() => fixture.relay.published.isNotEmpty); + expect( + fixture.relay.requests, + isEmpty, + reason: 'A start must wait for the message OK', + ); + fixture.relay.messageAck!.complete(); + await send; + await fixture.settle( + () => fixture.relay.requests.isNotEmpty && fixture.entries.isEmpty, + ); + final request = fixture.relay.requests.single; + expect(request['op'], 'start'); + expect( + request['payload']['trigger'], + fixture.relay.published.first.toJson(), + ); + expect(request['payload']['agent_pubkey'], agent.public); + expect(request.toString(), isNot(contains('private_key_nsec'))); + expect(fixture.relay.published.last.pubkey, owner.public); + expect(fixture.relay.unsubscribed, 1); + }, + ); + + test( + 'retry escapes a cached manager refusal without duplicating the sent message', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.refusal = 'launch_not_registered'; + fixture.relay.cacheFailures = true; + final removed = []; + await fixture.sender(removeLocal: (_, id) => removed.add(id))( + channelId: channelId, + content: '@Bifrost retry', + mentionPubkeys: [agent.public], + parentEventId: 'b' * 64, + rootEventId: 'a' * 64, + ); + await fixture.settle(() => fixture.entries.single.error != null); + final entry = fixture.entries.single; + expect(entry.threadRoot, 'a' * 64); + expect( + removed, + isEmpty, + reason: 'Agent failure cannot roll back an accepted message', + ); + final firstId = fixture.relay.requests.single['request_id']; + fixture.relay.refusal = null; + await fixture.queue.retry(entry.id); + await fixture.settle(() => fixture.entries.isEmpty); + expect( + fixture.relay.published.where((event) => event.kind == 9), + hasLength(1), + ); + expect(fixture.relay.requests.last['request_id'], isNot(firstId)); + expect( + fixture.relay.requests.last['payload'], + fixture.relay.requests.first['payload'], + ); + }, + ); + + test( + 'an interrupted explicit retry restores its persisted operation ID', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.refusal = 'launch_not_registered'; + fixture.relay.cacheFailures = true; + await fixture.sender()( + channelId: channelId, + content: '@Bifrost resume', + mentionPubkeys: [agent.public], + ); + await fixture.settle(() => fixture.entries.single.error != null); + fixture.relay.answer = false; + await fixture.queue.retry(fixture.entries.single.id); + await fixture.settle(() => fixture.relay.requests.length == 2); + final retryId = fixture.relay.requests.last['request_id']; + final journal = fixture.stored; + fixture.dispose(); + + final restored = LaunchFixture()..stored = journal; + addTearDown(restored.dispose); + await restored.queue.ready; + restored.queue.resume(); + await restored.settle(() => restored.entries.isEmpty); + expect(restored.relay.requests.single['request_id'], retryId); + expect( + restored.relay.published.where((event) => event.kind == 9), + isEmpty, + ); + }, + ); + + test( + 'a restart reconciles a pre-publication intent against the relay before starting', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + final event = signed('33' * 32, 9, '@Bifrost work', [ + ['h', channelId], + ['p', agent.public], + ]); + await fixture.queue.prepare(event); + fixture.relay.published.add( + event, + ); // Relay accepted before the old process stopped. + final restored = fixture.newQueue(); + await restored.ready; + restored.resume(); + await fixture.settle(() => fixture.entries.isEmpty); + expect( + fixture.relay.requests.single['payload']['trigger']['id'], + event.id, + ); + expect( + fixture.relay.published.where((event) => event.kind == 9), + hasLength(1), + ); + }, + ); + + test('an unaccepted message cannot start an agent after restart', () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + final event = signed('33' * 32, 9, '@Bifrost work', [ + ['h', channelId], + ['p', agent.public], + ]); + await fixture.queue.prepare(event); + final restored = fixture.newQueue(); + await restored.ready; + restored.resume(); + await fixture.settle(() => fixture.entries.single.error != null); + expect(fixture.relay.requests, isEmpty); + expect(fixture.entries.single.error, contains('not found')); + }); + + test( + 'a failed durable intent prevents message publication and preserves the draft path', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.denySave = true; + final removed = []; + await expectLater( + fixture.sender(removeLocal: (_, id) => removed.add(id))( + channelId: channelId, + content: '@Bifrost work', + mentionPubkeys: [agent.public], + ), + throwsStateError, + ); + expect(fixture.relay.published, isEmpty); + expect(removed, hasLength(1)); + }, + ); + + test( + 'another owner, local agent, and display-only reference never request startup', + () async { + for (final record in [ + signed('44' * 32, 30177, '{}', [ + ['d', agent.public], + ['runtime', 'voltage.remote-workspace/sandbox'], + ]), + signed('33' * 32, 30177, '{}', [ + ['d', agent.public], + ['runtime', 'local/sandbox'], + ]), + ]) { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.listings = [record]; + await fixture.sender()( + channelId: channelId, + content: '@Bifrost', + mentionPubkeys: [agent.public], + ); + expect(fixture.relay.requests, isEmpty); + } + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + await fixture.sender()( + channelId: channelId, + content: 'referencing Bifrost', + mentionPubkeys: [], + mediaTags: [ + ['mention', agent.public], + ], + ); + expect(fixture.relay.requests, isEmpty); + }, + ); + + test('switching communities while preparing cancels publication', () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.onSave = () => fixture.current = false; + final send = fixture.sender()( + channelId: channelId, + content: '@Bifrost', + mentionPubkeys: [agent.public], + ); + await expectLater(send, throwsStateError); + expect(fixture.relay.published, isEmpty); + expect(fixture.relay.requests, isEmpty); + }); + + test( + 'a missing registration has a useful failure and remains retryable', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.refusal = 'launch_not_registered'; + await fixture.sender()( + channelId: channelId, + content: '@Bifrost', + mentionPubkeys: [agent.public], + ); + await fixture.settle(() => fixture.entries.single.error != null); + expect(fixture.entries.single.error, contains('once from desktop')); + expect(jsonDecode(fixture.stored!) as List, hasLength(1)); + expect(fixture.entries.single.error, isNot(contains('nsec'))); + }, + ); +} diff --git a/mobile/test/features/agents/mobile_agent_lookup_failure_test.dart b/mobile/test/features/agents/mobile_agent_lookup_failure_test.dart new file mode 100644 index 00000000000..5d2d8cf2070 --- /dev/null +++ b/mobile/test/features/agents/mobile_agent_lookup_failure_test.dart @@ -0,0 +1,72 @@ +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; + +import 'launch_fixture.dart'; + +void main() { + test( + 'lookup failures leave ordinary mentions sendable and explain skipped startup', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.listingError = StateError('history unavailable'); + await fixture.sender()( + channelId: channelId, + content: '@Alice hello', + mentionPubkeys: [agent.public], + ); + expect(fixture.relay.published.map((event) => event.kind), [9]); + expect(fixture.relay.requests, isEmpty); + expect(fixture.entries, isEmpty); + expect( + fixture.notices.single.message.id, + fixture.relay.published.single.id, + ); + expect(fixture.notices.single.text, contains('startup was skipped')); + }, + ); + + test( + 'malformed and forged listings cannot block publication or authorize startup', + () async { + for (final forged in [false, true]) { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + var record = signed('33' * 32, 30177, 'unreadable', [ + ['d', agent.public], + ['runtime', 'voltage.remote-workspace/sandbox'], + ]); + if (forged) { + record = NostrEvent.fromJson(record.toJson()..['sig'] = '0' * 128); + } + fixture.relay.listings = [record]; + await fixture.sender()( + channelId: channelId, + content: '@Alice hello', + mentionPubkeys: [agent.public], + ); + expect(fixture.relay.published.map((event) => event.kind), [9]); + expect(fixture.relay.requests, isEmpty); + expect(fixture.entries, isEmpty); + } + }, + ); + + test( + 'a broken launch journal does not block a confirmed ordinary mention', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.stored = 'invalid journal'; + await expectLater(fixture.queue.ready, throwsFormatException); + fixture.relay.listings = []; + await fixture.sender()( + channelId: channelId, + content: '@Alice hello', + mentionPubkeys: [agent.public], + ); + expect(fixture.relay.published.map((event) => event.kind), [9]); + expect(fixture.relay.requests, isEmpty); + }, + ); +} diff --git a/mobile/test/features/agents/remote_agent_launch_status_test.dart b/mobile/test/features/agents/remote_agent_launch_status_test.dart new file mode 100644 index 00000000000..498359bcf60 --- /dev/null +++ b/mobile/test/features/agents/remote_agent_launch_status_test.dart @@ -0,0 +1,131 @@ +import 'package:buzz/shared/agents/agent_launch.dart'; +import 'package:buzz/shared/agents/remote_agent_launch_provider.dart'; +import 'package:buzz/shared/agents/remote_agent_launch_status.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter/material.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; + +import 'launch_fixture.dart'; + +class _Launches extends RemoteAgentLaunches { + final LaunchFixture fixture; + _Launches(this.fixture); + @override + AsyncValue> build() { + queue = fixture.queue; + fixture.onEntries = (entries) => state = AsyncData(entries); + ref.onDispose(() => fixture.onEntries = null); + return AsyncData(fixture.entries); + } +} + +class _Notice extends RemoteAgentLaunchNotice { + final AgentLaunchNotice notice; + _Notice(this.notice); + @override + AgentLaunchNotice? build() => notice; +} + +void main() { + test( + 'queue storage separates full relay endpoints and preserves scheme aliases', + () { + String key(String endpoint) => remoteAgentLaunchStorageKey( + RelayConfig(baseUrl: endpoint, nsec: owner.nsec), + owner.public, + ); + expect( + key('https://relay.example/a?network=one'), + key('wss://relay.example/a?network=one'), + ); + expect( + key('https://relay.example/a?network=one'), + isNot(key('https://relay.example/b?network=one')), + ); + expect( + key('https://relay.example/a?network=one'), + isNot(key('https://relay.example/a?network=two')), + ); + }, + ); + + testWidgets('skipped startup has a dismissible notice in its own thread', ( + tester, + ) async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + final message = signed('33' * 32, 9, '@Alice', [ + ['h', channelId], + ['p', agent.public], + ]); + await tester.pumpWidget( + ProviderScope( + overrides: [ + remoteAgentLaunchesProvider.overrideWith(() => _Launches(fixture)), + remoteAgentLaunchNoticeProvider.overrideWith( + () => _Notice( + AgentLaunchNotice(message, 'Agent startup was skipped.'), + ), + ), + ], + child: MaterialApp( + home: Scaffold( + body: RemoteAgentLaunchStatus( + channelId: channelId, + threadRoot: message.id, + ), + ), + ), + ), + ); + expect(find.text('Agent startup was skipped.'), findsOneWidget); + await tester.tap(find.byTooltip('Dismiss agent startup notice')); + await tester.pumpAndSettle(); + expect(find.text('Agent startup was skipped.'), findsNothing); + expect(fixture.relay.published, isEmpty); + expect(tester.takeException(), isNull); + }); + + testWidgets( + 'a visible startup failure retries without resending the message', + (tester) async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.refusal = 'launch_not_registered'; + await tester.runAsync(() async { + await fixture.sender()( + channelId: channelId, + content: '@Bifrost', + mentionPubkeys: [agent.public], + ); + await fixture.settle(() => fixture.entries.single.error != null); + }); + await tester.pumpWidget( + ProviderScope( + overrides: [ + remoteAgentLaunchesProvider.overrideWith(() => _Launches(fixture)), + ], + child: const MaterialApp( + home: Scaffold(body: RemoteAgentLaunchStatus(channelId: channelId)), + ), + ), + ); + expect(find.textContaining('once from desktop'), findsOneWidget); + expect(find.byTooltip('Dismiss startup for Bifrost'), findsOneWidget); + fixture.relay.refusal = null; + await tester.runAsync(() async { + await tester.tap(find.text('Retry')); + await fixture.settle(() => fixture.entries.isEmpty); + }); + await tester.pumpAndSettle(); + expect(find.textContaining('once from desktop'), findsNothing); + expect(fixture.relay.requests, hasLength(2)); + expect( + fixture.relay.published.where((event) => event.kind == 9), + hasLength(1), + ); + expect(tester.takeException(), isNull); + }, + ); +} diff --git a/mobile/test/features/agents/remote_agent_start_client_test.dart b/mobile/test/features/agents/remote_agent_start_client_test.dart new file mode 100644 index 00000000000..03c753ba25d --- /dev/null +++ b/mobile/test/features/agents/remote_agent_start_client_test.dart @@ -0,0 +1,127 @@ +import 'dart:async'; + +import 'package:buzz/shared/agents/remote_agent_start_client.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; + +import 'launch_fixture.dart'; + +void main() { + NostrEvent trigger() => signed('33' * 32, 9, '@Bifrost investigate', [ + ['h', channelId], + ['p', agent.public], + ]); + + test( + 'a verified manager completion confirms startup even when relay OK is lost', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.controlAck = Completer(); + await fixture.client.start(agent.public, trigger()); + expect(fixture.relay.unsubscribed, 1); + fixture.relay.controlAck!.completeError( + StateError('late relay disconnect'), + ); + await Future.delayed(Duration.zero); + }, + ); + + test( + 'wrong signer, correlation and tampered replies cannot confirm a start', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.answer = false; + var complete = false; + final started = fixture.client + .start(agent.public, trigger()) + .then((_) => complete = true); + await fixture.settle(() => fixture.relay.requests.isNotEmpty); + final request = fixture.relay.requests.single; + final sent = fixture.relay.published.single; + fixture.relay.reply( + sent, + request, + 'done', + response: {'ok': true}, + signingKey: '44' * 32, + ); + await Future.delayed(Duration.zero); + expect(complete, false); + fixture.relay.reply( + sent, + request, + 'done', + response: {'ok': true}, + eventId: 'a' * 64, + ); + await Future.delayed(Duration.zero); + expect(complete, false); + fixture.relay.reply( + sent, + request, + 'done', + response: {'ok': true}, + requestId: 'another-request', + ); + await Future.delayed(Duration.zero); + expect(complete, false); + final listener = fixture.relay.listener!; + fixture.relay.listener = (event) { + final tampered = event.toJson()..['sig'] = '0' * 128; + listener(NostrEvent.fromJson(tampered)); + }; + fixture.relay.reply(sent, request, 'done', response: {'ok': true}); + await Future.delayed(Duration.zero); + expect(complete, false); + fixture.relay.listener = listener; + fixture.relay.reply(sent, request, 'done', response: {'ok': true}); + await started; + expect(complete, true); + expect(fixture.relay.unsubscribed, 1); + }, + ); + + test( + 'an unavailable manager produces a bounded actionable failure', + () async { + final relay = LaunchRelay()..answer = false; + final client = RemoteAgentStartClient( + session: relay, + config: RelayConfig(baseUrl: 'https://relay.example', nsec: owner.nsec), + manager: manager.public, + isCurrent: () => true, + acceptTimeout: const Duration(milliseconds: 30), + finishTimeout: const Duration(seconds: 1), + ); + addTearDown(client.dispose); + await expectLater( + client.start(agent.public, trigger()), + throwsA( + isA().having( + (e) => e.message, + 'message', + contains('did not answer'), + ), + ), + ); + expect(relay.unsubscribed, 1); + }, + ); + + test( + 'switching accounts cancels in-flight requests and releases subscriptions', + () async { + final fixture = LaunchFixture(); + addTearDown(fixture.dispose); + fixture.relay.answer = false; + final result = fixture.client.start(agent.public, trigger()); + final failure = expectLater(result, throwsA(isA())); + await fixture.settle(() => fixture.relay.requests.isNotEmpty); + fixture.dispose(); + await failure; + expect(fixture.relay.unsubscribed, 1); + }, + ); +}