diff --git a/.gitignore b/.gitignore index b3b51ab8d2..a7a0fe8ff9 100644 --- a/.gitignore +++ b/.gitignore @@ -39,3 +39,4 @@ dist-buzz-mcp/ # its bundled resources. src-tauri/resources/bin/* !src-tauri/resources/bin/.gitkeep +dist-agent-host/ diff --git a/biome.json b/biome.json index e3b00792d0..070d89fc44 100644 --- a/biome.json +++ b/biome.json @@ -15,6 +15,7 @@ "!!dist-author", "!!dist-plugins", "!!dist-buzz-mcp", + "!!dist-agent-host", "!!.profiles", "!!src-tauri/target", "!!src-tauri/gen", diff --git a/crates/agent-controller/src/app_agent.rs b/crates/agent-controller/src/app_agent.rs index ac6efd651b..a0fcdbadb4 100644 --- a/crates/agent-controller/src/app_agent.rs +++ b/crates/agent-controller/src/app_agent.rs @@ -205,6 +205,23 @@ impl AppAgent { } key.profile_auth(url, body) } + /// Its NIP-42 login to its own community, with its owner attestation. + pub fn login(&self, key: &Secret, relay: &str, challenge: &str) -> Result { + if relay != self.relay { + return Err("Agents sign in only to their community".into()); + } + if challenge.is_empty() || challenge.len() > 256 { + return Err("Invalid login challenge".into()); + } + let tags = self.attested( + key, + vec![ + vec!["relay".into(), relay.into()], + vec!["challenge".into(), challenge.into()], + ], + )?; + key.signed(22242, String::new(), tags) + } /// NIP-98 authorization for Git over HTTP to repository `repo` of `owner` /// in its community, and the URL it is bound to. Git reuses it for every /// request of one clone, fetch or push; the community honors it for a minute. diff --git a/crates/agent-controller/src/app_agent/tests.rs b/crates/agent-controller/src/app_agent/tests.rs index 6f5f188e7e..77d4887821 100644 --- a/crates/agent-controller/src/app_agent/tests.rs +++ b/crates/agent-controller/src/app_agent/tests.rs @@ -425,6 +425,34 @@ fn signs_git_auth_for_one_repository_in_its_own_community() { assert!(agent.git_auth(&other, &owner, "hive").is_err()); } +#[test] +fn signs_a_login_only_to_its_own_community() { + let dir = tempfile::tempdir().unwrap(); + let credentials = Memory::default(); + let (_, agent) = created(dir.path(), &credentials); + let key = agent.read_key(&credentials).unwrap(); + let saved: Vec = serde_json::from_str(&agent.auth).unwrap(); + let login = agent.login(&key, &agent.relay, "challenge").unwrap(); + assert_eq!(login["kind"], 22242); + assert_eq!(login["pubkey"], agent.pubkey); + let tags: Vec> = serde_json::from_value(login["tags"].clone()).unwrap(); + assert_eq!( + tags, + [ + vec!["relay".to_owned(), agent.relay.clone()], + vec!["challenge".to_owned(), "challenge".to_owned()], + saved + ] + ); + assert!(agent + .login(&key, "wss://elsewhere.test", "challenge") + .is_err()); + assert!(agent.login(&key, &agent.relay, "").is_err()); + assert!(agent.login(&key, &agent.relay, &"c".repeat(257)).is_err()); + let other = Secret::generate().unwrap(); + assert!(agent.login(&other, &agent.relay, "challenge").is_err()); +} + #[test] fn signs_uploads_and_memory_for_its_own_community_and_owner() { let dir = tempfile::tempdir().unwrap(); diff --git a/package.json b/package.json index afbfa03488..143bd74b4d 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "preview": "vite preview", "author:build": "node scripts/build-author.mjs", "buzz-mcp": "vite build --config vite.buzz-mcp.config.ts", + "agent-host": "vite build --config vite.agent-host.config.ts", "typecheck": "tsc --noEmit", "format": "biome format --write .", "format:check": "biome format .", diff --git a/src-tauri/src/agent_host.rs b/src-tauri/src/agent_host.rs new file mode 100644 index 0000000000..f1237d5e31 --- /dev/null +++ b/src-tauri/src/agent_host.rs @@ -0,0 +1,192 @@ +//! The app's custody of one agent without its window, for `buzz-agent-host`, +//! which runs the agent while the app is closed. It answers the requests the +//! WebView sends for its agents (`app_agents.rs`, `agent_state.rs`) with the +//! same code: one JSON line per request on stdin, one per answer on stdout. +//! Like a dev build, it runs the agent only while the installed app does not. +//! It exits when stdin closes, and the OS then lets go of its hosting locks. +use crate::agent_state::{self, AgentState}; +use crate::app_agents::{self, AppAgentHost}; +use serde::{de::DeserializeOwned, Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::io::{BufRead, Write}; +use std::sync::{Arc, Mutex, PoisonError}; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Request { + id: u64, + command: String, + #[serde(default)] + args: Value, +} + +/// Serves custody of agent `pubkey` of the app `identifier` until stdin closes. +pub(crate) fn serve(identifier: &str, pubkey: &str) -> i32 { + let Some(data) = dirs::data_dir() else { + eprintln!("Could not resolve local agent storage"); + return 1; + }; + let agents = data.join(identifier).join(crate::sign_out::AGENTS); + let host = Arc::new(AppAgentHost::new( + Ok(agents.join("identities.json")), + Err("Not available without the app".into()), + false, + )); + let state = Arc::new(AgentState::new(Ok(agents.join("state")))); + let out = Arc::new(Mutex::new(std::io::stdout())); + let pubkey: Arc = pubkey.into(); + for line in std::io::stdin().lock().lines() { + let Ok(line) = line else { break }; + let (host, state, out, pubkey) = (host.clone(), state.clone(), out.clone(), pubkey.clone()); + tauri::async_runtime::spawn(async move { + let (id, answer) = match serde_json::from_str::(&line) { + Ok(request) => ( + json!(request.id), + answer(&host, &state, &pubkey, &request.command, &request.args).await, + ), + Err(_) => (Value::Null, Err("Invalid request".into())), + }; + let line = match answer { + Ok(value) => json!({ "id": id, "ok": value }), + Err(error) => json!({ "id": id, "error": error }), + }; + let mut out = out.lock().unwrap_or_else(PoisonError::into_inner); + let _ = writeln!(out, "{line}"); + let _ = out.flush(); + }); + } + 0 +} + +fn arg(args: &Value, name: &str) -> Result { + serde_json::from_value(args.get(name).cloned().unwrap_or(Value::Null)) + .map_err(|_| format!("Invalid {name}")) +} + +fn reply(value: impl Serialize) -> Result { + serde_json::to_value(value).map_err(|_| "Invalid answer".into()) +} + +async fn answer( + host: &AppAgentHost, + state: &AgentState, + agent: &str, + command: &str, + args: &Value, +) -> Result { + // Every agent's name, for team instructions; anything else only for its own. + if command != "app_agent_list" + && !command.starts_with("app_agent_state_") + && arg::(args, "pubkey")? != agent + { + return Err("Not this host's agent".into()); + } + let pubkey = agent.to_owned(); + match command { + "app_agent_list" => reply(app_agents::list(host).await?), + "app_agent_claim" => reply(app_agents::claim(host, pubkey).await?), + "app_agent_release" => { + app_agents::release(host, &pubkey); + Ok(Value::Null) + } + "app_agent_publish" => app_agents::publish(host, pubkey, arg(args, "event")?).await, + "app_agent_query" => app_agents::query(host, pubkey, arg(args, "filters")?).await, + "app_agent_search_profiles" => reply( + app_agents::find_profiles(host, pubkey, arg(args, "query")?, arg(args, "limit")?) + .await?, + ), + "app_agent_upload" => { + app_agents::upload_blob(host, pubkey, arg(args, "data")?, arg(args, "mime")?).await + } + "app_agent_media" => reply(app_agents::media(host, pubkey, arg(args, "url")?).await?), + "app_agent_git_auth" => { + app_agents::git_auth(host, pubkey, arg(args, "owner")?, arg(args, "repo")?).await + } + "app_agent_remember" => { + app_agents::remember( + host, + pubkey, + arg(args, "slug")?, + arg(args, "body")?, + arg(args, "after")?, + ) + .await + } + "app_agent_state_load" => reply(state.load()?), + "app_agent_state_set" => reply(agent_state::save( + state, + arg(args, "key")?, + arg(args, "value")?, + arg(args, "absent")?, + )?), + // What the app does as the owner, this host does as the agent: it + // hears the agent's community and reads its memory with the agent's key. + "app_agent_login" => { + app_agents::login(host, pubkey, arg(args, "relay")?, arg(args, "challenge")?).await + } + "app_agent_read_auth" => { + app_agents::read_auth(host, pubkey, arg(args, "url")?, arg(args, "body")?).await + } + "app_agent_memories" => app_agents::memories(host, pubkey, arg(args, "slug")?).await, + _ => Err("Not available without the app".into()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn serves_only_its_agent_and_the_shared_state() { + let dir = tempfile::tempdir().unwrap(); + let host = AppAgentHost::new( + Ok(dir.path().join("identities.json")), + Err("unused".into()), + false, + ); + let state = AgentState::new(Ok(dir.path().join("state"))); + let agent = "a".repeat(64); + let ask = |command: &'static str, args: Value| { + let (host, state, agent) = (&host, &state, agent.clone()); + async move { answer(host, state, &agent, command, &args).await } + }; + let other = json!({ "pubkey": "b".repeat(64) }); + for command in ["app_agent_claim", "app_agent_publish", "app_agent_login"] { + assert_eq!( + ask(command, other.clone()).await, + Err("Not this host's agent".into()) + ); + } + assert_eq!( + ask("app_agent_create", json!({ "pubkey": agent })).await, + Err("Not available without the app".into()) + ); + assert_eq!(ask("app_agent_list", Value::Null).await, Ok(json!([]))); + // A missing agent is refused by the app's own custody. + assert!(ask( + "app_agent_publish", + json!({ "pubkey": agent, "event": { "kind": 9, "content": "" } }) + ) + .await + .is_err()); + assert_eq!( + ask("app_agent_claim", json!({ "pubkey": agent })).await, + Ok(json!(true)) + ); + ask("app_agent_state_set", json!({ "key": "k", "value": "v" })) + .await + .unwrap(); + assert_eq!( + ask( + "app_agent_state_set", + json!({ "key": "k", "value": "w", "absent": true }) + ) + .await, + Ok(json!("v")) + ); + assert_eq!( + ask("app_agent_state_load", Value::Null).await, + Ok(json!({ "k": "v" })) + ); + } +} diff --git a/src-tauri/src/agent_state.rs b/src-tauri/src/agent_state.rs index 43b5e4927c..e89618aaeb 100644 --- a/src-tauri/src/agent_state.rs +++ b/src-tauri/src/agent_state.rs @@ -222,6 +222,14 @@ pub(crate) async fn app_agent_state_set( key: String, value: Option, absent: Option, +) -> Result, String> { + save(&state, key, value, absent) +} +pub(crate) fn save( + state: &AgentState, + key: String, + value: Option, + absent: Option, ) -> Result, String> { match (value, absent.unwrap_or(false)) { (Some(value), true) => state.add(&key, &value), @@ -305,6 +313,11 @@ mod tests { ); assert_eq!(key.len(), 251); assert!(encode(&key).len() > NAME_LIMIT); + // The name the agent host's `fileName` gives it (storage.test.ts). + assert_eq!( + file_name(&key), + "~2418a80781b72ec15bca6c4423fa41f0937d29e13b19fe1247bf0daaa571773f" + ); for key in [key.as_str(), &":".repeat(86), &"k".repeat(NAME_LIMIT + 1)] { state.set(key, Some("{\"a\":1}")).unwrap(); let name = file_name(key); diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index 4cf291a9f1..839cabd04e 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -273,6 +273,9 @@ pub(crate) struct Ported { pub(crate) async fn app_agent_list( state: tauri::State<'_, AppAgentHost>, ) -> Result, String> { + list(&state).await +} +pub(crate) async fn list(state: &AppAgentHost) -> Result, String> { let agents = state.agents.clone()?; blocking(move || Ok(agents.list()?.into_iter().map(Into::into).collect())).await } @@ -467,6 +470,9 @@ pub(crate) async fn app_agent_claim( state: tauri::State<'_, AppAgentHost>, pubkey: String, ) -> Result { + claim(&state, pubkey).await +} +pub(crate) async fn claim(state: &AppAgentHost, pubkey: String) -> Result { let hosting = state.hosting.clone(); blocking(move || hosting.claim(&pubkey)).await } @@ -474,7 +480,10 @@ pub(crate) async fn app_agent_claim( /// Lets another copy of the app run the agent, as when this one cannot. #[tauri::command] pub(crate) fn app_agent_release(state: tauri::State<'_, AppAgentHost>, pubkey: String) { - state.hosting.held().remove(&pubkey); + release(&state, &pubkey); +} +pub(crate) fn release(state: &AppAgentHost, pubkey: &str) { + state.hosting.held().remove(pubkey); } #[derive(Deserialize)] @@ -492,6 +501,13 @@ pub(crate) async fn app_agent_publish( state: tauri::State<'_, AppAgentHost>, pubkey: String, event: AppAgentEvent, +) -> Result { + publish(&state, pubkey, event).await +} +pub(crate) async fn publish( + state: &AppAgentHost, + pubkey: String, + event: AppAgentEvent, ) -> Result { let (agent, key) = state.key(state.agent(pubkey).await?).await?; let signed = agent.sign(&key, event.kind, event.content, event.tags)?; @@ -539,6 +555,13 @@ pub(crate) async fn app_agent_query( state: tauri::State<'_, AppAgentHost>, pubkey: String, filters: Vec, +) -> Result { + query(&state, pubkey, filters).await +} +pub(crate) async fn query( + state: &AppAgentHost, + pubkey: String, + filters: Vec, ) -> Result { if filters.is_empty() || filters.len() > 8 { return Err("Agents read with one to eight filters".into()); @@ -594,6 +617,14 @@ pub(crate) async fn app_agent_search_profiles( pubkey: String, query: String, limit: u32, +) -> Result, String> { + find_profiles(&state, pubkey, query, limit).await +} +pub(crate) async fn find_profiles( + state: &AppAgentHost, + pubkey: String, + query: String, + limit: u32, ) -> Result, String> { let (agent, key) = state.key(state.agent(pubkey).await?).await?; search_profiles(&agent, &key, &query, limit).await @@ -650,6 +681,14 @@ pub(crate) async fn app_agent_upload( pubkey: String, data: String, mime: String, +) -> Result { + upload_blob(&state, pubkey, data, mime).await +} +pub(crate) async fn upload_blob( + state: &AppAgentHost, + pubkey: String, + data: String, + mime: String, ) -> Result { use base64::Engine; // The type is a hint: the community sniffs the bytes and decides what it @@ -714,6 +753,14 @@ pub(crate) async fn app_agent_git_auth( pubkey: String, owner: String, repo: String, +) -> Result { + git_auth(&state, pubkey, owner, repo).await +} +pub(crate) async fn git_auth( + state: &AppAgentHost, + pubkey: String, + owner: String, + repo: String, ) -> Result { use base64::Engine; let (agent, key) = state.key(state.agent(pubkey).await?).await?; @@ -758,6 +805,13 @@ pub(crate) async fn app_agent_media( state: tauri::State<'_, AppAgentHost>, pubkey: String, url: String, +) -> Result { + media(&state, pubkey, url).await +} +pub(crate) async fn media( + state: &AppAgentHost, + pubkey: String, + url: String, ) -> Result { use base64::Engine; let dir = std::env::temp_dir().join("buzz-agent-media"); @@ -862,12 +916,85 @@ pub(crate) async fn app_agent_remember( slug: String, body: String, after: u64, +) -> Result { + remember(&state, pubkey, slug, body, after).await +} +pub(crate) async fn remember( + state: &AppAgentHost, + pubkey: String, + slug: String, + body: String, + after: u64, ) -> Result { let (agent, key) = state.key(state.agent(pubkey).await?).await?; let signed = agent.memory(&key, &slug, &body, after)?; post(&agent, &key, signed).await } +/// The agent's login to its own community's live connection, for a host +/// that hears its community as the agent. +pub(crate) async fn login( + state: &AppAgentHost, + pubkey: String, + relay: String, + challenge: String, +) -> Result { + let (agent, key) = state.key(state.agent(pubkey).await?).await?; + agent.login(&key, &relay, &challenge) +} + +/// The headers that post `body` to the agent's community as the agent: a +/// read (`/query`) or a profile name search (`/profiles/search`) only, so the +/// caller can send it and read the whole response itself. +pub(crate) async fn read_auth( + state: &AppAgentHost, + pubkey: String, + url: String, + body: String, +) -> Result { + let (agent, key) = state.key(state.agent(pubkey).await?).await?; + if url != agent.query_url() && url != agent.profile_search_url() { + return Err("Agents read only from their community".into()); + } + Ok(serde_json::json!({ + "authorization": authorization(agent.http_auth(&key, &url, body.as_bytes())?)?, + "authTag": agent.auth, + })) +} + +/// The agent's memory, read as the agent: every entry it wrote for its owner, +/// or only entry `slug`. +pub(crate) async fn memories( + state: &AppAgentHost, + pubkey: String, + slug: Option, +) -> Result { + let (agent, key) = state.key(state.agent(pubkey).await?).await?; + let secret = zeroize::Zeroizing::new( + nostr::key::SecretKey::from_hex(&key.hex()) + .map_err(|_| "Invalid identity")? + .to_secret_bytes(), + ); + crate::relay::own_memories_read( + secret, + agent.pubkey.clone(), + agent.owner.clone(), + slug, + |body| { + let (agent, key) = (&agent, &key); + async move { + let (status, body) = + signed_read(agent, key, agent.query_url(), body.into_bytes()).await?; + Ok(( + status.as_u16(), + String::from_utf8(body).map_err(|_| "Memory read failed")?, + )) + } + }, + ) + .await +} + /// One client for every agent request, so frequent small posts such as /// typing pulses reuse a connection instead of opening a new one each time. /// Each request sets its own timeout. @@ -1226,6 +1353,56 @@ mod tests { served.await.unwrap(); } + #[tokio::test] + async fn a_host_reads_and_signs_in_as_its_agent_but_posts_nothing() { + let dir = tempfile::tempdir().unwrap(); + let key = Secret::generate().unwrap(); + let mut agent = agent_at("wss://relay.example".into(), &key); + agent.auth = r#"["auth","owner","","sig"]"#.into(); + std::fs::write( + dir.path().join("identities.json"), + serde_json::to_vec(&[&agent]).unwrap(), + ) + .unwrap(); + let host = AppAgentHost { + agents: Ok(AppAgents::open(dir.path().join("identities.json"))), + old: Err("unused".into()), + credentials: Arc::new(Busy(key.hex().to_string(), AtomicU32::new(0))), + keys: Arc::default(), + profiles: Arc::default(), + hosting: Arc::new(hosting(Err("unused".into()), false)), + }; + let pubkey = agent.pubkey.clone(); + for url in [agent.query_url(), agent.profile_search_url()] { + let auth = read_auth(&host, pubkey.clone(), url, "[]".into()) + .await + .unwrap(); + assert!(auth["authorization"] + .as_str() + .unwrap() + .starts_with("Nostr ")); + assert_eq!(auth["authTag"], agent.auth); + } + // Its writes go through custody, which checks each kind. + assert!( + read_auth(&host, pubkey.clone(), agent.events_url(), "{}".into()) + .await + .is_err() + ); + let signed = login( + &host, + pubkey.clone(), + agent.relay.clone(), + "challenge".into(), + ) + .await + .unwrap(); + assert_eq!(signed["kind"], 22242); + assert!(login(&host, pubkey, "wss://elsewhere".into(), "c".into()) + .await + .is_err()); + } + fn hosting(dir: Result, packaged: bool) -> Hosting { Hosting { dir, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 1f04427e09..b0fbda9cb6 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -14,6 +14,7 @@ use browser::{ browser_action, browser_attach, browser_detach, browser_navigate, browser_set_bounds, browser_status, }; +mod agent_host; mod agent_state; mod app_agents; mod decisions; @@ -628,6 +629,18 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen ] } #[cfg_attr(mobile, tauri::mobile_entry_point)] +/// Serves custody of one agent for `buzz-agent-host`; see `agent_host`. +pub fn agent_host(pubkey: Option) -> i32 { + let Some(pubkey) = pubkey.filter(|key| { + key.len() == 64 && key.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) + }) else { + eprintln!("Usage: agent-host "); + return 2; + }; + let identifier = app_context::().config().identifier.clone(); + agent_host::serve(&identifier, &pubkey) +} + pub fn run() { let context = app_context(); // Reject unsafe test launch settings before any identity or pending erase work. diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index caba166905..4762b64ce3 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -1,5 +1,10 @@ #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] fn main() { + // `buzz-agent-host` runs an agent with the app closed through this. + let mut args = std::env::args().skip(1); + if args.next().as_deref() == Some("agent-host") { + std::process::exit(buzz_foundation_lib::agent_host(args.next())); + } buzz_foundation_lib::run(); } diff --git a/src-tauri/src/relay.rs b/src-tauri/src/relay.rs index 6394118ad4..beeaad08b8 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -3,8 +3,8 @@ pub(crate) mod agent; mod huddle; use crate::identity::{EventTemplate, IdentityHost}; pub(crate) use agent::{ - relay_agent_library, relay_agent_log_proof, relay_agent_memories_read, relay_agent_observer, - relay_agent_resolve, + own_memories_read, relay_agent_library, relay_agent_log_proof, relay_agent_memories_read, + relay_agent_observer, relay_agent_resolve, }; use base64::{engine::general_purpose::STANDARD, Engine}; pub(crate) use huddle::{huddle_close, huddle_open, huddle_pcm, huddle_touch, Huddles}; diff --git a/src-tauri/src/relay/agent.rs b/src-tauri/src/relay/agent.rs index 0f1582ba3d..43a9ba777b 100644 --- a/src-tauri/src/relay/agent.rs +++ b/src-tauri/src/relay/agent.rs @@ -427,6 +427,40 @@ pub(crate) async fn relay_agent_memories_read( }) }) }; + memory_listing( + budget, + filter, + slug, + |body| { + let url = url.clone(); + let host = host.inner(); + async move { + let response = + send(host, url, "POST", Some(body), true, MEMORY_RESPONSE_BYTES).await?; + Ok((response.status, response.body)) + } + }, + decode, + ) + .await +} + +/// The memory listing `filter` selects, paged with `send` (one `/query` body +/// to the status and body it got) and decoded with `decode`; only entry `slug` +/// when given. +async fn memory_listing( + budget: MemoryBudget, + filter: Value, + slug: Option, + send: S, + decode: D, +) -> Result +where + S: Fn(String) -> F, + F: std::future::Future>, + D: Fn(Vec>, MemoryDecoding, std::time::Instant) -> G, + G: std::future::Future>, +{ let mut listing = read_memories( budget, |cursor| { @@ -435,31 +469,19 @@ pub(crate) async fn relay_agent_memories_read( page["until"] = json!(until); page["before_id"] = json!(before_id); } - let url = url.clone(); - let host = host.inner(); + let sent = send(json!([page]).to_string()); async move { - let response = send( - host, - url, - "POST", - Some(json!([page]).to_string()), - true, - MEMORY_RESPONSE_BYTES, - ) - .await?; - if response.status == 429 { + let (status, body) = sent.await?; + if status == 429 { return Ok(MemoryPage::Limited); } - if response.status == 401 || response.status == 403 { + if status == 401 || status == 403 { return Err("MemoryDenied".into()); } - if response.status != 200 || response.body.len() > MEMORY_RESPONSE_BYTES { + if status != 200 || body.len() > MEMORY_RESPONSE_BYTES { return Err("Memory read failed".into()); } - Ok(MemoryPage::Records( - memory_wire_records(&response.body)?, - response.body.len(), - )) + Ok(MemoryPage::Records(memory_wire_records(&body)?, body.len())) } }, decode, @@ -473,6 +495,63 @@ pub(crate) async fn relay_agent_memories_read( Ok(listing) } +/// The agent's own memory, read with its key as its owner reads it with +/// theirs: the entries it wrote for `owner`, paged with `send` as the agent. +pub(crate) async fn own_memories_read( + secret: zeroize::Zeroizing<[u8; 32]>, + agent: String, + owner: String, + slug: Option, + send: S, +) -> Result +where + S: Fn(String) -> F, + F: std::future::Future>, +{ + if !key(&agent) + || !key(&owner) + || agent == owner + || slug.as_deref().is_some_and(|name| !self::slug(name)) + { + return Err("Invalid memory target".into()); + } + let mut filter = json!({"kinds":[30174], "authors":[agent], "#p":[owner], "limit":MEMORY_PAGE}); + if let Some(name) = &slug { + let secret = NostrSecretKey::from_slice(&secret[..]).map_err(|_| "Invalid identity")?; + let peer = NostrPublicKey::from_hex(&owner).map_err(|_| "Invalid memory target")?; + filter["#d"] = json!([memory_address( + &ConversationKey::derive(&secret, &peer).map_err(|_| "Invalid memory target")?, + name, + )?]); + } + let secret = std::sync::Arc::new(secret); + memory_listing( + MemoryBudget::new(), + filter, + slug, + send, + |records, decoder, stop| { + let (secret, owner, agent) = (secret.clone(), owner.clone(), agent.clone()); + async move { + tokio::task::spawn_blocking(move || { + decode_memory_events( + &decoder, + &secret, + &owner, + &owner, + &agent, + &records, + || std::time::Instant::now() >= stop, + ) + }) + .await + .map_err(|_| "Memory read failed")? + } + }, + ) + .await +} + type MemoryDecoding = std::sync::Arc>; /// How long decoding a page fetched in time may run past the read's deadline; @@ -660,7 +739,17 @@ fn decode_memory_events( } let secret_key = NostrSecretKey::from_slice(secret).map_err(|_| "Invalid identity")?; - let public_key = NostrPublicKey::from_hex(agent).map_err(|_| "Invalid memory target")?; + // The owner decrypts with its key and the agent's; the agent itself with + // its key and the owner's. + let own = Keypair::from_secret_key( + &Secp256k1::signing_only(), + &secp256k1::SecretKey::from_byte_array(*secret).map_err(|_| "Invalid identity")?, + ) + .x_only_public_key() + .0 + .to_string(); + let peer = if own == agent { viewer } else { agent }; + let public_key = NostrPublicKey::from_hex(peer).map_err(|_| "Invalid memory target")?; let conversation = ConversationKey::derive(&secret_key, &public_key).map_err(|_| "Invalid memory target")?; let lock = || decoder.lock().unwrap_or_else(|e| e.into_inner()); @@ -1013,6 +1102,61 @@ mod tests { assert_ne!(agent, owner); } + #[tokio::test] + async fn an_agent_reads_its_own_memory_with_its_key() { + let agent_key = buzz_agent_controller::Secret::generate().unwrap(); + let agent = agent_key.pubkey().to_owned(); + let owner = signed_event(&[2; 32], 1, vec![], String::new()).pubkey; + let events = [("core", "my core"), ("mem/one", "my entry")] + .into_iter() + .map(|(slug, body)| { + agent_key + .memory_event(&owner, slug, body, 1_700_000_001) + .unwrap() + }) + .collect::>(); + let secret = || { + zeroize::Zeroizing::new( + NostrSecretKey::from_hex(&agent_key.hex()) + .unwrap() + .to_secret_bytes(), + ) + }; + let sent = std::sync::Mutex::new(Vec::::new()); + let send = |body: String| { + sent.lock() + .unwrap() + .push(serde_json::from_str(&body).unwrap()); + let page = json!(events).to_string(); + async move { Ok((200, page)) } + }; + let listing = own_memories_read(secret(), agent.clone(), owner.clone(), None, &send) + .await + .unwrap(); + assert_eq!(listing["entries"][0]["body"], "my core"); + assert_eq!(listing["entries"][1]["body"], "my entry"); + let filter = sent.lock().unwrap()[0][0].clone(); + assert_eq!(filter["authors"], json!([agent])); + assert_eq!(filter["#p"], json!([owner])); + // One entry is read by the address its owner would read it by. + let listing = own_memories_read( + secret(), + agent.clone(), + owner.clone(), + Some("mem/one".into()), + &send, + ) + .await + .unwrap(); + assert_eq!(listing["entries"].as_array().unwrap().len(), 1); + assert_eq!(sent.lock().unwrap()[1][0]["#d"][0], events[1]["tags"][0][1]); + assert!( + own_memories_read(secret(), agent.clone(), agent, None, &send) + .await + .is_err() + ); + } + #[test] fn encrypted_memory_and_observer_fixtures() { let (viewer_secret, viewer) = fixture(); diff --git a/src/agent-host/custody.test.ts b/src/agent-host/custody.test.ts new file mode 100644 index 0000000000..65ad4c00d3 --- /dev/null +++ b/src/agent-host/custody.test.ts @@ -0,0 +1,94 @@ +// @vitest-environment node +import type { ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { PassThrough } from "node:stream"; +import { expect, it, vi } from "vitest"; +import { appExecutable, startCustody } from "./custody"; + +/** The app's executable in `agent-host` mode, answered by `answer`. */ +function fakeApp( + answer: (request: { id: number; command: string; args?: unknown }) => unknown, +) { + const child = Object.assign(new EventEmitter(), { + stdin: new PassThrough(), + stdout: new PassThrough(), + }); + const started: string[][] = []; + createInterface({ input: child.stdin }).on("line", (line) => { + const request = JSON.parse(line) as { id: number; command: string }; + child.stdout.write(`${JSON.stringify(answer(request))}\n`); + }); + child.stdin.on("finish", () => child.emit("exit", 0, null)); + const start = (app: string, args: string[]) => { + started.push([app, ...args]); + return child as unknown as ChildProcess; + }; + return { child, start, started }; +} + +it("asks the app's executable for one agent's custody, one line per request", async () => { + const onExit = vi.fn(); + const app = fakeApp(({ id, command, args }) => + command === "app_agent_claim" + ? { id, ok: true } + : { id, error: `No ${command} ${JSON.stringify(args)}` }, + ); + const custody = startCustody("/app", "a".repeat(64), onExit, app.start); + expect(app.started).toEqual([["/app", "agent-host", "a".repeat(64)]]); + await expect( + custody.call("app_agent_claim", { pubkey: "a".repeat(64) }), + ).resolves.toBe(true); + await expect(custody.call("app_agent_create", { x: 1 })).rejects.toThrow( + 'No app_agent_create {"x":1}', + ); + await custody.close(); + // Closing it is not its stopping on its own. + expect(onExit).not.toHaveBeenCalled(); + await expect(custody.call("app_agent_claim")).rejects.toThrow("stopped"); +}); + +it("fails what is pending and tells the host when it stops on its own", async () => { + const onExit = vi.fn(); + const app = fakeApp(() => undefined); + app.child.stdout.write = () => true; + const custody = startCustody("/app", "a".repeat(64), onExit, app.start); + const pending = custody.call("app_agent_claim"); + app.child.emit("exit", null, "SIGKILL"); + await expect(pending).rejects.toThrow( + "The app's agent custody stopped (SIGKILL)", + ); + expect(onExit).toHaveBeenCalledWith( + "The app's agent custody stopped (SIGKILL)", + ); +}); + +it("runs only Buzz Desktop's executable", () => { + const dir = mkdtempSync(join(tmpdir(), "agent-host-app-")); + const bundle = (identifier: string) => { + const path = join(dir, identifier, "Buzz.app"); + mkdirSync(join(path, "Contents/MacOS"), { recursive: true }); + writeFileSync( + join(path, "Contents/Info.plist"), + `CFBundleExecutablebuzz-foundationCFBundleIdentifier${identifier}`, + ); + return path; + }; + const desktop = bundle("dev.local.buzz.foundation"); + expect(appExecutable({ BUZZ_APP: desktop })).toBe( + join(desktop, "Contents/MacOS/buzz-foundation"), + ); + expect(() => appExecutable({ BUZZ_APP: bundle("xyz.block.buzz") })).toThrow( + "is not Buzz Desktop", + ); + // An executable, as a dev build's, is taken as it is. + const executable = join(desktop, "Contents/MacOS/buzz-foundation"); + writeFileSync(executable, ""); + expect(appExecutable({ BUZZ_APP: executable })).toBe(executable); + expect(() => appExecutable({ BUZZ_APP: join(dir, "missing") })).toThrow( + "Buzz is not installed", + ); +}); diff --git a/src/agent-host/custody.ts b/src/agent-host/custody.ts new file mode 100644 index 0000000000..e0a7976dab --- /dev/null +++ b/src/agent-host/custody.ts @@ -0,0 +1,105 @@ +// The app's custody of the agent, served by the installed app's own executable +// (` agent-host `, src-tauri agent_host.rs) without its window. +// The agent's key, its saved identity, its hosting locks and its storage stay +// with the app's code; this host asks for one signed request, read or write +// at a time, with the commands the app's WebView uses (`agentsNative`). +import { type ChildProcess, spawn } from "node:child_process"; +import { existsSync, readFileSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import type { NativeCall } from "../features/agents/native"; + +/** The app's identifier: installed and dev builds share it, and its data. */ +const APP = "dev.local.buzz.foundation"; + +/** The app's executable: `BUZZ_APP`, a bundle or an executable, or else the + * installed app. A bundle must be this app, not another with its name. */ +export function appExecutable(env = process.env) { + const path = env.BUZZ_APP ?? "/Applications/Buzz.app"; + if (!existsSync(path)) throw new Error(`Buzz is not installed at ${path}`); + if (!statSync(path).isDirectory()) return path; + const plist = readFileSync(join(path, "Contents/Info.plist"), "utf8"); + const value = (key: string) => + new RegExp(`${key}\\s*([^<]*)`).exec( + plist, + )?.[1]; + if (value("CFBundleIdentifier") !== APP) + throw new Error(`${path} is not Buzz Desktop`); + const executable = value("CFBundleExecutable"); + if (!executable) throw new Error(`${path} has no executable`); + return join(path, "Contents/MacOS", executable); +} + +export type Custody = Readonly<{ + call: NativeCall; + /** Ends custody; the app's code lets go of the agent's locks as it exits. */ + close(): Promise; +}>; + +/** Starts the app's custody of agent `pubkey`. `onExit` hears it stop on its + * own: the agent is then no longer held, so it must stop running. */ +export function startCustody( + app: string, + pubkey: string, + onExit: (reason: string) => void, + start: (app: string, args: string[]) => ChildProcess = (app, args) => + // Its own process group: a Ctrl-C stops the host, which then stops it. + spawn(app, args, { stdio: ["pipe", "pipe", "inherit"], detached: true }), +): Custody { + const child = start(app, ["agent-host", pubkey]); + const pending = new Map< + number, + { resolve(value: unknown): void; reject(error: Error): void } + >(); + let next = 0; + let ended: string | undefined; + let closing = false; + const end = (reason: string) => { + if (ended) return; + ended = reason; + for (const { reject } of pending.values()) reject(new Error(reason)); + pending.clear(); + if (!closing) onExit(reason); + }; + const exited = new Promise((resolve) => { + child.once("exit", (code, signal) => { + end(`The app's agent custody stopped (${signal ?? code})`); + resolve(); + }); + child.once("error", (error) => { + end(`The app's agent custody could not run: ${error.message}`); + resolve(); + }); + }); + if (child.stdout) + createInterface({ input: child.stdout }).on("line", (line) => { + let answer: { id?: unknown; ok?: unknown; error?: unknown }; + try { + answer = JSON.parse(line) as typeof answer; + } catch { + return; + } + const waiting = + typeof answer.id === "number" ? pending.get(answer.id) : undefined; + if (!waiting) return; + pending.delete(answer.id as number); + if ("error" in answer) waiting.reject(new Error(String(answer.error))); + else waiting.resolve(answer.ok); + }); + const call: NativeCall = (command, args) => { + if (ended) return Promise.reject(new Error(ended)); + const id = next++; + return new Promise((resolve, reject) => { + pending.set(id, { resolve: resolve as (value: unknown) => void, reject }); + child.stdin?.write(`${JSON.stringify({ id, command, args })}\n`); + }); + }; + return Object.freeze({ + call, + close: async () => { + closing = true; + child.stdin?.end(); + await exited; + }, + }); +} diff --git a/src/agent-host/host.test.ts b/src/agent-host/host.test.ts new file mode 100644 index 0000000000..9e5b705abb --- /dev/null +++ b/src/agent-host/host.test.ts @@ -0,0 +1,61 @@ +// @vitest-environment node +import { expect, it } from "vitest"; +import { DEFAULT_CONFIG } from "../bundled/claude-code/runtime"; +import { defaults as codexDefaults } from "../bundled/codex/config"; +import type { AgentIdentity } from "../features/agents/native"; +import { fileStore } from "../features/agents/state"; +import { changed, claudeResolver, codexResolver } from "./host"; + +it("resolves blank settings from the owner's saved defaults, as they are now", async () => { + const storage = fileStore(new Map(), async () => {}); + const claude = claudeResolver(storage); + const codex = codexResolver(storage); + // Without saved defaults, the CLI's own. + expect(claude(DEFAULT_CONFIG).model).toBe(""); + expect(codex(codexDefaults)).toMatchObject({ model: "", effort: "" }); + // The app saves defaults while the host runs. + await storage.set( + "buzz.claude-code.provider-defaults.v1", + JSON.stringify({ version: 1, model: "sonnet" }), + ); + await storage.set( + "buzz.codex.provider-defaults.v1", + JSON.stringify({ version: 1, model: "gpt-5", effort: "high" }), + ); + expect(claude(DEFAULT_CONFIG).model).toBe("sonnet"); + expect(codex(codexDefaults)).toMatchObject({ + model: "gpt-5", + effort: "high", + }); + // An agent's own choice wins. + expect(claude({ ...DEFAULT_CONFIG, model: "opus" }).model).toBe("opus"); + expect(codex({ ...codexDefaults, model: "o3", effort: "low" })).toMatchObject( + { model: "o3", effort: "low" }, + ); +}); + +it("stops when the app deletes, erases or changes the agent", () => { + const agent: AgentIdentity = { + pubkey: "a".repeat(64), + relay: "wss://relay.example.test", + owner: "b".repeat(64), + type: "buzz.claude-code/claude-code", + name: "Ada", + deleted: false, + }; + expect(changed(agent, [{ ...agent, name: "Renamed" }])).toBeUndefined(); + for (const saved of [ + [], + [{ ...agent, deleted: true }], + [{ ...agent, deleting: true as const }], + ]) + expect(changed(agent, saved)).toBe( + "The agent is no longer saved on this machine", + ); + expect(changed(agent, [{ ...agent, type: "buzz.codex/codex" }])).toBe( + "The agent's type changed on this machine", + ); + expect(changed(agent, [{ ...agent, owner: "c".repeat(64) }])).toBe( + "The agent's owner changed on this machine", + ); +}); diff --git a/src/agent-host/host.ts b/src/agent-host/host.ts new file mode 100644 index 0000000000..e3113bc454 --- /dev/null +++ b/src/agent-host/host.ts @@ -0,0 +1,545 @@ +// A standalone agent host: runs one of the agents the desktop app saved on +// this machine, with the same runner and agent type code the app runs, while +// the app does not run it. The app's own claim protocol decides which copy +// runs the agent, so opening the installed app takes it over, and quitting +// the app lets this host take it back. +import claudeManifest from "../bundled/claude-code/manifest.json"; +import { claudeRun } from "../bundled/claude-code/run"; +import { withScript as claudeScript } from "../bundled/claude-code/setup"; +import { + readDefaults as claudeSaved, + resolveConfig as claudeConfig, +} from "../bundled/claude-code/defaults"; +import { ClaudeRuntime, DEFAULT_CONFIG } from "../bundled/claude-code/runtime"; +import { defaults as codexDefaults } from "../bundled/codex/config"; +import { + readDefaults as codexSaved, + resolveConfig as codexConfig, +} from "../bundled/codex/defaults"; +import codexManifest from "../bundled/codex/manifest.json"; +import { codexRun } from "../bundled/codex/run"; +import { withScript as codexScript } from "../bundled/codex/setup"; +import { CodexRuntime } from "../bundled/codex/runtime"; +import { CoreMemoryCache } from "../features/agents/core-memory-cache"; +import { createLocalAgentActivity } from "../features/agents/local-activity"; +import { + type AgentReads, + type MemoryListing, + ownerReads, +} from "../features/agents/reads"; +import { RecordChanges } from "../features/agents/record-changes"; +import { rosterDm, rosterMember } from "../features/agents/roster"; +import { soloChannel } from "../features/agents/solo-chat"; +import { communityDestination } from "../features/communities/destination"; +import type { Checkpoint } from "../features/agents/catch-up"; +import type { Interrupted } from "../features/agents/interrupted"; +import { + type AgentIdentity, + agentsNative, + type NativeCall, +} from "../features/agents/native"; +import { AgentRunners } from "../features/agents/runner"; +import { + type Agent, + type RegisteredAgentType, + teamPrompt, +} from "../features/agents/service"; +import { + type AgentRecord, + type AgentStore, + readOne, + readTeams, + recordKey, + setAttention, + writeRecord, +} from "../features/agents/store"; +import { fileStore, nativeFiles } from "../features/agents/state"; +import { relayPartition } from "../features/relay/partition"; +import { type AgentSession, connectAgentSession } from "./session"; +import { type Declared, processHost } from "./spawn"; + +/** How often due timers are looked for, as in the app. */ +const TICK_MS = 5_000; + +/** An agent's `core` memory, as `Agents.core` gives it. */ +type Core = ( + pubkey: string, + signal?: AbortSignal, +) => Promise; + +type TypeRun = Readonly<{ + type: RegisteredAgentType; + sync(agents: readonly Agent[]): void; + /** Settles once every process the type started has exited. */ + dispose(): Promise; +}>; + +export type HostOptions = Readonly<{ + identity: AgentIdentity; + /** The app's custody of the agent (`custody.ts`): its key, the agents the + * app saved, the hosting locks, and the agent storage the app shares. */ + call: NativeCall; + log(message: string): void; + /** Called once the agent is deleted, erased or changed in the app: this + * host has stopped hosting it, and must stop. */ + onGone(reason: string): void; +}>; + +/** Agent storage as the app keeps it, read again at each check, since the app + * may save while this host runs. A read that a write of this host's own + * overlaps is dropped, so it cannot put back what the write replaced. */ +async function agentFiles(call: NativeCall) { + const files = nativeFiles(call); + const items = new Map(Object.entries(await files.load())); + const store = fileStore(items, files.set, files.add); + let writes = 0; + let pending = 0; + const counted = (write: Promise) => { + writes++; + pending++; + return write.finally(() => pending--); + }; + const storage: AgentStore = Object.freeze({ + ...store, + set: (key: string, value: string | null) => counted(store.set(key, value)), + add: (key: string, value: string) => counted(store.add(key, value)), + }); + const reload = async () => { + const before = writes; + if (pending) return; + const saved = await files.load(); + if (pending || writes !== before) return; + items.clear(); + for (const [key, value] of Object.entries(saved)) items.set(key, value); + }; + return { storage, reload }; +} + +/** Why this host must stop running `identity`, by the agents the app saved + * here now: it was deleted, an erase removed it, or the app changed it. */ +export function changed( + identity: AgentIdentity, + saved: readonly AgentIdentity[], +): string | undefined { + const now = saved.find((agent) => agent.pubkey === identity.pubkey); + if (!now || now.deleted || now.deleting) + return "The agent is no longer saved on this machine"; + for (const field of ["type", "owner", "relay"] as const) + if (now[field] !== identity[field]) + return `The agent's ${field} changed on this machine`; + return undefined; +} + +/** Connects as the agent, then runs it until the returned stop is called. */ +export async function host(options: HostOptions): Promise<() => Promise> { + const { identity, call, log, onGone } = options; + const { pubkey } = identity; + const native = agentsNative(call); + // The app's agent storage: the agent's settings, teams and sessions are the + // ones the app shows and saves. + const files = await agentFiles(call); + const { storage } = files; + // As the app scopes the agent's community, so saved sessions match. + const scope = relayPartition( + communityDestination(identity.relay).url, + identity.owner, + ); + // The app's own relay session, as the agent: it hears the agent's channels + // and answers its reads, as the owner's session does in the app. + // Its live connection reconnects by itself; only the first connection, + // which reads the community's information, is tried again here. + let connected: AgentSession | undefined; + for (let attempt = 0; !connected; attempt++) + try { + connected = await connectAgentSession({ + relay: identity.relay, + pubkey, + call, + scope, + }); + } catch (error) { + const wait = Math.min(30_000, 1_000 * 2 ** attempt); + log( + `Could not connect (${String(error)}); trying again in ${wait / 1000}s`, + ); + await new Promise((resolve) => setTimeout(resolve, wait)); + } + const connection = connected; + const { session } = connection; + const reads: AgentReads = { + ...(ownerReads(connection.relay, pubkey) as AgentReads), + // Memory is encrypted to the owner; the app's custody reads it as the + // agent, with the agent's key, as the owner's session reads it in the app. + memoryListing: async (signal, slug) => { + signal.throwIfAborted(); + const listing = await call("app_agent_memories", { + pubkey, + slug, + }); + signal.throwIfAborted(); + return listing; + }, + // The channel's Solo label decides, as for every member and agent. + soloChat: async (_owner, channel, signal) => { + signal.throwIfAborted(); + return soloChannel(session, channel); + }, + }; + let live = false; + const readsOf = (agent: string) => + live && agent === pubkey ? reads : undefined; + + // The agent's core memory, read through its own reads, as the app keeps it. + const coreMemory = new CoreMemoryCache(readsOf); + const core: Core = (agent, signal) => coreMemory.core(agent, signal); + + const [pluginId = "", typeId = ""] = identity.type.split("/"); + const run = typeRun(typeId, pluginId, identity.type, readsOf, core, storage); + const activity = createLocalAgentActivity(); + + // The agent's record as last read, and the stored text it was read from: + // the app may save a change at any time, and each tick reads it again. + let saved = storage.get(recordKey(pubkey)); + let stored = readOne(storage, pubkey); + let view: Agent | undefined; + const recordOf = (): AgentRecord => { + if (!stored) { + const made = run.type.defaults(); + let record: AgentRecord = { + pubkey, + attention: {}, + config: made.config, + }; + for (const [slug, value] of Object.entries(made.attention ?? {})) + record = setAttention(record, slug, value); + stored = record; + } + return stored; + }; + const reread = () => { + const now = storage.get(recordKey(pubkey)); + if (now === saved) return; + saved = now; + stored = readOne(storage, pubkey); + view = undefined; + run.sync([agentView()]); + }; + // One object while nothing changes: the runner compares it to see edits. + const agent = (): Agent => { + if (view) return view; + const record = recordOf(); + view = Object.freeze({ + pubkey, + name: identity.name, + type: identity.type, + owner: identity.owner, + relay: identity.relay, + place: "here", + attention: record.attention, + skipped: record.skipped ?? {}, + timers: record.timers ?? {}, + attentionEnabled: record.attentionEnabled === true, + config: record.config, + }); + return view; + }; + /** Shows the agent's newest record to its runner. */ + const show = () => { + view = undefined; + run.sync([agentView()]); + }; + // As the app's agents service: the record's changes run in order, each + // from the record the one before it kept, and its timers wait for them. + const changes = new RecordChanges({ + saved: (agent) => (agent === pubkey ? recordOf() : undefined), + commit: async (agent, _, next) => { + mine(agent); + if (!next) throw new Error("This host cannot remove its agent"); + await writeRecord(storage, next); + saved = storage.get(recordKey(pubkey)); + stored = next; + }, + keep: (agent, next) => { + mine(agent); + stored = next; + }, + changed: show, + settled: () => runners.wake(pubkey), + }); + // Whether the app deleted, erased or changed the agent: then this host + // stops, and tells its caller why. + let gone = false; + // Every agent the app saved here, named in team instructions. + let names: readonly AgentIdentity[] = []; + const isGone = async () => { + let reason: string | undefined; + try { + names = await native.list(); + reason = changed(identity, names); + } catch (error) { + log(`Could not read the saved agent: ${String(error)}`); + } + if (reason && !gone) { + gone = true; + // After this function has set the host up, if it is still doing so. + queueMicrotask(() => void stop().finally(() => onGone(reason))); + } + return gone; + }; + const mine = (agent: string) => { + if (agent !== pubkey) throw new Error("Not this host's agent"); + }; + const started = Math.floor(Date.now() / 1000); + // Catch-up progress, kept only while this host runs. This host does not + // catch up yet, so it neither reads nor writes the agent's saved progress + // (its own key in agent storage): as before catch-up, it hears what happens + // while it is connected. + let checkpoints: Readonly> = {}; + let interrupted: Interrupted = {}; + const runners = new AgentRunners({ + ready: (agent) => (live && agent === pubkey ? agentView() : undefined), + agent: (agent) => (agent === pubkey ? agentView() : undefined), + member: (_, event) => rosterMember(session, pubkey, event), + dm: (_, event) => rosterDm(session, pubkey, event), + // No catch-up: see `checkpoints`. + community: () => undefined, + records: () => [recordOf()], + checkpoints: { + read: () => checkpoints, + write: (next) => { + checkpoints = next; + }, + }, + // Turns cut off by a stop, kept as checkpoints are: this host does not + // resume them when it starts again. + interrupted: { + read: () => interrupted, + write: (agent, turns) => { + interrupted = { ...interrupted, [agent]: turns }; + }, + changed: () => {}, + }, + scope: () => scope, + saved: (agent) => (agent === pubkey ? stored : undefined), + record: (agent) => { + mine(agent); + return recordOf(); + }, + change: async (agent, apply) => { + mine(agent); + const record = await changes.change(agent, (saved) => + apply(saved ?? recordOf()), + ); + if (!record) throw new Error("No such agent on this host"); + return record; + }, + changing: (agent) => changes.changing(agent), + timers: (agent) => changes.timers(agent), + progress: (agent, ran) => changes.progress(agent, ran), + // Each run starts from its saved record, as the app's agents do while the + // app binds no synced config to them: the synced copies are in the app's + // web storage, which this host cannot read. + start: (agent) => ({ ok: true, config: agent.config, revisions: [] }), + team: (agent) => + teamPrompt( + readTeams(storage), + names.filter((saved) => !saved.deleted), + agent, + ), + // Channel templates live in the app's web storage, which this host + // cannot read; its agent offers none. + templates: () => [], + // Thread tabs are the owner's, in the app. + tabs: () => undefined, + // The classifier key is in the app's Keychain entry: no event watch here + // is checked by a classifier, as in an app without one. + classifier: () => "unavailable", + decisions: () => undefined, + keyStatus: () => undefined, + classifierKey: () => "missing", + classifierKeyGeneration: () => 0, + setClassifierKey: () => {}, + context: async () => { + throw new Error("This host runs no classifier"); + }, + displayName: (_, person) => reads.name(person) ?? null, + custody: { + claim: async (agent) => { + mine(agent); + // Checked here too, not only each tick, so an event cannot start a + // turn for an agent the app has deleted. + if (gone || (await isGone())) return false; + try { + return await native.claim(agent); + } catch (error) { + // Unlike the app, which always runs its agents, this host runs one + // only when it knows no other copy does. + log(`Could not check who runs the agent: ${String(error)}`); + return false; + } + }, + release: native.release, + publish: native.publish, + query: native.query, + searchProfiles: native.searchProfiles, + upload: native.upload, + media: native.media, + gitAuth: native.gitAuth, + remember: native.remember, + }, + activity, + // Typing goes where the turn answers, read the same way. + soloChat: (agent, channel, signal) => + reads.soloChat(agent.owner, channel, signal), + remembered: (agent, slug, body) => coreMemory.wrote(agent, slug, body), + finished: (agent) => void coreMemory.refresh(agent), + // Naming a Solo chat is the owner's signed write; this host holds only + // the agent's key, so it leaves the name as it is. + named: () => {}, + }); + function agentView() { + return agent(); + } + + runners.reconcile( + [{ pubkey, type: identity.type }], + new Map([[identity.type, run.type]]), + ); + run.sync([agentView()]); + // As the app's agents hear its session (`AgentsService.listen`): a replay + // (the recent history a subscription sends again before it goes live) is + // heard only for what happened since this host started. + const stopEvents = session.subscribeLive( + (batch) => runners.dispatch(batch, () => true, started), + { replay: true }, + ); + const onStatus = () => { + const now = session.live.snapshot().status === "connected"; + if (now === live) return; + live = now; + log(now ? "Listening" : "Not listening"); + // Its core memory may have changed elsewhere while it was not connected. + if (now) void coreMemory.refresh(pubkey); + runners.resume(new Set(now ? [pubkey] : [])); + }; + const stopStatus = session.live.subscribe(onStatus); + onStatus(); + // A roster the relay confirms decides the events held for it. + const stopRosters = session.channels.subscribeList?.(() => + runners.releaseHeld(() => true), + ); + let ticking = false; + const tick = setInterval(async () => { + if (ticking) return; + ticking = true; + try { + if (gone || (await isGone())) return; + await files.reload(); + reread(); + } catch (error) { + log(`Could not read the agent's settings: ${String(error)}`); + } finally { + ticking = false; + } + if (!gone) runners.tick(); + }, TICK_MS); + log(`Running ${identity.name} (${identity.type})`); + + let stopped: Promise | undefined; + const stop = () => (stopped ??= stopHosting()); + return stop; + + async function stopHosting() { + clearInterval(tick); + stopEvents(); + stopStatus(); + stopRosters?.(); + connection.relay.disconnect(); + runners.clear(); + // Only once its processes have exited may another copy run the agent. + await run.dispose(); + await native.release(pubkey); + } +} + +/** Claude settings as the app resolves them: blank ones inherit the owner's + * defaults, read as they are saved now, since the app may change them. */ +export const claudeResolver = (storage: AgentStore) => (raw: unknown) => + claudeConfig(raw, claudeSaved(storage)); +/** Codex settings as the app resolves them, without the app's model list: a + * default model Codex no longer lists is left to Codex to report. */ +export const codexResolver = (storage: AgentStore) => (raw: unknown) => + codexConfig(raw, codexSaved(storage)); + +function typeRun( + typeId: string, + pluginId: string, + key: string, + reads: (pubkey: string) => AgentReads | undefined, + core: Core, + storage: AgentStore, +): TypeRun { + const registered = ( + type: Pick & + Partial, + ) => + Object.freeze({ + ...type, + key, + pluginId, + revision: "standalone", + }) as RegisteredAgentType; + if (typeId === "claude-code") { + const processes = processHost( + claudeManifest.host.processes as readonly Declared[], + ); + // Runs Buzz's own copy of Claude, which the desktop app keeps current. + const runtime = new ClaudeRuntime( + { spawn: claudeScript(processes.spawn) }, + reads, + core, + storage, + claudeResolver(storage), + ); + return { + type: registered({ + id: typeId, + title: "Claude", + defaults: () => ({ config: DEFAULT_CONFIG }), + ...claudeRun(runtime), + }), + sync: (agents) => runtime.sync(agents), + dispose: () => { + runtime.dispose(); + return processes.stopAll(); + }, + }; + } + if (typeId === "codex") { + const processes = processHost( + codexManifest.host.processes as readonly Declared[], + ); + // Runs Buzz's own copy of Codex, which the desktop app keeps current. + const runtime = new CodexRuntime( + codexScript(processes.spawn), + reads, + core, + storage, + codexResolver(storage), + ); + return { + type: registered({ + id: typeId, + title: "Codex", + defaults: () => ({ config: { ...codexDefaults } }), + ...codexRun(runtime), + }), + sync: (agents) => runtime.sync(agents), + dispose: () => { + runtime.dispose(); + return processes.stopAll(); + }, + }; + } + throw new Error(`This host cannot run ${key} agents`); +} diff --git a/src/agent-host/main.ts b/src/agent-host/main.ts new file mode 100644 index 0000000000..38d6934fc7 --- /dev/null +++ b/src/agent-host/main.ts @@ -0,0 +1,62 @@ +// buzz-agent-host: runs one of this machine's Buzz agents without the desktop app. +// +// buzz-agent-host run +// +// The agent is one the app saved here, and it runs with the settings, teams +// and sessions the app saved for it. The app's own executable holds its key +// and its claim on the agent (`custody.ts`): BUZZ_APP names it, or else it is +// the installed app. While the installed app runs the agent, this host waits. +import { agentsNative } from "../features/agents/native"; +import { appExecutable, startCustody } from "./custody"; +import { host } from "./host"; + +const usage = "Usage: buzz-agent-host run "; +const log = (message: string) => + console.error(`${new Date().toISOString()} ${message}`); + +async function main(args: readonly string[]) { + const [command, pubkey, ...rest] = args; + if ( + command !== "run" || + !pubkey || + !/^[0-9a-f]{64}$/.test(pubkey) || + rest.length + ) { + console.error(usage); + process.exit(2); + } + const custody = startCustody(appExecutable(), pubkey, (reason) => { + log(`${reason}; stopped`); + process.exit(1); + }); + const identity = (await agentsNative(custody.call).list()).find( + (agent) => agent.pubkey === pubkey && !agent.deleted, + ); + if (!identity) throw new Error(`No agent ${pubkey} is saved on this machine`); + let stopping = false; + let stop: (() => Promise) | undefined; + const quit = () => { + if (stopping || !stop) process.exit(stop ? 1 : 0); + stopping = true; + log("Stopping"); + void stop() + .then(() => custody.close()) + .finally(() => process.exit(0)); + }; + process.on("SIGINT", quit); + process.on("SIGTERM", quit); + stop = await host({ + identity, + call: custody.call, + log, + onGone: (reason) => { + log(`${reason}; stopped`); + void custody.close().finally(() => process.exit(0)); + }, + }); +} + +main(process.argv.slice(2)).catch((error: unknown) => { + console.error(error instanceof Error ? error.message : error); + process.exit(1); +}); diff --git a/src/agent-host/session.test.ts b/src/agent-host/session.test.ts new file mode 100644 index 0000000000..8c593aaea1 --- /dev/null +++ b/src/agent-host/session.test.ts @@ -0,0 +1,129 @@ +// @vitest-environment node +import { + finalizeEvent, + generateSecretKey, + getPublicKey, + verifyEvent, +} from "nostr-tools/pure"; +import { afterEach, expect, it, vi } from "vitest"; +import type { NativeCall } from "../features/agents/native"; +import { connectAgentSession } from "./session"; + +/** A relay connection the test answers by hand. */ +class Socket { + static last: Socket | undefined; + readyState = 1; + onmessage?: (event: { data: string }) => Promise; + onclose?: () => void; + onerror?: () => void; + readonly sent: unknown[][] = []; + constructor() { + Socket.last = this; + } + send(text: string) { + this.sent.push(JSON.parse(text) as unknown[]); + } + close() { + this.readyState = 3; + } + async receive(message: unknown[]) { + await this.onmessage?.({ data: JSON.stringify(message) }); + } +} + +afterEach(() => vi.unstubAllGlobals()); + +it("signs in and reads through the app's custody, and publishes nothing", async () => { + const secret = generateSecretKey(); + const pubkey = getPublicKey(secret); + const auth = ["auth", "b".repeat(64), "", "f".repeat(128)]; + const posted: { path: string; headers: Headers }[] = []; + vi.stubGlobal("WebSocket", Socket); + vi.stubGlobal( + "fetch", + vi.fn(async (url: string, init?: RequestInit) => { + if (init?.method !== "POST") + return Response.json({ self: "c".repeat(64) }); + posted.push({ + path: new URL(url).pathname, + headers: new Headers(init.headers), + }); + return Response.json([]); + }), + ); + const relay = "wss://relay.example.test"; + // The app's custody, as `agent_host.rs` answers: it signs with the key. + const calls: [string, Record | undefined][] = []; + const call = (async (command: string, args?: Record) => { + calls.push([command, args]); + if (command === "app_agent_login") + return finalizeEvent( + { + kind: 22242, + content: "", + created_at: Math.floor(Date.now() / 1000), + tags: [ + ["relay", String(args?.relay)], + ["challenge", String(args?.challenge)], + auth, + ], + }, + secret, + ); + if (command === "app_agent_read_auth") + return { authorization: "Nostr signed", authTag: JSON.stringify(auth) }; + throw new Error(`Unexpected ${command}`); + }) as NativeCall; + const { session, relay: data } = await connectAgentSession({ + relay, + pubkey, + call, + scope: "scope", + }); + try { + const socket = Socket.last as Socket; + await socket.receive(["AUTH", "challenge"]); + const login = socket.sent.find((message) => message[0] === "AUTH")?.[1] as + | { id: string; pubkey: string; tags: string[][] } + | undefined; + expect(login?.pubkey).toBe(pubkey); + expect(verifyEvent(login as never)).toBe(true); + expect(login?.tags).toContainEqual(auth); + expect(calls[0]).toEqual([ + "app_agent_login", + { pubkey, relay, challenge: "challenge" }, + ]); + await socket.receive(["OK", login?.id, true, ""]); + await vi.waitFor(() => + expect(session.live.snapshot().status).toBe("connected"), + ); + expect(session.workflowAuthority).toBe("c".repeat(64)); + // The session has no writer, and its socket no publishing. + expect( + await session.publishAgentPresence( + "online", + [pubkey], + new AbortController().signal, + ), + ).toBe(false); + expect(socket.sent.filter((message) => message[0] === "EVENT")).toEqual([]); + expect(posted.map(({ path }) => path)).not.toContain("/events"); + // Its reads carry the headers custody signed for each one. + await session.read([{ kinds: [9], "#h": ["c"], limit: 1 }], { + signal: new AbortController().signal, + }); + const reads = calls.filter( + ([command]) => command === "app_agent_read_auth", + ); + expect(reads.length).toBeGreaterThan(0); + for (const [, args] of reads) + expect(args?.url).toBe("https://relay.example.test/query"); + expect(posted.length).toBe(reads.length); + for (const { headers } of posted) { + expect(headers.get("authorization")).toBe("Nostr signed"); + expect(headers.get("x-auth-tag")).toBe(JSON.stringify(auth)); + } + } finally { + data.disconnect(); + } +}); diff --git a/src/agent-host/session.ts b/src/agent-host/session.ts new file mode 100644 index 0000000000..d6d96d7b3b --- /dev/null +++ b/src/agent-host/session.ts @@ -0,0 +1,146 @@ +// The agent's connection to its community on a standalone host: the app's own +// relay session (`createRelaySession`), signed in as the agent. It hears the +// agent's channels live, keeps their rosters and labels, and answers the +// agent's reads, as the owner's session does in the app. It is read-only: the +// agent publishes only through the app's custody, which checks each kind. +import type { EventTemplate, VerifiedEvent } from "nostr-tools"; +import type { NativeCall } from "../features/agents/native"; +import { communityDestination } from "../features/communities/destination"; +import { audienceAuthorityCommunity } from "../features/sessions/audience-authority"; +import { + createRelaySession, + type RelaySession, +} from "../features/relay/session"; +import type { RelayData } from "../features/relay/service"; +import { + connectSignedTransport, + type ReadTransport, + type Signer, +} from "../features/relay/transport"; + +export type AgentSession = Readonly<{ + session: RelaySession; + /** The session as the app's relay service gives it, always ready. */ + relay: RelayData; +}>; + +/** The community's NIP-11 information. */ +export async function relayInfo(relay: string, signal?: AbortSignal) { + const response = await fetch(relay.replace(/^ws/, "http"), { + headers: { accept: "application/nostr+json" }, + signal: signal ?? null, + }); + if (!response.ok) + throw new Error(`Community information: ${response.status}`); + const info = (await response.json()) as Record | null; + if (!info || typeof info !== "object") + throw new Error("Invalid community information"); + return info; +} + +/** Connects as the agent. `scope` partitions what the app keeps for it. + * `call` is the app's custody: it signs the agent's login and reads. */ +export async function connectAgentSession( + options: Readonly<{ + relay: string; + pubkey: string; + call: NativeCall; + scope: string; + }>, +): Promise { + const { pubkey, call, scope } = options; + const origin = communityDestination(options.relay).url; + const info = await relayInfo(origin); + // As the app reads it (`connectNativeTransport`). + const author = info.self; + if (typeof author !== "string" || !/^[a-f0-9]{64}$/.test(author)) + throw new Error("The community did not give its identity"); + const extensions = Array.isArray(info.supported_extensions) + ? info.supported_extensions + : []; + const tag = (template: EventTemplate, name: string) => + template.tags.find((item) => item[0] === name)?.[1]; + const signer: Signer = { + getPublicKey: async () => pubkey, + signEvent: async (template: EventTemplate) => { + if (template.kind !== 22242) + throw new Error("The agent's session signs only its login"); + return call("app_agent_login", { + pubkey, + relay: tag(template, "relay"), + challenge: tag(template, "challenge"), + }); + }, + request: async (url, body, signal) => { + const auth = await call<{ authorization: string; authTag: string }>( + "app_agent_read_auth", + { pubkey, url, body }, + ); + return fetch(url, { + method: "POST", + headers: { + authorization: auth.authorization, + "content-type": "application/json", + "x-auth-tag": auth.authTag, + }, + body, + signal: signal ?? null, + }); + }, + }; + // Nothing the session does publishes as the agent: it has no writer, and + // its live connection sends only the login and its subscriptions. + const { + writer: _writer, + subscribe, + ...signed + } = await connectSignedTransport(signer, origin, author); + const transport: ReadTransport = { + ...signed, + ...(subscribe + ? { + subscribe: (callbacks) => { + const { + publish: _publish, + publishPresence: _presence, + publishTyping: _typing, + ...traffic + } = subscribe(callbacks); + return traffic; + }, + } + : {}), + archiveAuthority: author, + ...(extensions.includes("nip-cl") + ? { channelIdentity: true as const } + : {}), + ...(audienceAuthorityCommunity(info.audience_authority) + ? { + audienceAuthorityCommunity: audienceAuthorityCommunity( + info.audience_authority, + ), + } + : {}), + }; + const store = createRelaySession(transport, { warm: false }); + await store.restore(); + const { session } = store; + const snapshot = Object.freeze({ + status: "ready" as const, + generation: 1, + scope, + origin, + session, + viewer: pubkey, + }); + return Object.freeze({ + session, + relay: { + snapshot: () => snapshot, + subscribe: () => () => {}, + retry: () => session.live.retry(), + disconnect: () => store.dispose(), + clearCache: async () => {}, + }, + }); +} diff --git a/src/agent-host/spawn.test.ts b/src/agent-host/spawn.test.ts new file mode 100644 index 0000000000..89640dc64c --- /dev/null +++ b/src/agent-host/spawn.test.ts @@ -0,0 +1,85 @@ +// @vitest-environment node +import { execFileSync } from "node:child_process"; +import { expect, it, vi } from "vitest"; +import { processHost } from "./spawn"; + +/** Whether the process still runs. A killed one whose new parent has not yet + * reaped it (a zombie, state `Z`) has exited, though signal 0 still finds it. */ +const alive = (pid: number) => { + try { + const state = execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { + encoding: "utf8", + }).trim(); + return !!state && !state.startsWith("Z"); + } catch { + // ps exits non-zero when no such process exists. + return false; + } +}; + +it("settles stopAll only once a process group that ignores SIGTERM is gone", async () => { + const processes = processHost([ + { id: "sh", program: "/bin/sh", args: ["-c"] }, + ]); + let out = ""; + let ready!: () => void; + const started = new Promise((resolve) => (ready = resolve)); + // The shell and the sleep it starts both ignore SIGTERM. + const child = await processes.spawn("sh", { + args: ["trap '' TERM; sleep 60 & echo $$ $!; wait"], + onStdout: (data) => { + out += data; + if (out.includes("\n")) ready(); + }, + }); + await started; + const pids = out.trim().split(" ").map(Number); + expect(pids).toHaveLength(2); + const before = Date.now(); + await processes.stopAll(); + // After the grace period's SIGKILL, not at once. + expect(Date.now() - before).toBeGreaterThanOrEqual(2_900); + for (const pid of pids) expect(alive(pid)).toBe(false); + expect(await child.exited).toBeNull(); +}, 10_000); + +it("kills what a process left running when it exits, stopped or not", async () => { + const processes = processHost([ + { id: "sh", program: "/bin/sh", args: ["-c"] }, + ]); + // A tool that ignores SIGTERM and lets go of the shell's pipes, so the + // shell's exit closes them while it still runs. + const tool = + "(trap '' TERM; exec sleep 60 /dev/null 2>&1) & echo $!"; + const start = async (script: string) => { + let out = ""; + let ready!: () => void; + const printed = new Promise((resolve) => (ready = resolve)); + const child = await processes.spawn("sh", { + args: [script], + onStdout: (data) => { + out += data; + if (out.includes("\n")) ready(); + }, + }); + await printed; + return { child, pid: Number(out.trim()) }; + }; + // Stopped: the shell exits on SIGTERM at once, and stopAll waits for the + // tool to be gone too, without the grace period. + const stopped = await start(`${tool}; wait`); + expect(alive(stopped.pid)).toBe(true); + const before = Date.now(); + await processes.stopAll(); + expect(Date.now() - before).toBeLessThan(2_900); + await vi.waitFor(() => expect(alive(stopped.pid)).toBe(false)); + // Not stopped: the shell ends by itself, and its tool goes with it. + const ended = await start(tool); + expect(await ended.child.exited).toBe(0); + await vi.waitFor(() => expect(alive(ended.pid)).toBe(false)); + await processes.stopAll(); +}, 10_000); + +it("settles stopAll with nothing running", async () => { + await expect(processHost([]).stopAll()).resolves.toBeUndefined(); +}); diff --git a/src/agent-host/spawn.ts b/src/agent-host/spawn.ts new file mode 100644 index 0000000000..1f15615f01 --- /dev/null +++ b/src/agent-host/spawn.ts @@ -0,0 +1,137 @@ +// The desktop app's process host (host_process.rs), for Node: starts only the +// processes a type's manifest declares, with their declared arguments first, +// without any Buzz identity in their environment, each in its own process +// group so stopping one stops what it started. +import { spawn as start } from "node:child_process"; +import { mkdirSync } from "node:fs"; +import { homedir } from "node:os"; +import { delimiter, join } from "node:path"; +import type { Host, HostProcess } from "../features/host/service"; + +export type Declared = Readonly<{ + id: string; + program: string; + args?: readonly string[]; +}>; + +const STOP_GRACE_MS = 3_000; +const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms).unref()); + +/** PATH with per-user installs first and Homebrew's last, as the app has it. */ +export function effectivePath(path = process.env.PATH ?? "") { + const directories = [ + join(homedir(), ".local/bin"), + ...(path || "/usr/bin:/bin").split(delimiter).filter(Boolean), + ]; + if (process.platform === "darwin") + directories.push("/opt/homebrew/bin", "/usr/local/bin"); + return directories.join(delimiter); +} + +function directory(cwd: string) { + if (cwd === "~") return homedir(); + if (cwd.startsWith("~/")) return join(homedir(), cwd.slice(2)); + if (!cwd.startsWith("/")) throw new Error("Invalid process directory"); + return cwd; +} + +/** Starts declared processes; `stopAll` stops every one still running, and + * settles once each has exited. */ +export function processHost(declared: readonly Declared[]) { + const live = new Set<() => Promise>(); + const spawn: NonNullable = async (id, options = {}) => { + const entry = declared.find((item) => item.id === id); + if (!entry) throw new Error("Process is not declared"); + const args = options.args ?? []; + if (args.length > 256 || args.some((arg) => arg.includes("\0"))) + throw new Error("Invalid process arguments"); + const env: Record = {}; + for (const [name, value] of Object.entries(process.env)) + if ( + value !== undefined && + !name.startsWith("BUZZ_") && + !name.startsWith("NOSTR_") + ) + env[name] = value; + env.PATH = effectivePath(); + for (const [name, value] of Object.entries(options.env ?? {})) { + if (!name || /[=\0]/.test(name)) + throw new Error("Invalid environment variable"); + if (value === null) delete env[name]; + else env[name] = value; + } + const cwd = options.cwd ? directory(options.cwd) : undefined; + if (cwd) mkdirSync(cwd, { recursive: true }); + const child = start(entry.program, [...(entry.args ?? []), ...args], { + cwd, + env, + stdio: ["pipe", "pipe", "pipe"], + detached: true, + }); + const started = new Promise((resolve, reject) => { + child.once("spawn", resolve); + child.once("error", reject); + }); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (data: string) => options.onStdout?.(data)); + child.stderr.on("data", (data: string) => options.onStderr?.(data)); + child.stdin.on("error", () => {}); + const signal = (name: NodeJS.Signals) => { + try { + if (child.pid) process.kill(-child.pid, name); + } catch {} + }; + const left = new Promise((resolve) => + child.once("exit", (code) => resolve(code)), + ); + const closed = new Promise((resolve) => + child.once("close", () => resolve()), + ); + // Settles once the process has exited, however it ended, and its group + // with it: as host_process.rs does, what it started and left running is + // killed then. Only its pipes may still be closing, for a while. + const exited = left.then(async (code) => { + signal("SIGKILL"); + await Promise.race([closed, sleep(STOP_GRACE_MS)]); + return code; + }); + let stopping: Promise | undefined; + // SIGTERM to its process group, then SIGKILL to what is still there. + const kill = () => { + stopping ??= (async () => { + signal("SIGTERM"); + const timer = setTimeout(() => signal("SIGKILL"), STOP_GRACE_MS); + await exited; + clearTimeout(timer); + })(); + return stopping; + }; + live.add(kill); + void exited.then(() => live.delete(kill)); + // One that never started has nothing to stop. + started.catch(() => live.delete(kill)); + await started; + const write = (data: string) => + new Promise((resolve, reject) => + child.stdin.write(data, (error) => (error ? reject(error) : resolve())), + ); + const handle: HostProcess = Object.freeze({ + write, + end: async (data?: string) => { + if (data) await write(data); + child.stdin.end(); + }, + kill, + exited, + }); + return handle; + }; + return { + spawn, + stopAll: async () => { + await Promise.all([...live].map((kill) => kill())); + }, + }; +} diff --git a/src/bundled/claude-code/defaults.ts b/src/bundled/claude-code/defaults.ts index b0bea2945a..92156bf2f0 100644 --- a/src/bundled/claude-code/defaults.ts +++ b/src/bundled/claude-code/defaults.ts @@ -6,7 +6,50 @@ export type ClaudeProviderDefaults = Readonly<{ effort?: string; }>; const NATIVE_DEFAULTS: ClaudeProviderDefaults = Object.freeze({ model: "" }); -const STORAGE_KEY = "buzz.claude-code.provider-defaults.v1"; +export const DEFAULTS_KEY = "buzz.claude-code.provider-defaults.v1"; + +/** The saved defaults; Claude Code's own, blank, when none are saved or + * they are malformed. */ +export function readDefaults( + storage: Pick, +): ClaudeProviderDefaults { + try { + const value: unknown = JSON.parse(storage.get(DEFAULTS_KEY) ?? "null"); + if ( + value && + typeof value === "object" && + (value as { version?: unknown }).version === 1 && + typeof (value as { model?: unknown }).model === "string" + ) + return Object.freeze({ + model: (value as { model: string }).model, + ...(typeof (value as { effort?: unknown }).effort === "string" + ? { effort: (value as { effort: string }).effort } + : {}), + }); + } catch { + // Malformed local preferences preserve Claude Code's native default. + } + return NATIVE_DEFAULTS; +} + +/** An agent's settings: its own model, else the defaults'; its own effort, + * else the defaults' when it uses the default model. */ +export function resolveConfig( + raw: unknown, + defaults: ClaudeProviderDefaults, +): Config { + const explicit = config(raw); + return { + ...explicit, + model: explicit.model.trim() ? explicit.model : defaults.model, + effort: + explicit.effort || + (!explicit.model.trim() || explicit.model === defaults.model + ? (defaults.effort ?? "") + : ""), + }; +} /** Device-wide defaults are separate from agent records; a blank agent value * inherits this store and a blank store value leaves Claude Code in charge. */ @@ -15,7 +58,7 @@ export class ClaudeDefaults { private readonly listeners = new Set<() => void>(); constructor(private readonly storage: AgentStore = localStore()) { - this.state = this.read(); + this.state = readDefaults(storage); } snapshot = () => this.state; @@ -24,48 +67,17 @@ export class ClaudeDefaults { return () => this.listeners.delete(listener); }; resolve(raw: unknown): Config { - const explicit = config(raw); - return { - ...explicit, - model: explicit.model.trim() ? explicit.model : this.state.model, - effort: - explicit.effort || - (!explicit.model.trim() || explicit.model === this.state.model - ? (this.state.effort ?? "") - : ""), - }; + return resolveConfig(raw, this.state); } /** Settles once the defaults are saved, and only then uses them; rejects, * keeping the saved ones, when they could not be saved. */ async save(next: ClaudeProviderDefaults) { const { model, effort = "" } = next; await this.storage.set( - STORAGE_KEY, + DEFAULTS_KEY, JSON.stringify({ version: 1, model, effort }), ); this.state = Object.freeze({ model, effort }); for (const listener of this.listeners) listener(); } - private read(): ClaudeProviderDefaults { - try { - const value: unknown = JSON.parse( - this.storage.get(STORAGE_KEY) ?? "null", - ); - if ( - value && - typeof value === "object" && - (value as { version?: unknown }).version === 1 && - typeof (value as { model?: unknown }).model === "string" - ) - return Object.freeze({ - model: (value as { model: string }).model, - ...(typeof (value as { effort?: unknown }).effort === "string" - ? { effort: (value as { effort: string }).effort } - : {}), - }); - } catch { - // Malformed local preferences preserve Claude Code's native default. - } - return NATIVE_DEFAULTS; - } } diff --git a/src/bundled/codex/defaults.ts b/src/bundled/codex/defaults.ts new file mode 100644 index 0000000000..4705ad1ba1 --- /dev/null +++ b/src/bundled/codex/defaults.ts @@ -0,0 +1,67 @@ +// The owner's device-wide Codex defaults, which an agent's blank model and +// thinking inherit. The app's provider dialog saves them in agent storage; the +// app and the standalone agent host resolve an agent's settings with them. +import type { AgentStore } from "../../features/agents/state"; +import { type Config, config, type Model, resolveModel } from "./config"; + +export type ProviderDefaults = Readonly<{ model: string; effort: string }>; +export const DEFAULTS_KEY = "buzz.codex.provider-defaults.v1"; +const NATIVE: ProviderDefaults = Object.freeze({ model: "", effort: "" }); + +/** The saved defaults; Codex's own, blank, when none are saved or they are + * malformed. */ +export function readDefaults( + storage: Pick, +): ProviderDefaults { + try { + const value: unknown = JSON.parse(storage.get(DEFAULTS_KEY) ?? "null"); + if ( + value && + typeof value === "object" && + (value as { version?: unknown }).version === 1 && + typeof (value as { model?: unknown }).model === "string" && + typeof (value as { effort?: unknown }).effort === "string" + ) + return Object.freeze({ + model: config(value).model, + effort: (value as { effort: string }).effort, + }); + } catch { + // Malformed local preferences follow Codex's recommendation. + } + return NATIVE; +} + +/** An agent's settings: its own model and thinking, else the defaults'. + * With Codex's `models`, a default model it no longer lists, and thinking the + * model does not support, fall back to Codex's choice. */ +export function resolveConfig( + raw: unknown, + defaults: ProviderDefaults, + models?: readonly Model[], +): Config { + const find = (model: string) => { + if (!models) return undefined; + const resolved = resolveModel(model, models); + return models.find((choice) => choice.model === resolved); + }; + const explicit = config(raw); + const inheritedModel = defaults.model; + const unavailable = Boolean( + inheritedModel && models && !find(inheritedModel), + ); + const model = explicit.model || (unavailable ? "" : inheritedModel); + if (explicit.effort) return { ...explicit, model }; + const selected = find(model); + const preferred = unavailable && !explicit.model ? "" : defaults.effort; + const effort = !preferred + ? "" + : !models + ? preferred + : selected?.supportedReasoningEfforts.some( + (choice) => choice.reasoningEffort === preferred, + ) + ? preferred + : (selected?.defaultReasoningEffort ?? ""); + return { ...explicit, model, effort }; +} diff --git a/src/bundled/codex/tabs.tsx b/src/bundled/codex/tabs.tsx index d5221861fc..d4cdbf443e 100644 --- a/src/bundled/codex/tabs.tsx +++ b/src/bundled/codex/tabs.tsx @@ -49,6 +49,12 @@ import { type Config, type Model, } from "./config"; +import { + DEFAULTS_KEY, + type ProviderDefaults, + readDefaults, + resolveConfig, +} from "./defaults"; import { AppServer, listModels, needsSignIn, type Spawn } from "./rpc"; import type { CodexRuntime, SavedConversation } from "./runtime"; import type { CodexSetup } from "./setup"; @@ -64,7 +70,6 @@ type Read = type Catalog = { models: Model[]; }; -type ProviderDefaults = Readonly<{ model: string; effort: string }>; type ProviderState = AgentProviderSnapshot & Readonly<{ checking: boolean; @@ -72,7 +77,6 @@ type ProviderState = AgentProviderSnapshot & error: string; defaults: ProviderDefaults; }>; -const STORAGE_KEY = "buzz.codex.provider-defaults.v1"; const SIGN_IN_ERROR = "Codex needs you to sign in."; const NOT_INSTALLED = "Codex is not installed."; const CONVERSATION_LIMIT = 30; @@ -113,7 +117,7 @@ export class CodexProvider { checking: false, data: undefined, error: "", - defaults: this.readDefaults(), + defaults: readDefaults(this.storage), }); // An install or sign-in holds the dialog open, then Codex is checked again. this.offSetup = setup?.subscribe(() => { @@ -203,32 +207,13 @@ export class CodexProvider { : (selected?.defaultReasoningEffort ?? ""); const defaults = Object.freeze({ model: next.model, effort }); await this.storage.set( - STORAGE_KEY, + DEFAULTS_KEY, JSON.stringify({ version: 1, ...defaults }), ); this.set({ defaults }); } resolve(raw: unknown): Config { - const explicit = readConfig(raw); - const inheritedModel = this.state.defaults.model; - const unavailable = Boolean( - inheritedModel && this.state.data && !this.model(inheritedModel), - ); - const model = explicit.model || (unavailable ? "" : inheritedModel); - if (explicit.effort) return { ...explicit, model }; - const selected = this.model(model); - const preferred = - unavailable && !explicit.model ? "" : this.state.defaults.effort; - const effort = !preferred - ? "" - : !this.state.data - ? preferred - : selected?.supportedReasoningEfforts.some( - (choice) => choice.reasoningEffort === preferred, - ) - ? preferred - : (selected?.defaultReasoningEffort ?? ""); - return { ...explicit, model, effort }; + return resolveConfig(raw, this.state.defaults, this.state.data?.models); } dispose() { this.offSetup?.(); @@ -238,27 +223,6 @@ export class CodexProvider { private model(model: string) { return catalogModel(model, this.state.data); } - private readDefaults(): ProviderDefaults { - try { - const value: unknown = JSON.parse( - this.storage.get(STORAGE_KEY) ?? "null", - ); - if ( - value && - typeof value === "object" && - (value as { version?: unknown }).version === 1 && - typeof (value as { model?: unknown }).model === "string" && - typeof (value as { effort?: unknown }).effort === "string" - ) - return Object.freeze({ - model: readConfig(value).model, - effort: (value as { effort: string }).effort, - }); - } catch { - // Malformed local preferences follow Codex's recommendation. - } - return Object.freeze({ model: "", effort: "" }); - } } const message = (error: unknown) => diff --git a/src/features/agents/native.ts b/src/features/agents/native.ts index deab7d24d6..e1f2d2c36e 100644 --- a/src/features/agents/native.ts +++ b/src/features/agents/native.ts @@ -164,63 +164,69 @@ export type AgentsNative = { release(pubkey: string): Promise; }; +/** Native custody through `call`: the app's own commands, or the same + * commands as the app serves them to `buzz-agent-host`. */ +export type NativeCall = ( + command: string, + args?: Record, +) => Promise; + export const nativeAgents = (): AgentsNative | undefined => - isTauri() - ? { - list: () => invoke("app_agent_list"), - create: (destination, owner, agentType, name) => - invoke("app_agent_create", { destination, owner, agentType, name }), - portable: () => invoke("app_agent_portable"), - port: (source, pubkey, destination, owner, agentType) => - invoke("app_agent_port", { - source, - pubkey, - destination, - owner, - agentType, - }), - rename: (pubkey, name) => invoke("app_agent_rename", { pubkey, name }), - setPicture: (pubkey, picture) => - invoke("app_agent_set_picture", { pubkey, picture }), - setType: (pubkey, agentType) => - invoke("app_agent_set_type", { pubkey, agentType }), - beginDelete: (pubkey) => invoke("app_agent_begin_delete", { pubkey }), - remove: (pubkey) => invoke("app_agent_delete", { pubkey }), - forget: (pubkey) => invoke("app_agent_forget", { pubkey }), - publish: (pubkey, event) => - invoke("app_agent_publish", { - pubkey, - event: { - kind: event.kind, - content: event.content, - tags: event.tags ?? [], - }, - }), - query: async (pubkey, filters) => - eventsDto(await invoke("app_agent_query", { pubkey, filters })), - audienceAuthority: (pubkey, destination, sources) => - invoke("app_agent_query", { - pubkey, - filters: audienceAuthorityFilter(destination, sources), - }), - searchProfiles: async (pubkey, query, limit) => { - const found = await invoke("app_agent_search_profiles", { - pubkey, - query, - limit, - }); - return found == null ? undefined : eventsDto(found); - }, - upload: (pubkey, data, mime) => - invoke("app_agent_upload", { pubkey, data, mime }), - media: (pubkey, url) => invoke("app_agent_media", { pubkey, url }), - gitAuth: (pubkey, owner, repo) => - invoke("app_agent_git_auth", { pubkey, owner, repo }), - remember: (pubkey, slug, body, after) => - invoke("app_agent_remember", { pubkey, slug, body, after }), - publishProfile: (pubkey) => - invoke("app_agent_publish_profile", { pubkey }), - claim: (pubkey) => invoke("app_agent_claim", { pubkey }), - release: (pubkey) => invoke("app_agent_release", { pubkey }), - } - : undefined; + isTauri() ? agentsNative(invoke) : undefined; + +export const agentsNative = (invoke: NativeCall): AgentsNative => ({ + list: () => invoke("app_agent_list"), + create: (destination, owner, agentType, name) => + invoke("app_agent_create", { destination, owner, agentType, name }), + portable: () => invoke("app_agent_portable"), + port: (source, pubkey, destination, owner, agentType) => + invoke("app_agent_port", { + source, + pubkey, + destination, + owner, + agentType, + }), + rename: (pubkey, name) => invoke("app_agent_rename", { pubkey, name }), + setPicture: (pubkey, picture) => + invoke("app_agent_set_picture", { pubkey, picture }), + setType: (pubkey, agentType) => + invoke("app_agent_set_type", { pubkey, agentType }), + beginDelete: (pubkey) => invoke("app_agent_begin_delete", { pubkey }), + remove: (pubkey) => invoke("app_agent_delete", { pubkey }), + forget: (pubkey) => invoke("app_agent_forget", { pubkey }), + publish: (pubkey, event) => + invoke("app_agent_publish", { + pubkey, + event: { + kind: event.kind, + content: event.content, + tags: event.tags ?? [], + }, + }), + query: async (pubkey, filters) => + eventsDto(await invoke("app_agent_query", { pubkey, filters })), + audienceAuthority: (pubkey, destination, sources) => + invoke("app_agent_query", { + pubkey, + filters: audienceAuthorityFilter(destination, sources), + }), + searchProfiles: async (pubkey, query, limit) => { + const found = await invoke("app_agent_search_profiles", { + pubkey, + query, + limit, + }); + return found == null ? undefined : eventsDto(found); + }, + upload: (pubkey, data, mime) => + invoke("app_agent_upload", { pubkey, data, mime }), + media: (pubkey, url) => invoke("app_agent_media", { pubkey, url }), + gitAuth: (pubkey, owner, repo) => + invoke("app_agent_git_auth", { pubkey, owner, repo }), + remember: (pubkey, slug, body, after) => + invoke("app_agent_remember", { pubkey, slug, body, after }), + publishProfile: (pubkey) => invoke("app_agent_publish_profile", { pubkey }), + claim: (pubkey) => invoke("app_agent_claim", { pubkey }), + release: (pubkey) => invoke("app_agent_release", { pubkey }), +}); diff --git a/src/features/agents/record-changes.ts b/src/features/agents/record-changes.ts new file mode 100644 index 0000000000..d0942ba27c --- /dev/null +++ b/src/features/agents/record-changes.ts @@ -0,0 +1,132 @@ +import { type AgentRecord, type TimerProgress, withProgress } from "./store"; + +const EMPTY = Object.freeze({}); + +/** Where the records are kept, for `RecordChanges`. */ +export type RecordStore = Readonly<{ + /** The agent's record as kept now. */ + saved: (pubkey: string) => AgentRecord | undefined; + /** Saves `next` over `base` (null removes it), then keeps it. */ + commit: ( + pubkey: string, + base: AgentRecord | undefined, + next: AgentRecord | null, + ) => Promise; + /** Keeps `next` without saving it. */ + keep: (pubkey: string, next: AgentRecord) => void; + /** A change was kept: tell the owner and the agent. */ + changed: () => void; + /** The agent has no changes left: its timers can run. */ + settled: (pubkey: string) => void; +}>; + +/** The changes to each agent's record. This is the only way a record + * changes: owner settings, the agent's attention tools, timer progress, a + * provider change and removal all send a change here. */ +export class RecordChanges { + /** Each agent's last change in its queue, until it settles. */ + private readonly changes = new Map>(); + /** How many changes other than timer progress each agent has queued. */ + private readonly counts = new Map(); + /** Each agent's timer progress that is not saved yet. */ + private readonly progressing = new Map(); + + constructor(private readonly store: RecordStore) {} + + /** Whether the agent has changes other than timer progress queued. */ + changing(pubkey: string) { + return this.counts.has(pubkey); + } + + /** Changes one agent's record. One agent's changes run in the order they + * are sent, each after the one before it is saved or has failed, so each + * one applies to the newest saved record, never to a copy another change + * replaces. `apply` gets that record (or undefined) and returns the next + * one, or null to remove it; it may throw to change nothing. The owner and + * the agent see a change once it is saved, and the promise settles then. + * When the save fails, the promise rejects and they keep the saved record, + * so a form keeps its unsaved changes. `saved` runs as the change is + * shown, before anyone is told. */ + change( + pubkey: string, + apply: ( + record: AgentRecord | undefined, + ) => AgentRecord | null | Promise, + saved?: (record: AgentRecord | null) => void, + ): Promise { + this.counts.set(pubkey, (this.counts.get(pubkey) ?? 0) + 1); + return this.queued(pubkey, async () => { + try { + const base = this.store.saved(pubkey); + const next = await apply(base); + await this.store.commit(pubkey, base, next); + saved?.(next); + this.store.changed(); + return next; + } finally { + const left = (this.counts.get(pubkey) ?? 1) - 1; + if (left) this.counts.set(pubkey, left); + else { + this.counts.delete(pubkey); + // Its timers waited for this change, saved or failed. + this.store.settled(pubkey); + } + } + }); + } + + /** Records that timers ran or started counting, as a change to their run + * state only. It applies to whatever the record is when its turn comes, so + * it never undoes a change sent before it: when a timer was rearmed or + * restarted meanwhile, its new deadline stays and only the occurrences it + * used are added. Timers see it at once (`timers`), and memory keeps it + * even when its save fails, so a timer does not run again at once. */ + progress(pubkey: string, ran: TimerProgress) { + const pending = this.progressing.get(pubkey) ?? []; + this.progressing.set(pubkey, [...pending, ran]); + void this.queued(pubkey, async () => { + try { + const base = this.store.saved(pubkey); + const next = base && withProgress(base, ran); + if (!base || !next) return; + await this.store.commit(pubkey, base, next).catch((error) => { + console.warn(`Agent ${pubkey} timer state was not saved`, error); + this.store.keep(pubkey, next); + }); + this.store.changed(); + } finally { + const left = (this.progressing.get(pubkey) ?? []).filter( + (entry) => entry !== ran, + ); + if (left.length) this.progressing.set(pubkey, left); + else this.progressing.delete(pubkey); + } + }); + } + + /** The agent's timers' run state as its timers go by: saved, with the + * progress still being saved. */ + timers(pubkey: string) { + const record = this.store.saved(pubkey); + const pending = this.progressing.get(pubkey); + if (!record || !pending) return record?.timers ?? EMPTY; + return ( + pending.reduce((next, ran) => withProgress(next, ran) ?? next, record) + .timers ?? EMPTY + ); + } + + /** Runs `step` after the agent's earlier changes have settled. */ + private queued(pubkey: string, step: () => Promise) { + const run = (this.changes.get(pubkey) ?? Promise.resolve()).then(step); + const tail = run.then( + () => {}, + () => {}, + ); + this.changes.set(pubkey, tail); + void tail.then(() => { + if (this.changes.get(pubkey) === tail) this.changes.delete(pubkey); + }); + return run; + } +} diff --git a/src/features/agents/roster.ts b/src/features/agents/roster.ts new file mode 100644 index 0000000000..3e942cdcfc --- /dev/null +++ b/src/features/agents/roster.ts @@ -0,0 +1,47 @@ +// Whether an agent is in an event's channel, and whether that channel is its +// DM, by the relay-signed roster of a session that hears the channel: the +// owner's in the app, the agent's own on a standalone host. +import type { RelayEvent } from "../relay/events"; +import type { RelaySession } from "../relay/session"; + +/** The event's channel as the session lists it, if at all. */ +function channelOf(session: RelaySession | undefined, event: RelayEvent) { + const channel = event.tags.find((tag) => tag[0] === "h")?.[1]; + const channels = session?.channels; + if (!channel || !channels) return undefined; + return ( + channels.list().channels.find((item) => item.id === channel) ?? + channels.get?.(channel) + ); +} + +/** Whether the agent is in the event's channel, by its relay-signed roster. + * The stream may be another's, so a watch, even of "all", sees only what the + * agent itself could read; an event with no channel, or whose roster is + * unknown, wakes no watch. A roster only loaded from disk decides nothing + * (undefined): the agent may have been removed or added since it was saved, + * so the event waits until the relay confirms it. */ +export function rosterMember( + session: RelaySession | undefined, + agent: string, + event: RelayEvent, +): boolean | undefined { + const summary = channelOf(session, event); + if (summary?.cached) return undefined; + return !!summary?.members?.includes(agent); +} + +/** Whether the event's channel is a DM of exactly the agent and one other, + * by the same roster. A group DM, or one whose roster is unknown, is not. */ +export function rosterDm( + session: RelaySession | undefined, + agent: string, + event: RelayEvent, +) { + const summary = channelOf(session, event); + return ( + summary?.channelType === "dm" && + summary.members?.length === 2 && + summary.members.includes(agent) + ); +} diff --git a/src/features/agents/runner.test.ts b/src/features/agents/runner.test.ts new file mode 100644 index 0000000000..2c0507066d --- /dev/null +++ b/src/features/agents/runner.test.ts @@ -0,0 +1,160 @@ +// AgentRunners through a host other than AgentsService, as the standalone +// agent host (src/agent-host) gives it: the RunnerHost contract alone. +import { expect, it, vi } from "vitest"; +import type { RelayEvent } from "../relay/events"; +import { createLocalAgentActivity } from "./local-activity"; +import { AgentRunners, type RunnerHost } from "./runner"; +import type { Agent, Delivery, RegisteredAgentType } from "./service"; +import type { AgentRecord } from "./store"; + +const pubkey = "a".repeat(64); +const owner = "b".repeat(64); +const someone = "c".repeat(64); +const agent: Agent = Object.freeze({ + pubkey, + name: "Echo", + type: "test/echo", + owner, + relay: "wss://relay.example.test", + place: "here", + attention: {}, + skipped: {}, + timers: {}, + attentionEnabled: false, + config: { reply: "hi" }, +}); +const mention = (id: string) => + ({ + id: id.padEnd(64, "0"), + pubkey: someone, + kind: 9, + created_at: Math.floor(Date.now() / 1000), + tags: [ + ["h", "channel"], + ["p", pubkey], + ], + content: "hello", + sig: "f".repeat(128), + }) as unknown as RelayEvent; + +function fixture(holds = true) { + let live = true; + const record: AgentRecord = { pubkey, attention: {}, config: agent.config }; + const custody = { + claim: vi.fn(async () => holds), + release: vi.fn(async () => {}), + publish: vi.fn(async (_: string, event: { kind: number }) => ({ + ...mention("e"), + pubkey, + kind: event.kind, + })), + query: vi.fn(async () => []), + upload: vi.fn(), + media: vi.fn(), + remember: vi.fn(), + }; + const host: RunnerHost = { + ready: (key) => (live && key === pubkey ? agent : undefined), + agent: (key) => (key === pubkey ? agent : undefined), + member: () => true, + dm: () => false, + community: () => undefined, + records: () => [record], + checkpoints: { read: () => ({}), write: () => {} }, + interrupted: { read: () => ({}), write: () => {}, changed: () => {} }, + scope: () => "scope", + saved: () => record, + record: () => record, + change: async (_, apply) => apply(record), + changing: () => false, + timers: () => ({}), + progress: () => {}, + start: () => ({ ok: true, config: agent.config, revisions: [] }), + team: () => "", + templates: () => [], + tabs: () => undefined, + classifier: () => "unavailable", + decisions: () => undefined, + keyStatus: () => undefined, + classifierKey: () => "missing", + classifierKeyGeneration: () => 0, + setClassifierKey: () => {}, + context: async () => { + throw new Error("No classifier"); + }, + displayName: () => null, + custody: custody as unknown as RunnerHost["custody"], + activity: createLocalAgentActivity(), + soloChat: async () => false, + remembered: () => {}, + finished: () => {}, + named: () => {}, + }; + const ran: Delivery[] = []; + const type = { + id: "echo", + title: "Echo", + key: "test/echo", + pluginId: "test", + revision: "one", + defaults: () => ({ config: agent.config }), + run: vi.fn(async (delivery: Delivery) => { + ran.push(delivery); + await delivery.agent.publish({ kind: 9, content: "hi", tags: [] }); + }), + } as unknown as RegisteredAgentType; + const runners = new AgentRunners(host); + runners.reconcile([{ pubkey, type: type.key }], new Map([[type.key, type]])); + return { + runners, + custody, + ran, + type, + offline: () => { + live = false; + }, + }; +} + +it("runs a mention once it holds the agent, and publishes as the agent", async () => { + const f = fixture(); + const event = mention("1"); + f.runners.dispatch({ events: [event], channelId: "channel" }, () => true, 0); + // The same event again, as after a reconnect's catch-up, runs once. + f.runners.dispatch({ events: [event], channelId: "channel" }, () => true, 0); + await vi.waitFor(() => expect(f.ran).toHaveLength(1)); + expect(f.custody.claim).toHaveBeenCalledWith(pubkey); + expect(f.ran[0]).toMatchObject({ + trigger: { type: "mention", event }, + channelId: "channel", + config: agent.config, + agent: { pubkey, owner }, + }); + await vi.waitFor(() => + expect(f.custody.publish).toHaveBeenCalledWith( + pubkey, + expect.objectContaining({ kind: 9, content: "hi" }), + ), + ); + expect(f.runners.runs(pubkey)).toBe(true); +}); + +it("leaves a mention to the copy that holds the agent", async () => { + const f = fixture(false); + f.runners.dispatch({ events: [mention("2")] }, () => true, 0); + await vi.waitFor(() => expect(f.custody.claim).toHaveBeenCalled()); + await Promise.resolve(); + expect(f.type.run).not.toHaveBeenCalled(); + expect(f.runners.runs(pubkey)).toBe(false); +}); + +it("releases the agent once its community is no longer live", async () => { + const f = fixture(); + f.runners.resume(new Set([pubkey])); + await vi.waitFor(() => expect(f.custody.claim).toHaveBeenCalled()); + f.offline(); + f.runners.resume(new Set()); + await vi.waitFor(() => + expect(f.custody.release).toHaveBeenCalledWith(pubkey), + ); +}); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 611fe8e762..14cfcd7c87 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -8,6 +8,8 @@ import { isSynced, type Mode, resolveStart } from "./agent-start"; import { AgentRunners, type Connection, TIMEOUT_LIMIT_MS } from "./runner"; import { CoreMemoryCache } from "./core-memory-cache"; import { ownerReads } from "./reads"; +import { RecordChanges } from "./record-changes"; +import { rosterDm, rosterMember } from "./roster"; // Agents: agents that are plugins from the start. A plugin registers an agent // type (its summary line, peek view, settings tabs and `run`); each agent made // from it has its own native-held key, an `agent-attention/v1` configuration the @@ -108,9 +110,7 @@ import { writeRecord, writeTeams, writeUnfinished, - withProgress, type AgentRecord, - type TimerProgress, type SkippedObject, type Team, type Unfinished, @@ -1128,12 +1128,15 @@ export class AgentsService extends Service implements Agents { private identities: readonly AgentIdentity[] = []; /** Each agent's settings as saved, and as the owner and the agent see them. */ private records: Record; - /** Each agent's last change in its queue, until it settles. */ - private readonly changes = new Map>(); - /** How many changes other than timer progress each agent has queued. */ - private readonly changing = new Map(); - /** Each agent's timer progress that is not saved yet. */ - private readonly progressing = new Map(); + private readonly changes = new RecordChanges({ + saved: (pubkey) => this.records[pubkey], + commit: (pubkey, base, next) => this.commit(pubkey, base, next), + keep: (pubkey, next) => { + this.records = { ...this.records, [pubkey]: next }; + }, + changed: () => this.update(), + settled: (pubkey) => this.runners.wake(pubkey), + }); private teamList: readonly Team[]; private state: AgentsSnapshot; /** By origin, so each agent listens in its own community, selected or not. */ @@ -1235,8 +1238,10 @@ export class AgentsService extends Service implements Agents { ownershipChanged: () => this.notify(), ready: (pubkey) => this.ready(pubkey), agent: (pubkey) => this.agent(pubkey), - member: (agent, event) => this.member(agent, event), - dm: (agent, event) => this.dm(agent, event), + member: (agent, event) => + rosterMember(this.bindingOf(agent)?.session, agent.pubkey, event), + dm: (agent, event) => + rosterDm(this.bindingOf(agent)?.session, agent.pubkey, event), starting: (agent, channelId) => this.bindingOf(agent)?.session.channelWrites.starting(channelId), community: (agent) => this.bindingOf(agent), @@ -1272,7 +1277,7 @@ export class AgentsService extends Service implements Agents { return this.recordOf(identity); }, change: async (pubkey, apply) => { - const record = await this.change(pubkey, (saved) => { + const record = await this.changes.change(pubkey, (saved) => { const identity = this.live(pubkey); if (!identity) throw new Error("No such agent on this device"); return apply(saved ?? this.recordOf(identity)); @@ -1280,9 +1285,9 @@ export class AgentsService extends Service implements Agents { if (!record) throw new Error("No such agent on this device"); return record; }, - changing: (pubkey) => this.changing.has(pubkey), - timers: (pubkey) => this.timersOf(pubkey), - progress: (pubkey, ran) => this.progress(pubkey, ran), + changing: (pubkey) => this.changes.changing(pubkey), + timers: (pubkey) => this.changes.timers(pubkey), + progress: (pubkey, ran) => this.changes.progress(pubkey, ran), start: (agent) => { const { settings, instructions } = this.configBinding?.(agent) ?? {}; const merge = this.types().find( @@ -2017,7 +2022,7 @@ export class AgentsService extends Service implements Agents { record = { ...record, config: made.config }; if (carry && kind.adopt) record = { ...record, config: kind.adopt(carry, record.config) }; - await this.change(identity.pubkey, () => record); + await this.changes.change(identity.pubkey, () => record); } catch (error) { // The agent exists either way; unsaved settings start from the defaults. console.warn(`Agent ${identity.name} settings were not saved`, error); @@ -2066,7 +2071,7 @@ export class AgentsService extends Service implements Agents { change.attention || change.attentionEnabled !== undefined ) - await this.change(pubkey, (saved) => + await this.changes.change(pubkey, (saved) => changed(saved ?? this.recordOf(identity), change), ); const picture = @@ -2213,7 +2218,7 @@ export class AgentsService extends Service implements Agents { ); await native.setType(pubkey, type); try { - await this.change( + await this.changes.change( pubkey, (saved) => ({ ...(saved ?? this.recordOf(identity)), @@ -2657,85 +2662,6 @@ export class AgentsService extends Service implements Agents { return agent; } - /** Changes one agent's record. This is the only way a record changes: - * owner settings, the agent's attention tools, timer progress, a provider - * change and removal all send a change here. One agent's changes run in - * the order they are sent, each after the one before it is saved or has - * failed, so each one applies to the newest saved record, never to a copy - * another change replaces. `apply` gets that record (or undefined) and - * returns the next one, or null to remove it; it may throw to change - * nothing. The owner and the agent see a change once it is saved, and the - * promise settles then. When the save fails, the promise rejects and they - * keep the saved record, so a form keeps its unsaved changes. `saved` - * runs as the change is shown, before anyone is told. */ - private change( - pubkey: string, - apply: ( - record: AgentRecord | undefined, - ) => AgentRecord | null | Promise, - saved?: (record: AgentRecord | null) => void, - ): Promise { - this.changing.set(pubkey, (this.changing.get(pubkey) ?? 0) + 1); - return this.queued(pubkey, async () => { - try { - const base = this.records[pubkey]; - const next = await apply(base); - await this.commit(pubkey, base, next); - saved?.(next); - this.update(); - return next; - } finally { - const left = (this.changing.get(pubkey) ?? 1) - 1; - if (left) this.changing.set(pubkey, left); - else { - this.changing.delete(pubkey); - // Its timers waited for this change, saved or failed. - this.runners.wake(pubkey); - } - } - }); - } - /** Records that timers ran or started counting, as a change to their run - * state only. It applies to whatever the record is when its turn comes, so - * it never undoes a change sent before it: when a timer was rearmed or - * restarted meanwhile, its new deadline stays and only the occurrences it - * used are added. Timers see it at once (`timersOf`), and memory keeps it - * even when its save fails, so a timer does not run again at once. */ - private progress(pubkey: string, ran: TimerProgress) { - const pending = this.progressing.get(pubkey) ?? []; - this.progressing.set(pubkey, [...pending, ran]); - void this.queued(pubkey, async () => { - try { - const base = this.records[pubkey]; - const next = base && withProgress(base, ran); - if (!base || !next) return; - await this.commit(pubkey, base, next).catch((error) => { - console.warn(`Agent ${pubkey} timer state was not saved`, error); - this.records = { ...this.records, [pubkey]: next }; - }); - this.update(); - } finally { - const left = (this.progressing.get(pubkey) ?? []).filter( - (entry) => entry !== ran, - ); - if (left.length) this.progressing.set(pubkey, left); - else this.progressing.delete(pubkey); - } - }); - } - /** Runs `step` after the agent's earlier changes have settled. */ - private queued(pubkey: string, step: () => Promise) { - const run = (this.changes.get(pubkey) ?? Promise.resolve()).then(step); - const tail = run.then( - () => {}, - () => {}, - ); - this.changes.set(pubkey, tail); - void tail.then(() => { - if (this.changes.get(pubkey) === tail) this.changes.delete(pubkey); - }); - return run; - } /** Saves `next` over `base` in the agent's own key, then keeps it. Only * this agent's key: others may be another program's to change. */ private async commit( @@ -2750,17 +2676,6 @@ export class AgentsService extends Service implements Agents { const { [pubkey]: _, ...others } = this.records; this.records = next ? { ...others, [pubkey]: next } : others; } - /** The agent's timers' run state as its timers go by: saved, with the - * progress still being saved. */ - private timersOf(pubkey: string) { - const record = this.records[pubkey]; - const pending = this.progressing.get(pubkey); - if (!record || !pending) return record?.timers ?? EMPTY; - return ( - pending.reduce((next, ran) => withProgress(next, ran) ?? next, record) - .timers ?? EMPTY - ); - } private live(pubkey: string) { return this.identities.find( (identity) => identity.pubkey === pubkey && !identity.deleted, @@ -2900,9 +2815,11 @@ export class AgentsService extends Service implements Agents { (pubkey) => !live.has(pubkey), ); for (const pubkey of gone) - void this.change(pubkey, () => null).catch((error) => - console.warn(`Agent ${pubkey} settings were not removed`, error), - ); + void this.changes + .change(pubkey, () => null) + .catch((error) => + console.warn(`Agent ${pubkey} settings were not removed`, error), + ); // A deleted agent leaves its teams; the teams stay. if ( this.teamList.some((team) => @@ -3301,37 +3218,6 @@ export class AgentsService extends Service implements Agents { private notify() { for (const listener of this.listeners) listener(); } - /** The event's channel as the agent's community knows it, if at all. */ - private channelOf(agent: Agent, event: RelayEvent) { - const channel = event.tags.find((tag) => tag[0] === "h")?.[1]; - const channels = this.bindingOf(agent)?.session.channels; - if (!channel || !channels) return undefined; - return ( - channels.list().channels.find((item) => item.id === channel) ?? - channels.get?.(channel) - ); - } - /** Whether the agent is in the event's channel, by its relay-signed roster. - * The stream is the owner's, so a watch, even of "all", sees only what the - * agent itself could read; an event with no channel, or whose roster is - * unknown, wakes no watch. A roster only loaded from disk decides nothing - * (undefined): the agent may have been removed or added since it was saved, - * so the event waits until the relay confirms it. */ - private member(agent: Agent, event: RelayEvent): boolean | undefined { - const summary = this.channelOf(agent, event); - if (summary?.cached) return undefined; - return !!summary?.members?.includes(agent.pubkey); - } - /** Whether the event's channel is a DM of exactly the agent and one other, - * by the same roster. A group DM, or one whose roster is unknown, is not. */ - private dm(agent: Agent, event: RelayEvent) { - const summary = this.channelOf(agent, event); - return ( - summary?.channelType === "dm" && - summary.members?.length === 2 && - summary.members.includes(agent.pubkey) - ); - } } /** An old Buzz team's copy for one owner in one community (`scope`), the same diff --git a/src/features/agents/solo-chat.ts b/src/features/agents/solo-chat.ts index 5494add818..1eed293c6a 100644 --- a/src/features/agents/solo-chat.ts +++ b/src/features/agents/solo-chat.ts @@ -1,3 +1,4 @@ +import type { RelaySession } from "../relay/session"; import type { RelayData } from "../relay/service"; /** Whether a native agent's mention continues a Solo chat: one @@ -14,7 +15,12 @@ export async function soloChat( const snapshot = relay()?.snapshot(); if (snapshot?.status !== "ready" || snapshot.viewer !== owner) return false; signal.throwIfAborted(); - const { channels } = snapshot.session; + return soloChannel(snapshot.session, channelId); +} + +/** Whether the session lists the channel as a Solo chat, by its label. */ +export function soloChannel(session: RelaySession, channelId: string) { + const { channels } = session; const list = channels.list(); const channel = channels.get?.(channelId) ?? diff --git a/src/features/agents/state.ts b/src/features/agents/state.ts index 125f3b33fe..97dac51218 100644 --- a/src/features/agents/state.ts +++ b/src/features/agents/state.ts @@ -12,6 +12,7 @@ import { invoke, isTauri } from "@tauri-apps/api/core"; import { CONFIG_STORAGE_KEY } from "./agent-config"; import { CATCH_UP_KEY, CATCH_UP_PREFIX } from "./catch-up"; import { INTERRUPTED_PREFIX } from "./interrupted"; +import type { NativeCall } from "./native"; import { localStore, type AgentStore, @@ -64,14 +65,17 @@ type Native = Readonly<{ add?(key: string, value: string): Promise; }>; -const native: Native = { - load: () => invoke("app_agent_state_load"), +/** The agent storage files through `call`, as `nativeAgents` uses them. */ +export const nativeFiles = (call: NativeCall): Native => ({ + load: () => call("app_agent_state_load"), set: async (key, value) => { - await invoke("app_agent_state_set", { key, value }); + await call("app_agent_state_set", { key, value }); }, add: (key, value) => - invoke("app_agent_state_set", { key, value, absent: true }), -}; + call("app_agent_state_set", { key, value, absent: true }), +}); +// Looked up per call, so a module that mocks only part of Tauri still loads. +const native = nativeFiles((command, args) => invoke(command, args)); /** Agent storage holds this key once every agent key in localStorage was * moved into it. */ diff --git a/src/features/agents/store.ts b/src/features/agents/store.ts index 00dc5a706f..b8a4ad17a3 100644 --- a/src/features/agents/store.ts +++ b/src/features/agents/store.ts @@ -106,6 +106,16 @@ export function readRecords(storage: AgentStore): Record { return records; } +/** One agent's saved record, from its own key or the legacy shared one. */ +export function readOne( + storage: AgentStore, + pubkey: string, +): AgentRecord | undefined { + const record = + parsed(storage.get(recordKey(pubkey))) ?? legacyRecords(storage)[pubkey]; + return record ? readRecord(pubkey, record as AgentRecord) : undefined; +} + function parsed(text: string | null): unknown { try { return JSON.parse(text ?? "null"); diff --git a/vite.agent-host.config.ts b/vite.agent-host.config.ts new file mode 100644 index 0000000000..d11801cf10 --- /dev/null +++ b/vite.agent-host.config.ts @@ -0,0 +1,16 @@ +import { defineConfig } from "vite"; + +// Builds the standalone agent host: one Node script with its dependencies +// bundled, run as `node dist-agent-host/buzz-agent-host.mjs run `. +export default defineConfig({ + publicDir: false, + ssr: { noExternal: true, target: "node" }, + build: { + outDir: "dist-agent-host", + emptyOutDir: true, + ssr: "src/agent-host/main.ts", + rolldownOptions: { + output: { entryFileNames: "buzz-agent-host.mjs", codeSplitting: false }, + }, + }, +});