diff --git a/agents.html b/agents.html new file mode 100644 index 0000000000..68336fe7ca --- /dev/null +++ b/agents.html @@ -0,0 +1,2 @@ + +Agents — Buzz diff --git a/docs/agent-protection.md b/docs/agent-protection.md index c66a1ce57a..a6cc45a238 100644 --- a/docs/agent-protection.md +++ b/docs/agent-protection.md @@ -37,7 +37,9 @@ complete ordinary version-2 multi-community store. A plugin injects `agentProtection`. `register` installs its checker and native process requirements; `registerModels` exposes its runtime's model choices and -bounded, tool-free completion. `snapshot`, `saveDefaults` and `saveAgent` read +bounded, tool-free completion. Call both from the plugin's `agent` entry: each +page that runs agents has its own `agentProtection`, and a run checks its +actions only with the checkers registered in its page. `snapshot`, `saveDefaults` and `saveAgent` read and update owner defaults and per-identity bindings. Saves use native revision checks; a plugin cannot replace another provider's existing binding. Defaults apply only when committing a new or adopted identity. diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index c31cb71b3d..339d808950 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -297,9 +297,10 @@ agent-record migrations; blank per-agent values inherit them. Code that agents need while they run, such as an agent type's runtime, tools, wakes and Attention sources, belongs in the plugin's optional `agent` export: `export const agent = { inject, apply(ctx) }`, with its own `inject`. `apply` -stays the plugin's UI. Both parts now start in the same page, under the same -plugin identity; the plugin is active once both are, and fails if either does. -This separation lets agents later run apart from the UI page. +stays the plugin's UI. Both parts start under the same plugin identity; the +plugin is active once both are, and fails if either does. In the macOS desktop +app, agent entries run in a hidden agents page and UI entries in the screens' +page; elsewhere both run in the screens' page. A harness plugin registers its type from its `agent` entry and presents it from `apply`. The `agent` entry registers what runs: `defaults`, `run`, @@ -345,7 +346,11 @@ other watches keep no such record, so they still match them. The bundled Agent Attention UI plugin calls it. Protection plugins inject `ctx.agentProtection` to register checks and review -models and save native owner defaults or per-agent policies. An `AgentType` +models and save native owner defaults or per-agent policies. Register the +checker and review models from the `agent` entry, which runs beside the agents +it checks; save policies from either entry. A checker registered from `apply` +does not reach agents that run in the agents page, so their protected runs stop +as with a missing provider. An `AgentType` declares `protection` / `protectionUnavailable` and retires native sessions with `retire(key)`, where `key` is the membership key. Protected runs receive `AgentHandle.protection`; missing providers or enforcement stop them. This diff --git a/src-tauri/build.rs b/src-tauri/build.rs index d82ea73e84..d1a5462bf6 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -138,6 +138,8 @@ fn main() { "clipboard_write_text", "media_stream_base", "get_os_idle_seconds", + "agents_port_listen", + "agents_port_send", "workspace_pick_folder", "repositories_folder_validate", "legacy_communities", diff --git a/src-tauri/capabilities/agents.json b/src-tauri/capabilities/agents.json new file mode 100644 index 0000000000..958136fec4 --- /dev/null +++ b/src-tauri/capabilities/agents.json @@ -0,0 +1,109 @@ +{ + "$schema": "../gen/schemas/desktop-schema.json", + "identifier": "agents-page", + "description": "The hidden page that runs agents: relay, agent, plugin host and classifier commands, without screens, windows, account setup or OS shell commands.", + "webviews": ["agents"], + "platforms": ["macOS"], + "permissions": [ + "allow-identity-restore", + "allow-relay-sign", + "allow-relay-socket-connect", + "allow-relay-socket-start", + "allow-relay-socket-send", + "allow-relay-socket-close", + "allow-identity-prepare-remote-agent-authorization", + "allow-identity-sign-builderlab-binding", + "allow-relay-decode-read-state", + "allow-relay-sign-read-state", + "allow-relay-publish-read-state", + "allow-relay-http", + "allow-relay-workflow-runs", + "allow-relay-admin-discover", + "allow-relay-admin-request", + "allow-relay-admin-attachment", + "allow-relay-admin-save-attachment", + "allow-relay-project-git", + "allow-relay-project-git-cancel", + "allow-relay-channel-sign", + "allow-relay-channel-publish", + "allow-relay-kit-sign", + "allow-relay-kit-prepare", + "allow-relay-kit-decode", + "allow-relay-direct-message", + "allow-relay-decode-sidebar", + "allow-relay-sign-sidebar", + "allow-relay-decode-reminders", + "allow-relay-sign-reminder", + "allow-relay-agent-resolve", + "allow-relay-git-authorization", + "allow-relay-agent-log-proof", + "allow-relay-agent-observer", + "allow-relay-archive", + "allow-relay-agent-memories-read", + "allow-relay-agent-library", + "allow-relay-upload-begin", + "allow-relay-upload-chunk", + "allow-relay-upload", + "allow-relay-upload-cancel", + "allow-media-download", + "allow-media-stream-base", + "allow-get-os-idle-seconds", + "allow-repositories-folder-validate", + "allow-plugin-catalog", + "allow-plugin-change", + "allow-plugin-reload", + "allow-plugin-module", + "allow-plugin-recover", + "allow-plugin-host-run-command", + "allow-plugin-host-request", + "allow-plugin-host-upload", + "allow-plugin-host-process-spawn", + "allow-plugin-host-process-write", + "allow-plugin-host-process-kill", + "allow-plugin-host-mcp-serve", + "allow-plugin-host-mcp-reply", + "allow-plugin-host-mcp-close", + "allow-plugin-host-read-file", + "allow-plugin-host-resolve-directory", + "allow-oauth-callback-begin", + "allow-oauth-callback-wait", + "allow-oauth-callback-cancel", + "allow-app-agent-list", + "allow-app-agent-create", + "allow-app-agent-portable", + "allow-app-agent-port", + "allow-app-agent-rename", + "allow-app-agent-set-picture", + "allow-app-agent-set-type", + "allow-app-agent-delete", + "allow-app-agent-begin-delete", + "allow-app-agent-begin-leave", + "allow-app-agent-hold-removal", + "allow-app-agent-release-removal", + "allow-app-agent-join", + "allow-app-agent-forget", + "allow-app-agent-publish", + "allow-app-agent-publish-profile", + "allow-app-agent-query", + "allow-app-agent-search-profiles", + "allow-app-agent-upload", + "allow-app-agent-media", + "allow-app-agent-git-auth", + "allow-app-agent-remember", + "allow-app-agent-claim", + "allow-app-agent-release", + "allow-app-agent-set-protection", + "allow-app-agent-protection-defaults", + "allow-app-agent-set-protection-defaults", + "allow-jev-classifier-status", + "allow-jev-classifier-set-key", + "allow-jev-classifier-clear-key", + "allow-app-agent-state-load", + "allow-app-agent-state-set", + "allow-decide", + "core:app:allow-version", + "core:resources:allow-close", + "allow-agents-port-listen", + "allow-agents-port-send" + ] +} diff --git a/src-tauri/capabilities/default.json b/src-tauri/capabilities/default.json index a4e0139d4e..4232c06183 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -183,6 +183,8 @@ "core:window:allow-close", "core:window:allow-start-dragging", "core:window:allow-internal-toggle-maximize", + "allow-agents-port-listen", + "allow-agents-port-send", { "identifier": "opener:allow-open-url", "allow": [ diff --git a/src-tauri/src/agents_port.rs b/src-tauri/src/agents_port.rs new file mode 100644 index 0000000000..a0375d6507 --- /dev/null +++ b/src-tauri/src/agents_port.rs @@ -0,0 +1,119 @@ +//! The port between the screens' page and the page that runs agents. Each +//! page opens its end with a channel, and a frame one page sends goes to the +//! other's channel. Only those two webviews take part, and the sender is the +//! webview the frame came from, so neither page listens to app events and no +//! other page can speak for either. + +use std::{collections::HashMap, sync::Mutex}; + +use serde_json::Value; +use tauri::{ipc::Channel, Runtime, State, Webview}; + +const PAGES: [&str; 2] = ["main", "agents"]; + +/// The other page of the pair, for `label`. +fn peer(label: &str) -> Option<&'static str> { + match label { + "main" => Some("agents"), + "agents" => Some("main"), + _ => None, + } +} + +/// Each page's open end, by its webview's label. A page that loads again +/// opens a new one, which replaces the last. +#[derive(Default)] +pub(crate) struct AgentsPort(Mutex>>); + +impl AgentsPort { + fn open(&self, label: &str, frames: Channel) -> Result<(), String> { + let page = PAGES + .into_iter() + .find(|page| *page == label) + .ok_or("Only Buzz's own pages use the agents port")?; + self.0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert(page, frames); + Ok(()) + } + /// Sends `frame` from `label`'s page to its peer's. A peer that is not + /// open yet misses it; it says when it is, and the sender sends again. + fn send(&self, label: &str, frame: Value) -> Result<(), String> { + let to = peer(label).ok_or("Only Buzz's own pages use the agents port")?; + let channel = self + .0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(to) + .cloned(); + match channel { + Some(channel) => channel.send(frame).map_err(|error| error.to_string()), + None => Ok(()), + } + } +} + +#[tauri::command] +pub(crate) fn agents_port_listen( + webview: Webview, + port: State<'_, AgentsPort>, + frames: Channel, +) -> Result<(), String> { + port.open(webview.label(), frames) +} + +#[tauri::command] +pub(crate) fn agents_port_send( + webview: Webview, + port: State<'_, AgentsPort>, + frame: Value, +) -> Result<(), String> { + port.send(webview.label(), frame) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex as StdMutex}; + use tauri::ipc::InvokeResponseBody; + + fn end() -> (Channel, Arc>>) { + let heard = Arc::new(StdMutex::new(Vec::new())); + let into = heard.clone(); + let channel = Channel::new(move |body| { + if let InvokeResponseBody::Json(json) = body { + into.lock() + .unwrap() + .push(serde_json::from_str(&json).unwrap()); + } + Ok(()) + }); + (channel, heard) + } + + #[test] + fn frames_go_only_to_the_other_page_of_the_pair() { + let port = AgentsPort::default(); + let (main, to_main) = end(); + let (agents, to_agents) = end(); + // Before the other page is open, a frame for it is missed, not held. + port.send("main", Value::from("early")).unwrap(); + port.open("main", main).unwrap(); + port.open("agents", agents).unwrap(); + port.send("main", Value::from(1)).unwrap(); + port.send("agents", Value::from(2)).unwrap(); + assert_eq!(*to_agents.lock().unwrap(), vec![Value::from(1)]); + assert_eq!(*to_main.lock().unwrap(), vec![Value::from(2)]); + // A reloaded page's new end replaces its old one. + let (again, to_again) = end(); + port.open("agents", again).unwrap(); + port.send("main", Value::from(3)).unwrap(); + assert_eq!(*to_again.lock().unwrap(), vec![Value::from(3)]); + assert_eq!(to_agents.lock().unwrap().len(), 1); + // No other page opens an end or sends. + let (huddle, _) = end(); + assert!(port.open("huddle", huddle).is_err()); + assert!(port.send("huddle", Value::from(4)).is_err()); + } +} diff --git a/src-tauri/src/app_agents.rs b/src-tauri/src/app_agents.rs index 853312e24d..999d2604f2 100644 --- a/src-tauri/src/app_agents.rs +++ b/src-tauri/src/app_agents.rs @@ -2,6 +2,7 @@ //! and owner attestation never enter the WebView; it asks for one bounded //! event, read, upload or memory write at a time, and native signs and sends it //! to the agent's community. +use crate::page::{Page, PageLoads}; use buzz_agent_controller::{ inline_picture, AppAgent, AppAgents, Carry, Credentials, OldStores, PlatformCredentials, PortSource, Portables, ProtectionBinding, ProtectionDefaults, Published, Removal, Secret, @@ -114,11 +115,13 @@ impl AppAgentHost { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } - /// Lets go of the removals the page held, as a new page starts none. - pub(crate) fn page_load(&self) { + /// Lets go of the removals `webview`'s page held, as a new page starts none. + pub(crate) fn page_load(&self, webview: &str) { let mut removals = self.removals(); - removals.page += 1; - removals.held.clear(); + removals.pages.load(webview); + removals + .held + .retain(|_, (page, _)| page.webview() != webview); } fn keys(&self) -> std::sync::MutexGuard<'_, HashMap>> { self.keys @@ -225,13 +228,13 @@ impl AppAgentHost { /// will ever have. The row stays, marked deleted, until the WebView has /// cleaned up the community and calls `app_agent_forget`. This copy of the /// app stops running the agent once it is in no community. - async fn delete(&self, pubkey: String, relay: String) -> Result<(), String> { - let page = self.removals().page; + async fn delete(&self, webview: &str, pubkey: String, relay: String) -> Result<(), String> { + let page = self.removals().pages.current(webview); let turn = self.profile_turn(&pubkey); let _turn = turn.lock().await; // Its grant goes with its last community. Checked before the removal, // so a request in flight is refused rather than outliving the key. - if self.holds_removal(page, &pubkey, &relay) { + if self.holds_removal(&page, &pubkey, &relay) { self.tokens.forget_row(&pubkey, &relay)?; if let Some(owner) = self.last_community(&pubkey, &relay).await? { self.tokens.forget(&owner, &pubkey).await?; @@ -243,13 +246,13 @@ impl AppAgentHost { let mut keys = host.keys(); let agents = host.agents.clone()?; let removals = host.removals(); - if removals.page != page { + if !removals.pages.is_current(&page) { return Err("The app reloaded while it was being removed".into()); } - let removal = removals + let (_, removal) = removals .held .get(&pubkey) - .filter(|removal| removal.is(&pubkey, &relay)) + .filter(|(holder, removal)| *holder == page && removal.is(&pubkey, &relay)) .ok_or("Its removal from that community isn't held by this app")?; keys.remove(&pubkey); agents.remove(removal, host.credentials.as_ref())?; @@ -266,13 +269,13 @@ impl AppAgentHost { .await } /// Whether this page holds the agent's removal from the community at `relay`. - fn holds_removal(&self, page: u64, pubkey: &str, relay: &str) -> bool { + fn holds_removal(&self, page: &Page, pubkey: &str, relay: &str) -> bool { let removals = self.removals(); - removals.page == page + removals.pages.is_current(page) && removals .held .get(pubkey) - .is_some_and(|removal| removal.is(pubkey, relay)) + .is_some_and(|(holder, removal)| holder == page && removal.is(pubkey, relay)) } /// The agent's owner, if the community at `relay` is the last one where it /// has its key. @@ -451,9 +454,10 @@ fn open_lock(path: &Path) -> Result { /// `AppAgents::hold_removal`. #[derive(Default)] struct Removals { - /// Counts page loads, so a hold taken for an earlier page is let go. - page: u64, - held: HashMap, + /// So a hold taken for an earlier page is let go. + pages: PageLoads, + /// Each hold, with the page that took it. + held: HashMap, } /// Runs blocking file and credential work off the async runtime. @@ -748,12 +752,13 @@ pub(crate) async fn app_agent_join( } #[tauri::command] -pub(crate) async fn app_agent_delete( +pub(crate) async fn app_agent_delete( + webview: tauri::Webview, state: tauri::State<'_, AppAgentHost>, pubkey: String, relay: String, ) -> Result<(), String> { - state.delete(pubkey, relay).await + state.delete(webview.label(), pubkey, relay).await } #[tauri::command] @@ -779,21 +784,22 @@ pub(crate) async fn app_agent_begin_leave( /// Holds its removal from the community at `relay` for this copy of the app, /// until `app_agent_release_removal`; see `AppAgents::hold_removal`. #[tauri::command] -pub(crate) async fn app_agent_hold_removal( +pub(crate) async fn app_agent_hold_removal( + webview: tauri::Webview, state: tauri::State<'_, AppAgentHost>, pubkey: String, relay: String, ) -> Result<(), String> { let host = state.inner().clone(); - let page = host.removals().page; + let page = host.removals().pages.current(webview.label()); blocking(move || { let removal = host.agents.clone()?.hold_removal(&pubkey, &relay)?; let mut removals = host.removals(); // A page that loaded meanwhile holds none, so it is let go. - if removals.page != page { + if !removals.pages.is_current(&page) { return Err("The app reloaded while it was being removed".into()); } - removals.held.insert(pubkey, removal); + removals.held.insert(pubkey, (page, removal)); Ok(()) }) .await @@ -808,7 +814,7 @@ pub(crate) fn app_agent_release_removal( let held = &mut state.removals().held; if held .get(&pubkey) - .is_some_and(|removal| removal.is(&pubkey, &relay)) + .is_some_and(|(_, removal)| removal.is(&pubkey, &relay)) { held.remove(&pubkey); } @@ -1581,22 +1587,30 @@ mod tests { )); let agents = host.agents().unwrap(); host.keys().insert(agent.pubkey.clone(), Arc::new(key)); + let page = host.removals().pages.current("main"); host.removals().held.insert( agent.pubkey.clone(), - agents.hold_removal(&agent.pubkey, &agent.relay).unwrap(), + ( + page, + agents.hold_removal(&agent.pubkey, &agent.relay).unwrap(), + ), ); let turn = host.profile_turn(&agent.pubkey); let publishing = turn.lock().await; - let mut deleting = Box::pin(host.delete(agent.pubkey.clone(), agent.relay.clone())); + let mut deleting = Box::pin(host.delete("main", agent.pubkey.clone(), agent.relay.clone())); // Poll through the generation capture into the held publication lock. assert!(matches!( futures_util::poll!(&mut deleting), std::task::Poll::Pending )); - host.page_load(); + host.page_load("main"); + let page = host.removals().pages.current("main"); host.removals().held.insert( agent.pubkey.clone(), - agents.hold_removal(&agent.pubkey, &agent.relay).unwrap(), + ( + page, + agents.hold_removal(&agent.pubkey, &agent.relay).unwrap(), + ), ); drop(publishing); assert_eq!( @@ -1608,7 +1622,7 @@ mod tests { assert!(host.keys().contains_key(&agent.pubkey)); assert!(host.removals().held.contains_key(&agent.pubkey)); // Only the current page may finish cleanup using its own hold. - host.delete(agent.pubkey.clone(), agent.relay.clone()) + host.delete("main", agent.pubkey.clone(), agent.relay.clone()) .await .unwrap(); assert!(agents.list().unwrap()[0].deleted); @@ -1677,7 +1691,10 @@ mod tests { let agents = host.agents().unwrap(); agents.begin_delete(&agent.pubkey).unwrap(); let removal = agents.hold_removal(&agent.pubkey, &agent.relay).unwrap(); - host.removals().held.insert(agent.pubkey.clone(), removal); + let page = host.removals().pages.current("main"); + host.removals() + .held + .insert(agent.pubkey.clone(), (page, removal)); } #[tokio::test] @@ -1745,7 +1762,7 @@ mod tests { let deleting = tokio::spawn({ let host = host.clone(); let (pubkey, relay) = (agent.pubkey.clone(), agent.relay.clone()); - async move { host.delete(pubkey, relay).await } + async move { host.delete("main", pubkey, relay).await } }); crate::federated_identity::agent::tests::bounded(async { while !host.tokens.closed(&agent.pubkey) { @@ -1809,15 +1826,18 @@ mod tests { agents.begin_leave(&other.pubkey, &other.relay).unwrap(); let removal = agents.hold_removal(&other.pubkey, &other.relay).unwrap(); - host.removals().held.insert(other.pubkey.clone(), removal); - host.delete(other.pubkey.clone(), other.relay.clone()) + let page = host.removals().pages.current("main"); + host.removals() + .held + .insert(other.pubkey.clone(), (page, removal)); + host.delete("main", other.pubkey.clone(), other.relay.clone()) .await .unwrap(); assert!(!host.tokens.closed(&home.pubkey), "still in its home"); host.removals().held.remove(&other.pubkey); hold_removal(&host, &home); - host.delete(home.pubkey.clone(), home.relay.clone()) + host.delete("main", home.pubkey.clone(), home.relay.clone()) .await .unwrap(); assert!(host.tokens.closed(&home.pubkey)); @@ -1871,7 +1891,7 @@ mod tests { // deleted and the same key is imported again as import II. let (old_agent, old_key, old_incarnation) = import().await; hold_removal(&host, &old_agent); - host.delete(old_agent.pubkey.clone(), old_agent.relay.clone()) + host.delete("main", old_agent.pubkey.clone(), old_agent.relay.clone()) .await .unwrap(); let (agent, _, _) = import().await; @@ -2366,8 +2386,11 @@ mod tests { let agents = host.agents().unwrap(); agents.begin_leave(&old.pubkey, &old.relay).unwrap(); let removal = agents.hold_removal(&old.pubkey, &old.relay).unwrap(); - host.removals().held.insert(old.pubkey.clone(), removal); - host.delete(old.pubkey.clone(), old.relay.clone()) + let page = host.removals().pages.current("main"); + host.removals() + .held + .insert(old.pubkey.clone(), (page, removal)); + host.delete("main", old.pubkey.clone(), old.relay.clone()) .await .unwrap(); host.removals().held.remove(&old.pubkey); diff --git a/src-tauri/src/browser_permissions_tests.rs b/src-tauri/src/browser_permissions_tests.rs index 8e2957d6b7..318e89f0a0 100644 --- a/src-tauri/src/browser_permissions_tests.rs +++ b/src-tauri/src/browser_permissions_tests.rs @@ -70,6 +70,10 @@ fn native_command_permissions_isolate_webviews() { let huddle = WebviewWindowBuilder::new(&app, "huddle", WebviewUrl::default()) .build() .unwrap(); + // The hidden page that runs agents. + let agents = WebviewWindowBuilder::new(&app, "agents", WebviewUrl::default()) + .build() + .unwrap(); let application_commands = [ "federated_identity_snapshot", @@ -137,6 +141,8 @@ fn native_command_permissions_isolate_webviews() { "clipboard_write_text", "media_stream_base", "get_os_idle_seconds", + "agents_port_listen", + "agents_port_send", "plugin_import_folder", "plugin_update_check", "plugin_update_preview", @@ -317,6 +323,7 @@ fn native_command_permissions_isolate_webviews() { invoke(main.as_ref(), command, local_origin).is_ok(), "{command}" ); + assert!(invoke(agents.as_ref(), command, "https://example.org").is_err()); assert_eq!( invoke(huddle.as_ref(), command, local_origin).is_ok(), matches!(command, "media_download" | "clipboard_write_text"), @@ -335,6 +342,38 @@ fn native_command_permissions_isolate_webviews() { ); } assert!(invoke(main.as_ref(), "plugin:window|close", local_origin).is_ok()); + // The hidden page that runs agents and their plugin hosts gets none of + // the screens, windows, account setup or OS shell commands. + #[cfg(target_os = "macos")] + for command in [ + "relay_sign", + "app_agent_list", + "plugin_host_run_command", + "decide", + "get_os_idle_seconds", + "agents_port_listen", + "agents_port_send", + ] { + assert!( + invoke(agents.as_ref(), command, local_origin).is_ok(), + "{command}" + ); + } + for command in [ + "identity_create", + "nsec_restore_begin", + "notification_show", + "terminal_spawn", + "browser_attach", + "plugin_import_folder", + "plugin:window|close", + "plugin:event|listen", + ] { + assert!( + invoke(agents.as_ref(), command, local_origin).is_err(), + "{command}" + ); + } #[cfg(all(feature = "native-mcp", target_os = "macos"))] { diff --git a/src-tauri/src/federated_identity/socket.rs b/src-tauri/src/federated_identity/socket.rs index e3a1cdfe91..83536a08b8 100644 --- a/src-tauri/src/federated_identity/socket.rs +++ b/src-tauri/src/federated_identity/socket.rs @@ -16,24 +16,46 @@ use tokio_tungstenite::tungstenite::{ use tokio_util::sync::CancellationToken; struct Connection { + /// The webview whose page opened it, and alone uses it. + webview: String, stop: CancellationToken, sender: mpsc::Sender, start: Option>, } +/// Each page's sockets: a webview that loads another page closes its own, +/// and leaves another webview's open. #[derive(Default)] pub(crate) struct Sockets(Arc>>); impl Sockets { pub(crate) fn clear(&self) { - for (_, connection) in self - .0 + self.close(|_| true); + } + /// `webview` started loading another page. + pub(crate) fn clear_page(&self, webview: &str) { + self.close(|connection| connection.webview == webview); + } + fn close(&self, closes: impl Fn(&Connection) -> bool) { + self.0 .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .drain() - { - connection.stop.cancel(); - } + .retain(|_, connection| { + let closed = closes(connection); + if closed { + connection.stop.cancel(); + } + !closed + }); } } +/// `id`'s connection, if `webview` opened it. +fn owned<'a>( + live: &'a mut HashMap, + id: &str, + webview: &str, +) -> Option<&'a mut Connection> { + live.get_mut(id) + .filter(|connection| connection.webview == webview) +} #[derive(Clone, Serialize)] #[serde(tag = "type", rename_all = "kebab-case")] pub(crate) enum SocketEvent { @@ -59,7 +81,8 @@ pub(crate) enum SocketMode { } #[tauri::command] -pub(crate) async fn relay_socket_connect( +pub(crate) async fn relay_socket_connect( + webview: tauri::Webview, sockets: tauri::State<'_, Sockets>, fi: tauri::State<'_, FederatedIdentity>, identity: tauri::State<'_, IdentityHost>, @@ -77,12 +100,18 @@ pub(crate) async fn relay_socket_connect( .0 .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if live.len() >= 16 || live.contains_key(&id) { + let page = webview.label(); + let open = live + .values() + .filter(|connection| connection.webview == page) + .count(); + if open >= 16 || live.contains_key(&id) { return Err("Native sockets are busy".into()); } live.insert( id.clone(), Connection { + webview: page.into(), stop: stop.clone(), sender, start: Some(start), @@ -198,22 +227,27 @@ pub(crate) async fn relay_socket_connect( Ok(SocketMode::Native) } #[tauri::command] -pub(crate) fn relay_socket_start( +pub(crate) fn relay_socket_start( + webview: tauri::Webview, sockets: tauri::State<'_, Sockets>, id: String, ) -> Result<(), String> { - sockets - .0 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get_mut(&id) - .and_then(|entry| entry.start.take()) - .ok_or("Native socket retired")? - .send(()) - .map_err(|_| "Native socket retired".into()) + owned( + &mut sockets + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + &id, + webview.label(), + ) + .and_then(|entry| entry.start.take()) + .ok_or("Native socket retired")? + .send(()) + .map_err(|_| "Native socket retired".into()) } #[tauri::command] -pub(crate) fn relay_socket_send( +pub(crate) fn relay_socket_send( + webview: tauri::Webview, sockets: tauri::State<'_, Sockets>, id: String, data: String, @@ -221,29 +255,35 @@ pub(crate) fn relay_socket_send( if data.len() > 1024 * 1024 { return Err("Socket frame too large".into()); } - sockets - .0 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(&id) - .ok_or("Native socket retired")? - .sender - .try_send(data) - .map_err(|_| "Socket send queue is full".into()) + owned( + &mut sockets + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + &id, + webview.label(), + ) + .ok_or("Native socket retired")? + .sender + .try_send(data) + .map_err(|_| "Socket send queue is full".into()) } #[tauri::command] -pub(crate) fn relay_socket_close( +pub(crate) fn relay_socket_close( + webview: tauri::Webview, sockets: tauri::State<'_, Sockets>, fi: tauri::State<'_, FederatedIdentity>, id: String, ) { - fi.close_discovery(&id); - if let Some(entry) = sockets + let mut live = sockets .0 .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(&id) - { + .unwrap_or_else(std::sync::PoisonError::into_inner); + if owned(&mut live, &id, webview.label()).is_none() { + return; + } + fi.close_discovery(&id); + if let Some(entry) = live.remove(&id) { entry.stop.cancel(); } } @@ -555,4 +595,35 @@ mod tests { ); community.assert_uncarried(&server.heads().await[0]); } + + /// A webview that loads another page closes its own sockets, and leaves + /// another webview's open; only the webview that opened one reaches it. + #[test] + fn a_page_load_closes_only_its_own_webviews_sockets() { + let sockets = Sockets::default(); + let open = |id: &str, webview: &str| { + let stop = CancellationToken::new(); + let (sender, _) = mpsc::channel(1); + sockets.0.lock().unwrap().insert( + id.into(), + Connection { + webview: webview.into(), + stop: stop.clone(), + sender, + start: None, + }, + ); + stop + }; + let screens = open("a", "main"); + let agents = open("b", "agents"); + assert!(owned(&mut sockets.0.lock().unwrap(), "b", "main").is_none()); + assert!(owned(&mut sockets.0.lock().unwrap(), "b", "agents").is_some()); + sockets.clear_page("main"); + assert!(screens.is_cancelled()); + assert!(!agents.is_cancelled()); + assert!(sockets.0.lock().unwrap().contains_key("b")); + sockets.clear(); + assert!(agents.is_cancelled()); + } } diff --git a/src-tauri/src/host_mcp.rs b/src-tauri/src/host_mcp.rs index c06ff038a2..083bacc0a3 100644 --- a/src-tauri/src/host_mcp.rs +++ b/src-tauri/src/host_mcp.rs @@ -28,7 +28,9 @@ //! - Each connection answers one request and closes. A client that closes //! before its answer drops the request; the page's late reply is refused. //! - A server belongs to the plugin and the page that started it. It stops -//! when the plugin closes it, when the page reloads, or when the app exits. +//! when the plugin closes it, when that webview loads another page, or when +//! the app exits. +use crate::page::{Page, PageLoads}; use crate::{with_manager, PluginManager}; use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; use http_body_util::{BodyExt, Full, Limited}; @@ -111,6 +113,7 @@ impl Shared { struct Entry { plugin: String, + page: Page, shared: Arc, } @@ -121,9 +124,9 @@ pub(crate) struct HostMcpServers { #[derive(Default)] struct Registry { next: u64, - /// Counts page loads, so a server the page asked for before it reloaded - /// is refused rather than left running with nobody answering. - page: u64, + /// So a server the page asked for before it reloaded is refused rather + /// than left running with nobody answering. + pages: PageLoads, entries: HashMap, } impl HostMcpServers { @@ -136,21 +139,21 @@ impl HostMcpServers { _ => Err("No such MCP server".into()), } } - /// The current page load, for `serve`. - pub(crate) fn page(&self) -> u64 { - self.registry().page + /// The page `webview` shows now, for `serve`. + pub(crate) fn page(&self, webview: &str) -> Page { + self.registry().pages.current(webview) } /// Starts a server owned by `plugin` whose messages go to `events`, unless /// the page that asked during `page` has since gone. fn serve( &self, - page: u64, + page: Page, plugin: String, events: Channel, shared_session: bool, ) -> Result { let mut registry = self.registry(); - if registry.page != page { + if !registry.pages.is_current(&page) { return Err("The page reloaded".into()); } if registry.entries.len() >= MAX_SERVERS { @@ -183,6 +186,7 @@ impl HostMcpServers { handle, Entry { plugin, + page, shared: shared.clone(), }, ); @@ -223,10 +227,21 @@ impl HostMcpServers { self.owned(plugin, handle)?.stop.cancel(); Ok(()) } - /// Stops every server: the page that answered them is gone. + /// Stops every server `webview`'s page started: the page that answered + /// them is gone. + pub(crate) fn page_load(&self, webview: &str) { + let mut registry = self.registry(); + registry.pages.load(webview); + for entry in registry.entries.values() { + if entry.page.webview() == webview { + entry.shared.stop.cancel(); + } + } + } + /// Stops every server: the app is exiting. pub(crate) fn stop_all(&self) { let mut registry = self.registry(); - registry.page += 1; + registry.pages.exit(); for entry in registry.entries.values() { entry.shared.stop.cancel(); } @@ -489,7 +504,8 @@ fn same(a: &[u8], b: &[u8]) -> bool { /// Starts a loopback MCP server for a plugin that declares `host.mcp`. Each /// message arrives on `on_event` and is answered with `plugin_host_mcp_reply`. #[tauri::command] -pub(crate) async fn plugin_host_mcp_serve( +pub(crate) async fn plugin_host_mcp_serve( + webview: tauri::Webview, manager: tauri::State<'_, PluginManager>, servers: tauri::State<'_, HostMcpServers>, id: String, @@ -497,7 +513,7 @@ pub(crate) async fn plugin_host_mcp_serve( on_event: Channel, shared_session: Option, ) -> Result { - let page = servers.page(); + let page = servers.page(webview.label()); let plugin = id.clone(); let granted = with_manager(manager, move |manager| { Ok(manager.host_grants(&id, &revision)?.mcp) diff --git a/src-tauri/src/host_mcp/tests.rs b/src-tauri/src/host_mcp/tests.rs index 88fd5afe73..29d282189a 100644 --- a/src-tauri/src/host_mcp/tests.rs +++ b/src-tauri/src/host_mcp/tests.rs @@ -59,7 +59,7 @@ async fn answers_a_request_with_the_page_reply() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); assert!(info.url.starts_with("http://127.0.0.1:")); let named = initialize(&servers, &info, &mut events).await; @@ -110,7 +110,7 @@ async fn accepts_a_notification_without_a_body() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); let named = initialize(&servers, &info, &mut events).await; let request = post( @@ -189,7 +189,7 @@ async fn admits_only_the_session_that_initialized() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); // Before `initialize`, discovery needs no session and joins none; a tool // call is refused. @@ -226,7 +226,7 @@ async fn stops_when_a_second_client_initializes() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); initialize(&servers, &info, &mut events).await; // Whoever claimed the session, two clients hold the token. @@ -240,7 +240,7 @@ async fn child_initialization_keeps_parent_callbacks_alive() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, true) + .serve(servers.page("main"), "plugin".into(), channel, true) .unwrap(); let parent = initialize(&servers, &info, &mut events).await; let request = post(&info, &info.token, "", INITIALIZE); @@ -306,7 +306,7 @@ async fn releases_a_claim_that_did_not_succeed() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); let claim = || claim(&servers, &info); // The page answered with an error. @@ -350,7 +350,7 @@ async fn keeps_the_session_across_requests() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); let named = initialize(&servers, &info, &mut events).await; let call = r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#; @@ -371,7 +371,7 @@ async fn holds_the_claim_while_initialize_waits() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); let request = post(&info, &info.token, "", INITIALIZE); let client = tokio::spawn({ @@ -406,7 +406,7 @@ async fn refuses_without_starting_oauth() { let servers = HostMcpServers::default(); let (channel, mut events) = channel(); let info = servers - .serve(servers.page(), "plugin".into(), channel, false) + .serve(servers.page("main"), "plugin".into(), channel, false) .unwrap(); let body = r#"{"jsonrpc":"2.0","id":1,"method":"ping"}"#; // A wrong token and a web page are both told there is nothing here. @@ -429,9 +429,18 @@ async fn refuses_without_starting_oauth() { async fn stops_with_the_page() { let servers = HostMcpServers::default(); let (events_channel, mut events) = channel(); - let page = servers.page(); + let page = servers.page("main"); let info = servers - .serve(page, "plugin".into(), events_channel, false) + .serve(page.clone(), "plugin".into(), events_channel, false) + .unwrap(); + let (other_channel, _other_events) = channel(); + let other = servers + .serve( + servers.page("agents"), + "plugin".into(), + other_channel, + false, + ) .unwrap(); let request = post( &info, @@ -444,11 +453,17 @@ async fn stops_with_the_page() { async move { send(&info, request).await } }); assert_eq!(next(&mut events).await["type"], "message"); - servers.stop_all(); + servers.page_load("main"); assert_eq!(next(&mut events).await["type"], "closed"); // The request waiting on the page is dropped, not left hanging. assert!(!client.await.unwrap().contains("200 OK")); assert!(servers.reply("plugin", info.handle, 0, None).is_err()); let (late, _events) = channel(); assert!(servers.serve(page, "plugin".into(), late, false).is_err()); + // Another webview's server stays up. + assert!(!servers + .owned("plugin", other.handle) + .unwrap() + .stop + .is_cancelled()); } diff --git a/src-tauri/src/host_process.rs b/src-tauri/src/host_process.rs index e19c91b860..07615fee70 100644 --- a/src-tauri/src/host_process.rs +++ b/src-tauri/src/host_process.rs @@ -1,15 +1,17 @@ //! Long-lived processes a plugin starts from its manifest's `host.processes`: //! stdin stays open for the plugin, and stdout and stderr stream back as they //! arrive. A process belongs to the plugin that started it and to the page that -//! asked; it is killed when the plugin asks, when the page reloads, or when the -//! app exits. A page that reloaded hears nothing of it after. It runs with the -//! user's own access, with optional native containment for protected workers. +//! asked; it is killed when the plugin asks, when that webview loads another +//! page, or when the app exits. A page that reloaded hears nothing of it after. +//! It runs with the user's own access, with optional native containment for +//! protected workers. mod artifact; mod protection; use artifact::verify_artifact; use protection::ProcessProtection; use crate::host_command::{effective_path, load_login_shell_path, resolve_program}; +use crate::page::{Page, PageLoads}; use crate::{with_manager, PluginManager}; use buzzodz_plugins::HostProcess; use serde::{Deserialize, Serialize}; @@ -65,6 +67,7 @@ pub(crate) enum ProcessEvent { struct Entry { plugin: String, + page: Page, /// Its process group, which is its own pid. #[cfg_attr(not(unix), allow(dead_code))] group: Option, @@ -79,9 +82,9 @@ pub(crate) struct HostProcesses { #[derive(Default)] struct Registry { next: u64, - /// Counts page loads, so a spawn the page asked for before it reloaded is - /// refused rather than left running with nobody listening. - page: u64, + /// So a spawn the page asked for before it reloaded is refused rather + /// than left running with nobody listening. + pages: PageLoads, entries: HashMap, } impl HostProcesses { @@ -102,15 +105,15 @@ impl HostProcesses { _ => Err("No such process".into()), } } - /// The current page load, for `start`. - pub(crate) fn page(&self) -> u64 { - self.registry().page + /// The page `webview` shows now, for `start`. + pub(crate) fn page(&self, webview: &str) -> Page { + self.registry().pages.current(webview) } /// Starts `command`, owned by `plugin`, and streams its output to `on_event`, /// unless the page that asked during `page` has since gone. fn start( &self, - page: u64, + page: Page, plugin: String, program: &str, mut command: Command, @@ -118,7 +121,7 @@ impl HostProcesses { ) -> Result { let state = self.clone(); let mut registry = self.registry(); - if registry.page != page { + if !registry.pages.is_current(&page) { return Err("The page reloaded".into()); } if registry.entries.len() >= MAX_PROCESSES { @@ -143,6 +146,7 @@ impl HostProcesses { handle, Entry { plugin, + page: page.clone(), group: child.id(), stdin: child .stdin @@ -210,7 +214,7 @@ impl HostProcesses { registry.entries.remove(&handle); // A page that reloaded is told nothing: it stopped the process by // going, and may run until it unloads, taking the exit for a failure. - let reloaded = registry.page != page; + let reloaded = !registry.pages.is_current(&page); drop(registry); if !reloaded { let _ = on_event.send(ProcessEvent::Exit { @@ -256,22 +260,21 @@ impl HostProcesses { } Ok(()) } - /// Stops every process, without telling the page: it is gone. - pub(crate) fn stop_all(&self) { + /// Stops every process `webview`'s page started, without telling the + /// page: it is gone. + pub(crate) fn page_load(&self, webview: &str) { let mut registry = self.registry(); - registry.page += 1; - for entry in registry.entries.values_mut() { - entry.stdin = None; - if let Some(stop) = entry.stop.take() { - let _ = stop.send(()); - } - } + registry.pages.load(webview); + registry.stop(|page| page.webview() == webview); } /// Kills every process group at once: the app is exiting, and its async /// tasks will not run again to stop them. On Windows each process's job /// kills it when the app's handle to the job closes. pub(crate) fn shutdown(&self) { - self.stop_all(); + let mut registry = self.registry(); + registry.pages.exit(); + registry.stop(|_| true); + drop(registry); #[cfg(unix)] for group in self .registry() @@ -283,6 +286,17 @@ impl HostProcesses { } } } +impl Registry { + /// Stops the processes of each page `gone` picks. + fn stop(&mut self, gone: impl Fn(&Page) -> bool) { + for entry in self.entries.values_mut().filter(|entry| gone(&entry.page)) { + entry.stdin = None; + if let Some(stop) = entry.stop.take() { + let _ = stop.send(()); + } + } + } +} #[derive(Clone, Copy)] enum Stream { @@ -297,6 +311,7 @@ enum Stream { #[allow(clippy::too_many_arguments)] pub(crate) async fn plugin_host_process_spawn( app: tauri::AppHandle, + webview: tauri::Webview, manager: tauri::State<'_, PluginManager>, processes: tauri::State<'_, HostProcesses>, id: String, @@ -309,7 +324,7 @@ pub(crate) async fn plugin_host_process_spawn( mcp: Option, protection: Option, ) -> Result { - let page = processes.page(); + let page = processes.page(webview.label()); let plugin = id.clone(); let (declared, registry) = with_manager(manager, move |manager| { let registry = manager.storage_root().to_path_buf(); diff --git a/src-tauri/src/host_process/tests.rs b/src-tauri/src/host_process/tests.rs index a40742b03d..c978925b48 100644 --- a/src-tauri/src/host_process/tests.rs +++ b/src-tauri/src/host_process/tests.rs @@ -58,7 +58,13 @@ async fn streams_stdin_to_stdout_until_input_closes() { let processes = HostProcesses::default(); let (channel, mut events) = channel(); let handle = processes - .start(0, "a.plugin".into(), "cat", command("cat", &[]), channel) + .start( + processes.page("main"), + "a.plugin".into(), + "cat", + command("cat", &[]), + channel, + ) .unwrap(); processes .write("a.plugin", handle, "one\n".into(), false) @@ -87,7 +93,7 @@ async fn reports_stderr_and_exit_code() { let (channel, mut events) = channel(); processes .start( - 0, + processes.page("main"), "a.plugin".into(), "sh", command("sh", &["-c", "echo oops >&2; exit 3"]), @@ -105,7 +111,13 @@ async fn only_the_owning_plugin_can_write_or_kill() { let processes = HostProcesses::default(); let (channel, mut events) = channel(); let handle = processes - .start(0, "a.plugin".into(), "cat", command("cat", &[]), channel) + .start( + processes.page("main"), + "a.plugin".into(), + "cat", + command("cat", &[]), + channel, + ) .unwrap(); assert!(processes .write("b.plugin", handle, "x".into(), false) @@ -125,7 +137,7 @@ async fn kill_ends_the_process_and_its_descendants() { // The shell ignores the polite signal; its child would outlive it. let handle = processes .start( - 0, + processes.page("main"), "a.plugin".into(), "sh", command("sh", &["-c", "trap '' TERM; sleep 30 & echo $!; wait"]), @@ -146,17 +158,29 @@ async fn kill_ends_the_process_and_its_descendants() { #[cfg(unix)] #[tokio::test] -async fn stop_all_ends_every_process() { +async fn a_page_load_ends_every_process_that_page_started() { let processes = HostProcesses::default(); let (first, mut first_events) = channel(); let (second, mut second_events) = channel(); processes - .start(0, "a.plugin".into(), "cat", command("cat", &[]), first) + .start( + processes.page("main"), + "a.plugin".into(), + "cat", + command("cat", &[]), + first, + ) .unwrap(); processes - .start(0, "b.plugin".into(), "cat", command("cat", &[]), second) + .start( + processes.page("main"), + "b.plugin".into(), + "cat", + command("cat", &[]), + second, + ) .unwrap(); - processes.stop_all(); + processes.page_load("main"); // The page that reloaded hears no exit: each channel closes without one. for events in [&mut first_events, &mut second_events] { while let Some(event) = tokio::time::timeout(Duration::from_secs(10), events.recv()) @@ -169,6 +193,39 @@ async fn stop_all_ends_every_process() { assert!(processes.registry().entries.is_empty()); } +#[cfg(unix)] +#[tokio::test] +async fn another_webviews_page_load_leaves_the_process_running() { + let processes = HostProcesses::default(); + let (channel, mut events) = channel(); + let handle = processes + .start( + processes.page("agents"), + "a.plugin".into(), + "cat", + command("cat", &[]), + channel, + ) + .unwrap(); + processes.page_load("main"); + processes + .write("a.plugin", handle, "still here\n".into(), false) + .await + .unwrap(); + let echoed = tokio::time::timeout(Duration::from_secs(10), events.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(echoed["data"], "still here\n"); + processes.page_load("agents"); + while let Some(event) = tokio::time::timeout(Duration::from_secs(10), events.recv()) + .await + .expect("the channel stays open after the process stops") + { + assert_ne!(event["type"], "exit"); + } +} + #[cfg(unix)] #[tokio::test] async fn a_page_that_stops_listening_stops_the_process() { @@ -176,7 +233,7 @@ async fn a_page_that_stops_listening_stops_the_process() { let channel = Channel::new(|_| Err(tauri::Error::FailedToReceiveMessage)); let handle = processes .start( - 0, + processes.page("main"), "a.plugin".into(), "sh", command("sh", &["-c", "echo hi; sleep 30"]), @@ -196,8 +253,8 @@ async fn a_page_that_stops_listening_stops_the_process() { #[tokio::test] async fn a_spawn_from_before_a_reload_is_refused() { let processes = HostProcesses::default(); - let page = processes.page(); - processes.stop_all(); + let page = processes.page("main"); + processes.page_load("main"); let (channel, _events) = channel(); let refused = processes.start(page, "a.plugin".into(), "cat", command("cat", &[]), channel); assert_eq!(refused.err().as_deref(), Some("The page reloaded")); @@ -211,7 +268,7 @@ async fn a_process_may_close_its_output_and_carry_on() { let (channel, mut events) = channel(); processes .start( - 0, + processes.page("main"), "a.plugin".into(), "sh", command("sh", &["-c", "exec >&- 2>&-; sleep 1; exit 4"]), @@ -229,7 +286,7 @@ async fn shutdown_kills_every_group_without_waiting() { let (channel, mut events) = channel(); processes .start( - 0, + processes.page("main"), "a.plugin".into(), "sh", command("sh", &["-c", "trap '' TERM; sleep 30 & echo $!; wait"]), diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index f0fe3cc9e5..f10208fa21 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1,3 +1,4 @@ +mod agents_port; mod archive; #[cfg(test)] mod build_resources; @@ -63,6 +64,7 @@ mod microphone; mod notifications; mod os_idle; +mod page; mod repositories_folder; mod window_controls; mod window_state; @@ -686,6 +688,8 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen clipboard::clipboard_write_text, media_stream_base, get_os_idle_seconds, + agents_port::agents_port_listen, + agents_port::agents_port_send, workspace_pick_folder, repositories_folder_validate, legacy_communities, @@ -971,6 +975,11 @@ pub fn run() { !tauri::is_dev() && app.config().identifier == sign_out::legacy_identifier::CURRENT, tokens, )); + // After the agent state it uses: its page may call it at once. + #[cfg(target_os = "macos")] + if let Err(error) = agents_window(app.handle()) { + eprintln!("Could not open the agents page: {error}"); + } Ok(()) }); #[cfg(not(target_os = "macos"))] @@ -1013,6 +1022,7 @@ pub fn run() { .manage(decisions::DecisionsHost::new()) .manage(pairing::Pairing::default()) .manage(Huddles::default()) + .manage(agents_port::AgentsPort::default()) .manage(HuddleWindow::default()) .manage(relay::Uploads::default()) .register_asynchronous_uri_scheme_protocol("buzz-media", relay::media_protocol) @@ -1055,24 +1065,26 @@ pub fn run() { { eprintln!("OAuth callback cleanup failed: {error}"); } + if matches!(payload.event(), tauri::webview::PageLoadEvent::Started) { + // Processes belong to the page that started them, in its own webview. + webview.state::().page_load(webview.label()); + webview.state::().page_load(webview.label()); + webview.state::().page_load(webview.label()); + webview.state::().cancel_page(&webview.state::(), webview.label()); + webview.state::().clear_page(webview.label()); + } if webview.label() == "main" && matches!(payload.event(), tauri::webview::PageLoadEvent::Started) { webview.state::().cancel_all(); webview.state::().document_load(); - webview.state::().cancel_all(&webview.state::()); - // Processes belong to the page that started them. webview.state::().stop_all(); webview.state::().clear(); - webview.state::().clear(); webview.state::().clear_discoveries(); - webview.state::().stop_all(); - webview.state::().stop_all(); - webview.state::().page_load(); } browser::page_load(webview, payload); }) .on_window_event(|window, event| { if window.label() == "main" && matches!(event, tauri::WindowEvent::Destroyed | tauri::WindowEvent::CloseRequested { .. }) { window.state::().cancel_all(); } - if window.label() == "main" && matches!(event, tauri::WindowEvent::Destroyed) { window.state::().cancel_all(&window.state::()); } + if matches!(event, tauri::WindowEvent::Destroyed) { window.state::().cancel_page(&window.state::(), window.label()); } #[cfg(target_os = "macos")] if window.label() == "main" { if let tauri::WindowEvent::CloseRequested { api, .. } = event { @@ -1101,6 +1113,20 @@ pub fn run() { }); } +/// The hidden page that runs agents apart from the screens, so a main window +/// that is hidden or busy cannot hold them up. It never shows and is never +/// throttled, and lives until Quit. The screens' page expects it on macOS +/// (`agentsRunApart` in src/features/agents/apart.ts). +#[cfg(target_os = "macos")] +fn agents_window(app: &tauri::AppHandle) -> tauri::Result<()> { + tauri::WebviewWindowBuilder::new(app, "agents", tauri::WebviewUrl::App("agents.html".into())) + .title("Buzz agents") + .visible(false) + .background_throttling(tauri::utils::config::BackgroundThrottlingPolicy::Disabled) + .build() + .map(|_| ()) +} + /// Best-effort native teardown when Buzz exits, from Quit or a fenced sign-out. pub(crate) fn shut_down(app: &tauri::AppHandle) { app.state::().close(); diff --git a/src-tauri/src/page.rs b/src-tauri/src/page.rs new file mode 100644 index 0000000000..9d3c4896f6 --- /dev/null +++ b/src-tauri/src/page.rs @@ -0,0 +1,62 @@ +//! Each webview's page loads. Work a page starts belongs to that load, in that +//! webview: it stops when the same webview loads a page again, and a request +//! the page made before it reloaded is refused. Another webview's load leaves +//! it alone. +use std::collections::HashMap; + +/// One page load in one webview. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct Page { + webview: String, + load: u64, +} +impl Page { + pub(crate) fn webview(&self) -> &str { + &self.webview + } +} + +/// Counts each webview's page loads. +#[derive(Default)] +pub(crate) struct PageLoads { + loads: HashMap, + exiting: bool, +} +impl PageLoads { + /// The page `webview` shows now. + pub(crate) fn current(&self, webview: &str) -> Page { + Page { + webview: webview.into(), + load: self.loads.get(webview).copied().unwrap_or_default(), + } + } + pub(crate) fn is_current(&self, page: &Page) -> bool { + !self.exiting && *page == self.current(&page.webview) + } + /// `webview` started loading another page. + pub(crate) fn load(&mut self, webview: &str) { + *self.loads.entry(webview.into()).or_default() += 1; + } + /// The app is exiting: no page is current any more. + pub(crate) fn exit(&mut self) { + self.exiting = true; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_load_ends_only_its_own_webviews_page() { + let mut pages = PageLoads::default(); + let main = pages.current("main"); + let agents = pages.current("agents"); + pages.load("main"); + assert!(!pages.is_current(&main)); + assert!(pages.is_current(&agents)); + assert!(pages.is_current(&pages.current("main"))); + pages.exit(); + assert!(!pages.is_current(&agents)); + } +} diff --git a/src-tauri/src/relay.rs b/src-tauri/src/relay.rs index e4924ffc85..0a9d3aadfb 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -1044,6 +1044,8 @@ const PENDING_CANCEL_LIFETIME: Duration = Duration::from_secs(60); struct UploadState { active: HashMap>, pending: HashMap, + /// The webview whose page started each upload that has one. + owners: HashMap, } #[derive(Default)] pub(crate) struct Uploads(std::sync::Mutex); @@ -1072,8 +1074,14 @@ impl Uploads { Ok(Some(receiver)) } + /// Records that `webview`'s page started `id`, so its next load stops it. + fn own(&self, id: &str, webview: &str) { + self.lock().owners.insert(id.into(), webview.into()); + } + fn cancel(&self, id: &str) { let mut uploads = self.lock(); + uploads.owners.remove(id); if let Some(sender) = uploads.active.remove(id) { let _ = sender.send(()); return; @@ -1098,13 +1106,35 @@ impl Uploads { } pub(crate) fn cancel_all(&self) { - for (_, sender) in self.lock().active.drain() { + let mut uploads = self.lock(); + uploads.owners.clear(); + for (_, sender) in uploads.active.drain() { let _ = sender.send(()); } } + /// Stops the uploads `webview`'s page started; returns their IDs. + fn cancel_page(&self, webview: &str) -> Vec { + let mut uploads = self.lock(); + let ids: Vec = uploads + .owners + .iter() + .filter(|(_, owner)| *owner == webview) + .map(|(id, _)| id.clone()) + .collect(); + for id in &ids { + uploads.owners.remove(id); + if let Some(sender) = uploads.active.remove(id) { + let _ = sender.send(()); + } + } + ids + } + fn finish(&self, id: &str) { - self.lock().active.remove(id); + let mut uploads = self.lock(); + uploads.owners.remove(id); + uploads.active.remove(id); } } @@ -1121,13 +1151,15 @@ fn upload_id(value: Option<&str>) -> Result<&str> { #[tauri::command] pub(crate) async fn relay_upload_begin( app: tauri::AppHandle, + webview: tauri::Webview, id: String, size: usize, ) -> Result<()> { use tauri::Manager as _; + let page = webview.label().to_owned(); tauri::async_runtime::spawn_blocking(move || { app.state::() - .begin(&app.state::(), &id, size) + .begin(&app.state::(), &page, &id, size) }) .await .map_err(|_| "Upload receiving could not complete")? diff --git a/src-tauri/src/relay/tests.rs b/src-tauri/src/relay/tests.rs index 440f3b0bdd..4dc1f8306c 100644 --- a/src-tauri/src/relay/tests.rs +++ b/src-tauri/src/relay/tests.rs @@ -3462,7 +3462,7 @@ async fn spool_processing_cleans_after_upstream_and_conversion_failures() { let parent = tempfile::tempdir().unwrap(); let spools = Spools::new(Ok(parent.path().to_owned())); let uploads = Uploads::default(); - spools.begin(&uploads, "failure", 3).unwrap(); + spools.begin(&uploads, "main", "failure", 3).unwrap(); spools.append("failure", 0, b"bad").unwrap(); let (spool, mut cancel) = spools.take("failure").unwrap(); let path = spool.directory.path().to_owned(); @@ -3486,7 +3486,7 @@ async fn spool_processing_cleans_after_upstream_and_conversion_failures() { assert!(result.is_err()); } assert!(!path.exists()); - assert!(spools.begin(&uploads, "next", 3).is_ok()); + assert!(spools.begin(&uploads, "main", "next", 3).is_ok()); } } @@ -3495,7 +3495,7 @@ async fn cancelled_spool_processing_never_starts_upstream_and_cleans() { let parent = tempfile::tempdir().unwrap(); let spools = Spools::new(Ok(parent.path().to_owned())); let uploads = Uploads::default(); - spools.begin(&uploads, "cancel", 3).unwrap(); + spools.begin(&uploads, "main", "cancel", 3).unwrap(); spools.append("cancel", 0, b"abc").unwrap(); let (spool, mut cancel) = spools.take("cancel").unwrap(); let path = spool.directory.path().to_owned(); @@ -3546,7 +3546,9 @@ async fn converted_spool_streams_signed_output_then_removes_both_files() { let bytes = std::fs::read(&input).unwrap(); let spools = Spools::new(Ok(parent.path().join("spools"))); let uploads = Uploads::default(); - spools.begin(&uploads, "converted", bytes.len()).unwrap(); + spools + .begin(&uploads, "main", "converted", bytes.len()) + .unwrap(); for (index, chunk) in bytes.chunks(UPLOAD_CHUNK).enumerate() { spools .append("converted", index * UPLOAD_CHUNK, chunk) @@ -3612,7 +3614,7 @@ async fn running_upstream_cancellation_closes_request_and_cleans_spool() { let parent = tempfile::tempdir().unwrap(); let spools = Spools::new(Ok(parent.path().to_owned())); let uploads = Uploads::default(); - spools.begin(&uploads, "running", 3).unwrap(); + spools.begin(&uploads, "main", "running", 3).unwrap(); spools.append("running", 0, b"abc").unwrap(); let (spool, mut cancel) = spools.take("running").unwrap(); let path = spool.directory.path().to_owned(); diff --git a/src-tauri/src/relay/upload_spool.rs b/src-tauri/src/relay/upload_spool.rs index 781c036b91..f41dc2aff2 100644 --- a/src-tauri/src/relay/upload_spool.rs +++ b/src-tauri/src/relay/upload_spool.rs @@ -146,7 +146,14 @@ impl Spools { }; drop(removed); } - pub(super) fn begin(&self, uploads: &Uploads, id: &str, size: usize) -> Result<()> { + /// Starts receiving upload `id` for the page in `webview`. + pub(super) fn begin( + &self, + uploads: &Uploads, + webview: &str, + id: &str, + size: usize, + ) -> Result<()> { upload_id(Some(id))?; validate_upload_size(size)?; let slot = self @@ -173,6 +180,7 @@ impl Spools { return Err("Upload is already in progress".into()); } let cancelled = uploads.start(id)?.ok_or("Upload cancelled")?; + uploads.own(id, webview); entries.insert(id.into(), entry.clone()); cancelled }; @@ -292,6 +300,19 @@ impl Spools { }; drop(removed); } + /// Stops what the page in `webview` was uploading, as it loads again or + /// closes. Other pages' uploads go on. + pub(crate) fn cancel_page(&self, uploads: &Uploads, webview: &str) { + let removed: Vec<_> = { + let mut entries = self.lock(); + uploads + .cancel_page(webview) + .into_iter() + .filter_map(|id| entries.remove(&id)) + .collect() + }; + drop(removed); + } pub(crate) fn cancel_all(&self, uploads: &Uploads) { let removed = { let mut entries = self.lock(); @@ -383,10 +404,10 @@ mod tests { fn simultaneous_files_have_exact_hashes_and_cleanup_holds_admission() { let (_parent, spools, uploads) = fixture(); for id in ["a", "b", "c", "d"] { - spools.begin(&uploads, id, 3).unwrap(); + spools.begin(&uploads, "main", id, 3).unwrap(); } - assert!(spools.begin(&uploads, "extra", 3).is_err()); - assert!(spools.begin(&uploads, "a", 3).is_err()); + assert!(spools.begin(&uploads, "main", "extra", 3).is_err()); + assert!(spools.begin(&uploads, "main", "a", 3).is_err()); spools.append("a", 0, b"abc").unwrap(); spools.append("b", 0, b"xyz").unwrap(); let (a, _cancel) = spools.take("a").unwrap(); @@ -395,11 +416,11 @@ mod tests { assert_eq!(std::fs::read(a.source()).unwrap(), b"abc"); assert_eq!(a.hash, format!("{:x}", Sha256::digest(b"abc"))); assert_eq!(std::fs::read(b.source()).unwrap(), b"xyz"); - assert!(spools.begin(&uploads, "extra", 3).is_err()); + assert!(spools.begin(&uploads, "main", "extra", 3).is_err()); drop(a); assert!(!path.exists()); uploads.finish("a"); - spools.begin(&uploads, "extra", 3).unwrap(); + spools.begin(&uploads, "main", "extra", 3).unwrap(); } #[test] fn cancellation_partial_and_invalid_chunks_remove_files() { @@ -409,30 +430,30 @@ mod tests { ("empty", 0, b""), ("overflow", 0, b"four"), ] { - spools.begin(&uploads, id, 3).unwrap(); + spools.begin(&uploads, "main", id, 3).unwrap(); let path = spool_path(&spools, id); assert!(spools.append(id, offset, bytes).is_err()); assert!(!path.exists()); uploads.finish(id); } - spools.begin(&uploads, "partial", 3).unwrap(); + spools.begin(&uploads, "main", "partial", 3).unwrap(); let path = spool_path(&spools, "partial"); spools.append("partial", 0, b"a").unwrap(); assert!(spools.take("partial").is_err()); assert!(!path.exists()); uploads.finish("partial"); - spools.begin(&uploads, "cancel", 3).unwrap(); + spools.begin(&uploads, "main", "cancel", 3).unwrap(); let path = spool_path(&spools, "cancel"); spools.cancel(&uploads, "cancel"); assert!(!path.exists()); assert!(spools.append("cancel", 0, b"abc").is_err()); uploads.cancel("before"); - assert!(spools.begin(&uploads, "before", 3).is_err()); + assert!(spools.begin(&uploads, "main", "before", 3).is_err()); } #[test] fn idle_and_page_reload_release_receiving_and_running_work() { let (_parent, spools, uploads) = fixture(); - spools.begin(&uploads, "idle", 3).unwrap(); + spools.begin(&uploads, "main", "idle", 3).unwrap(); let path = spool_path(&spools, "idle"); receiving(&spools, "idle") .lock() @@ -442,12 +463,23 @@ mod tests { .touched = Instant::now() - IDLE; spools.reap(&uploads); assert!(!path.exists()); - spools.begin(&uploads, "running", 3).unwrap(); + spools.begin(&uploads, "main", "running", 3).unwrap(); spools.append("running", 0, b"abc").unwrap(); let (running, mut cancel) = spools.take("running").unwrap(); let path = running.directory.path().to_owned(); - spools.cancel_all(&uploads); + // Another webview's page goes on uploading while this one reloads. + spools.begin(&uploads, "agents", "other", 3).unwrap(); + let other = spool_path(&spools, "other"); + spools.begin(&uploads, "agents", "sending", 3).unwrap(); + spools.append("sending", 0, b"abc").unwrap(); + let (sending, mut kept) = spools.take("sending").unwrap(); + spools.cancel_page(&uploads, "main"); assert!(cancel.try_recv().is_ok()); + assert!(other.exists()); + spools.append("other", 0, b"abc").unwrap(); + assert!(kept.try_recv().is_err()); + drop(sending); + uploads.finish("sending"); assert!( path.exists(), "running operation owns cleanup until stopped" @@ -458,8 +490,8 @@ mod tests { #[test] fn busy_upload_does_not_hold_map_or_other_uploads_and_cancel_wins() { let (_parent, spools, uploads) = fixture(); - spools.begin(&uploads, "busy", 3).unwrap(); - spools.begin(&uploads, "other", 3).unwrap(); + spools.begin(&uploads, "main", "busy", 3).unwrap(); + spools.begin(&uploads, "main", "other", 3).unwrap(); let busy = receiving(&spools, "busy"); let path = spool_path(&spools, "busy"); // Hold the exact mutex used across write_all. No timing/sleep required. @@ -470,7 +502,7 @@ mod tests { spools.append("other", 0, b"abc").unwrap(); let (other, _) = spools.take("other").unwrap(); assert_eq!(std::fs::read(other.source()).unwrap(), b"abc"); - spools.begin(&uploads, "new", 3).unwrap(); + spools.begin(&uploads, "main", "new", 3).unwrap(); spools.reap(&uploads); spools.cancel(&uploads, "busy"); assert!(spools.entry("busy").is_err()); @@ -536,7 +568,7 @@ mod tests { let removal = std::fs::remove_dir_all(stale.path()); let spools = Spools::new(Ok(parent.path().to_owned())); let uploads = Uploads::default(); - let admission = spools.begin(&uploads, "new", 3); + let admission = spools.begin(&uploads, "main", "new", 3); // Release the fixture's restriction before assertions, even on failure. #[cfg(unix)] std::fs::set_permissions(&blocked, original).unwrap(); @@ -559,7 +591,7 @@ mod tests { let spools = Spools::new(Ok(file)); let uploads = Uploads::default(); assert!(spools - .begin(&uploads, "failure", 3) + .begin(&uploads, "main", "failure", 3) .unwrap_err() .contains("temporary storage")); assert_eq!(spools.slots.available_permits(), SLOTS); @@ -572,13 +604,13 @@ mod tests { std::fs::write(&storage, b"unavailable").unwrap(); let spools = Spools::new(Ok(storage.clone())); let uploads = Uploads::default(); - assert!(spools.begin(&uploads, "failed", 3).is_err()); + assert!(spools.begin(&uploads, "main", "failed", 3).is_err()); assert_eq!(spools.slots.available_permits(), SLOTS); assert!(uploads.lock().active.is_empty()); std::fs::remove_file(&storage).unwrap(); let live = root(&storage).unwrap(); let live_path = live.directory.path().to_owned(); - spools.begin(&uploads, "recovered", 3).unwrap(); + spools.begin(&uploads, "main", "recovered", 3).unwrap(); assert!(live_path.exists()); spools.append("recovered", 0, b"abc").unwrap(); let (spool, _) = spools.take("recovered").unwrap(); @@ -600,7 +632,7 @@ mod tests { let uploads_ref = &uploads; let begin = scope.spawn(move || { started.send(()).unwrap(); - spools_ref.begin(uploads_ref, "pending", 3) + spools_ref.begin(uploads_ref, "main", "pending", 3) }); entering.recv().unwrap(); // Hold root initialization until the cancellation has been recorded. @@ -615,7 +647,7 @@ mod tests { } fn assert_write_failure_cleans_partial_file(file: impl FnOnce(&Path) -> File) { let (_parent, spools, uploads) = fixture(); - spools.begin(&uploads, "failure", 3).unwrap(); + spools.begin(&uploads, "main", "failure", 3).unwrap(); spools.append("failure", 0, b"a").unwrap(); let path = spool_path(&spools, "failure"); receiving(&spools, "failure") @@ -631,7 +663,7 @@ mod tests { assert!(!path.exists()); assert_eq!(spools.slots.available_permits(), SLOTS); uploads.cancel("failure"); - spools.begin(&uploads, "retry", 3).unwrap(); + spools.begin(&uploads, "main", "retry", 3).unwrap(); spools.append("retry", 0, b"abc").unwrap(); assert_eq!( std::fs::read(spools.take("retry").unwrap().0.source()).unwrap(), diff --git a/src-tauri/tauri.native-mcp.conf.json b/src-tauri/tauri.native-mcp.conf.json index e55d90c0c2..e1909f58a6 100644 --- a/src-tauri/tauri.native-mcp.conf.json +++ b/src-tauri/tauri.native-mcp.conf.json @@ -6,6 +6,7 @@ "main-window", "desktop-window", "huddle-window", + "agents-page", { "identifier": "native-mcp-callbacks", "description": "Callbacks for the opt-in native MCP test bridge.", diff --git a/src/agents.ts b/src/agents.ts new file mode 100644 index 0000000000..cbb173dd54 --- /dev/null +++ b/src/agents.ts @@ -0,0 +1,9 @@ +// FOUNDATION: Startup of the page that runs agents. It has no screens. It +// reads agent storage once the screens' page has loaded it. +import { createAgentsHost } from "./app/agents-host"; + +const host = createAgentsHost(); +if (import.meta.hot) + import.meta.hot.dispose(() => { + void host.dispose().catch(console.error); + }); diff --git a/src/app/agents-host.test.ts b/src/app/agents-host.test.ts new file mode 100644 index 0000000000..38780f73bc --- /dev/null +++ b/src/app/agents-host.test.ts @@ -0,0 +1,42 @@ +// @vitest-environment jsdom +import { expect, it, vi } from "vitest"; +import { lanes } from "../features/agents/lanes"; +import { link } from "../features/agents/link"; +import { type PageState, servePage } from "../features/agents/page-link"; +import { createAgentsHost } from "./agents-host"; + +it("starts the Runner once the screens' page says where agent storage is, and serves the Manager its agent types", async () => { + const channel = new MessageChannel(); + const host = createAgentsHost(channel.port1); + const screens = lanes(channel.port2, ["agents", "page"]); + let state: PageState | undefined; + const listeners = new Set<() => void>(); + const page = link( + screens.page, + servePage({ + snapshot: () => state, + subscribe(listener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }), + ); + const manager = link(screens.agents); + manager.watch("runner.types", null, () => {}); + page.watch("activity.work", null, () => {}); + // The agents page shows its local work at once. + await vi.waitFor(() => expect(page.read("activity.work")).toEqual([])); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(manager.read("runner.types")).toBeUndefined(); + state = Object.freeze({ + storage: "local", + communities: { memberships: [], selected: null }, + }); + for (const listener of [...listeners]) listener(); + await vi.waitFor(() => + expect(manager.read("runner.types")).toEqual(expect.any(Array)), + ); + manager.close(); + page.close(); + await host.dispose(); +}); diff --git a/src/app/agents-host.ts b/src/app/agents-host.ts new file mode 100644 index 0000000000..35ef345a99 --- /dev/null +++ b/src/app/agents-host.ts @@ -0,0 +1,123 @@ +// FOUNDATION: Compose the page that runs agents apart from the screens: the +// Runner and the plugins' agent entries. It reaches the Manager, in the +// screens' page, over the agents port. It shows nothing, and keeps nothing +// on the device for the screens: their communities, outbox and history stay +// theirs. +import { Context } from "@deepseek-ai/cordis"; +import { bundledPlugins } from "../bundled"; +import { AGENTS_WEBVIEW } from "../features/agents/apart"; +import { eventPort } from "../features/agents/event-port"; +import { lanes } from "../features/agents/lanes"; +import { link, type LinkPort } from "../features/agents/link"; +import { createLocalAgentActivity } from "../features/agents/local-activity"; +import { nativeAgents } from "../features/agents/native"; +import { followPage } from "../features/agents/page-link"; +import { serveActivity } from "../features/agents/runner-link"; +import { + AgentsRunnerService, + RunnerAgents, +} from "../features/agents/runner-service"; +import { agentStorage, readAgentStorage } from "../features/agents/state"; +import { pageCommunities } from "../features/communities/page"; +import { AgentProtectionService } from "../features/agent-protection/service"; +import { DecisionsService } from "../features/decisions/service"; +import { HostService } from "../features/host/service"; +import { connectNativeTransport } from "../features/relay/native"; +import { createPluginManager } from "../plugins/manager"; +import { withTimeout } from "../plugins/timeout"; + +export function createAgentsHost( + port: LinkPort = eventPort(AGENTS_WEBVIEW, "main"), +) { + const ctx = new Context(); + new HostService(ctx); + // Each agent entry starts once the Runner gives it `ctx.agents`. + const plugins = createPluginManager(ctx, { + bundled: bundledPlugins, + parts: "agent", + }); + const localActivity = createLocalAgentActivity(); + const decisions = new DecisionsService(ctx); + const communities = pageCommunities(ctx, connectNativeTransport, { + localAgents: localActivity.reader, + }); + const { agents: agentsLane, page: pageLane } = lanes(port, [ + "agents", + "page", + ]); + const page = link(pageLane, serveActivity(localActivity.reader)); + let disposed = false; + let started = false; + // The screens' page loads agent storage first, and moves what localStorage + // had into it; then this page reads it, and starts the Runner. + const start = (storage: "files" | "local") => + (storage === "files" ? readAgentStorage() : Promise.resolve()).then(() => { + if (disposed) return; + const native = nativeAgents(); + // Protection plugins' agent entries register their checkers here, + // where the agents they check run. Their UI entries save policies from + // the screens' page. + const protection = new AgentProtectionService( + ctx, + () => ({ + owner: communities.snapshot().viewer, + agents: runner.running(), + types: runner.agentTypes(), + }), + { + read: async (owner) => { + if (!native?.protectionDefaults) + throw new Error("Native agent protection is unavailable"); + return native.protectionDefaults(owner); + }, + save: async (owner, revision, provider, policy) => { + if (!native?.setProtectionDefaults) + throw new Error("Native agent protection is unavailable"); + await native.setProtectionDefaults( + owner, + revision, + provider, + policy, + ); + }, + update: (pubkey, revision, provider, policy) => + runner.protect(pubkey, revision, provider, policy), + }, + ); + const runner: AgentsRunnerService = new AgentsRunnerService( + ctx, + agentsLane, + { + communities, + native, + storage: agentStorage(), + localActivity, + decisions, + protection: (agent) => protection.bind(agent), + }, + ); + new RunnerAgents(ctx, runner); + }); + const stopFollowing = followPage(page, (state) => { + communities.follow(state.communities); + if (started) return; + started = true; + start(state.storage).catch((error: unknown) => { + console.error("Agents could not start", error); + }); + }); + let disposal: Promise | undefined; + return { + dispose() { + disposed = true; + stopFollowing(); + page.close(); + agentsLane.close(); + disposal ??= withTimeout( + Promise.all([plugins.dispose(), ctx.fiber.dispose()]), + "Agents cleanup timed out; restart the app", + ).then(() => {}); + return disposal; + }, + }; +} diff --git a/src/app/services.ts b/src/app/services.ts index a3205b4c4a..a483b0f991 100644 --- a/src/app/services.ts +++ b/src/app/services.ts @@ -2,7 +2,12 @@ import { ArtifactStorageService } from "../features/artifact-storage/service"; import { createAdminAccess } from "../bundled/admin-access"; import { nativeRelayStaff } from "../features/relay-staff/native"; import { createLocalAgentActivity } from "../features/agents/local-activity"; -import { link } from "../features/agents/link"; +import { AGENTS_WEBVIEW, agentsRunApart } from "../features/agents/apart"; +import { eventPort } from "../features/agents/event-port"; +import { lanes } from "../features/agents/lanes"; +import { link, type Link } from "../features/agents/link"; +import { type PageState, servePage } from "../features/agents/page-link"; +import { agentStorageInFiles } from "../features/agents/state"; import { linkedActivity, serveActivity } from "../features/agents/runner-link"; // FOUNDATION: Compose the bundled distribution, plugin runtime, and services here. import { @@ -57,25 +62,50 @@ export function createServices() { const ctx = new Context(); new HostService(ctx); const federatedIdentity = new FederatedIdentityService(ctx); + // Agents run in a hidden page of their own where the app opens one; this + // page then runs only the plugins' UI entries. + const apart = agentsRunApart(); const plugins = createPluginManager(ctx, { bundled: bundledPlugins, builtIn: conversationFeatures, + parts: apart ? "ui" : "both", }); const localAgents = createLocalAgents(); + // The Manager, which the screens use, reaches the Runner, which runs + // agents, over a link: to the agents page, or within this page. Only JSON + // crosses. Beside it, the Runner's page shows its local work to this one, + // which tells it the communities it follows. + const agentsPort = apart + ? lanes(eventPort("main", AGENTS_WEBVIEW), ["agents", "page"]) + : undefined; const localActivity = createLocalAgentActivity(); - // The Manager, which the screens use, and the Runner, which runs agents, - // share this page for now. Only JSON crosses the link between them. - const agentsLink = new MessageChannel(); - const runnerLink = link( - agentsLink.port1, - serveActivity(localActivity.reader), - ); - const managerLink = link(agentsLink.port2); - const runnerActivity = linkedActivity(managerLink); + let pageState: PageState | undefined; + const pageListeners = new Set<() => void>(); + let pageLink: Link; + let runnerPage: Link | undefined; + if (agentsPort) + pageLink = link( + agentsPort.page, + servePage({ + snapshot: () => pageState, + subscribe(listener) { + pageListeners.add(listener); + return () => { + pageListeners.delete(listener); + }; + }, + }), + ); + else { + const channel = new MessageChannel(); + runnerPage = link(channel.port1, serveActivity(localActivity.reader)); + pageLink = link(channel.port2); + } + const runnerActivity = linkedActivity(pageLink); ctx.effect(() => () => { runnerActivity.stop(); - managerLink.close(); - runnerLink.close(); + pageLink.close(); + runnerPage?.close(); }); const navigationHost = provideNavigation(ctx); const navigation = navigationHost.navigation; @@ -121,6 +151,22 @@ export function createServices() { const adminAccess = createAdminAccess(relay, communities, nativeRelayStaff()); ctx.provide("adminAccess", adminAccess); ctx.effect(() => () => adminAccess.dispose()); + if (agentsPort) { + const follow = () => { + const { viewer, memberships, selected } = communities.snapshot(); + pageState = Object.freeze({ + storage: agentStorageInFiles() ? "files" : "local", + communities: { + ...(viewer !== undefined && { viewer }), + memberships: memberships.map((membership) => membership.id), + selected, + }, + }); + for (const listener of [...pageListeners]) listener(); + }; + follow(); + ctx.effect(() => communities.subscribe(follow)); + } const agents = new AgentsService( ctx, communities, @@ -129,6 +175,8 @@ export function createServices() { undefined, localActivity, decisions, + undefined, + agentsPort?.agents, ); ctx.effect(() => localAgents.bind({ diff --git a/src/features/agents/agents-runner.test.ts b/src/features/agents/agents-runner.test.ts index c9904cf816..a56e1f2906 100644 --- a/src/features/agents/agents-runner.test.ts +++ b/src/features/agents/agents-runner.test.ts @@ -29,7 +29,6 @@ function saved() { function runner(storage: ReturnType) { const manager = { ownershipChanged: () => {}, - changed: () => {}, open: () => undefined, types: () => [], defaults: () => { diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index efd70524b0..7c6abb014c 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -14,6 +14,7 @@ import { type Checkpoints, } from "./catch-up"; import { originOf, RunnerConnections } from "./connections"; +import { AgentPresence } from "./presence"; import { CoreMemoryCache } from "./core-memory-cache"; import { readInterrupted, writeInterrupted } from "./interrupted"; import type { LocalAgentActivity } from "./local-activity"; @@ -57,8 +58,6 @@ export type RunnerManager = Pick< "createAgent" | "updateAgent" | "templates" | "protection" > & Readonly<{ - /** A record or turn in progress changed what the screens show. */ - changed(): void; /** The community's relay, while the viewer is in it. */ open(origin: string): RelayData | undefined; types(): readonly RegisteredAgentType[]; @@ -71,6 +70,9 @@ export type RunnerManager = Pick< publishType(key: AgentKey, type: string, config: unknown): Promise; /** Where an agent's closed tabs leave the viewer. */ navigation?: Navigation | undefined; + /** Native's memberships may have changed, as when a claim is refused: + * the Manager reads them again, and sends them. */ + refresh(): Promise; }>; export class AgentsRunner { @@ -81,6 +83,8 @@ export class AgentsRunner { readonly runners: AgentRunners; /** The Runner's own live session of each community it runs agents in. */ readonly connections: RunnerConnections; + /** Says which agents this copy runs in each community. */ + readonly presence: AgentPresence; /** The agents running here, in every community, and those of them with * turns the app left working last time. */ private state: Readonly<{ @@ -119,7 +123,8 @@ export class AgentsRunner { constructor( storage: AgentStore, private readonly custody: - | (NonNullable & Pick) + | (NonNullable & + Pick) | undefined, activity: LocalAgentActivity, decisions: Decisions | undefined, @@ -141,7 +146,7 @@ export class AgentsRunner { }, changed: () => { for (const listener of [...this.recordListeners]) listener(); - this.changed(); + this.update(); }, // Its timers waited for this change, saved or failed. settled: (key) => this.runners.wake(key), @@ -154,7 +159,35 @@ export class AgentsRunner { changed: (origin, renewed) => { this.update(); if (renewed) this.refreshCore(origin); - this.manager.changed(); + }, + }); + this.presence = new AgentPresence({ + running: () => { + const online = new Map>(); + for (const agent of this.state.running) { + const origin = originOf(agent.relay); + if ( + origin && + this.connections.binding(agent) && + this.runners.runs(agent.key) + ) + online.set( + origin, + (online.get(origin) ?? new Set()).add(agent.pubkey), + ); + } + return online; + }, + communities: () => this.connections.entries(), + // Each pubkey said online there is one membership's, the one there. + yielded: (origin, pubkey) => { + const key = this.identities.find( + (identity) => + identity.pubkey === pubkey && + !identity.deleted && + originOf(identity.relay) === origin, + )?.key; + return !!key && this.runners.yielded(key); }, }); this.coreMemory = new CoreMemoryCache((key) => @@ -195,7 +228,7 @@ export class AgentsRunner { console.warn("Agent turns in progress were not saved", error), ); }, - changed: () => this.changed(), + changed: () => this.update(), }, scope: (agent) => this.connections.binding(agent)?.scope, saved: (key) => this.records.saved(key), @@ -261,7 +294,24 @@ export class AgentsRunner { decisions: () => decisions, context: (agent, event) => this.context(agent, event), displayName: (agent, pubkey) => this.displayName(agent, pubkey), - custody, + // A claim names the protection the agent is saved with; a refused one + // may mean it changed. + custody: custody && { + ...custody, + claim: async (pubkey: string) => { + try { + return await custody.claim( + pubkey, + this.identities.find( + (agent) => agent.pubkey === pubkey && !agent.deleted, + )?.protectionRevision ?? 0, + ); + } catch (error) { + void manager.refresh().catch(() => undefined); + throw error; + } + }, + }, activity, soloChat: (agent, channel, signal) => soloChat(() => this.relay(agent.key), agent.owner, channel, signal), @@ -350,12 +400,6 @@ export class AgentsRunner { console.warn(`Agent ${key} settings were not removed`, error), ); } - /** Decides again what runs once a record, a turn in progress or one of its - * sessions changed, and lets the Manager show it. */ - private changed() { - this.update(); - this.manager.changed(); - } /** Agent types were loaded or unloaded. */ typesChanged() { this.update(); @@ -436,6 +480,35 @@ export class AgentsRunner { await kind.sessionControls?.run(current, place, action); }); } + /** Saves `pubkey`'s protection policy, while none of its memberships + * runs: each one's sessions are retired first, and it runs again with the + * memberships native lists after. */ + async protect( + pubkey: string, + revision: number, + provider: string, + policy: unknown, + ) { + const setProtection = this.custody?.setProtection?.bind(this.custody); + if (!setProtection) + throw new Error("Native agent protection is unavailable"); + const members = this.identities.filter( + (agent) => agent.pubkey === pubkey && !agent.deleted, + ); + const agent = members[0]; + if (!agent) throw new Error("No such agent on this device"); + const type = this.manager.types().find((type) => type.key === agent.type); + const retire = type?.retire?.bind(type); + if (!retire) + throw new Error( + "This runtime cannot safely change protection while Buzz is open", + ); + await this.runners.retype(agent.key, agent.name, async () => { + for (const member of members) await retire(member.key); + await setProtection(pubkey, revision, provider, policy); + await this.manager.refresh(); + }); + } /** Gives the agent the type `type`, starting with `config`, while no turn * of it works and this copy holds it. With `publish`, its synced settings * are saved as the new type's first. It shows with the new type once that @@ -569,6 +642,7 @@ export class AgentsRunner { ), ); this.runners.catchUp(false); + this.presence.announce(); } /** The agents running here, in every community. */ running() { @@ -651,6 +725,7 @@ export class AgentsRunner { clear() { this.connections.clear(); this.runners.clear(); + this.presence.clear(); this.identities = []; this.viewer = undefined; this.list = undefined; diff --git a/src/features/agents/apart.ts b/src/features/agents/apart.ts new file mode 100644 index 0000000000..66c19ad8c5 --- /dev/null +++ b/src/features/agents/apart.ts @@ -0,0 +1,16 @@ +import { isTauri } from "@tauri-apps/api/core"; + +/** The webview of the page that runs agents, apart from the screens. */ +export const AGENTS_WEBVIEW = "agents"; + +/** Whether agents run in their own hidden page, so a screen that is hidden + * or busy cannot hold them up. The desktop app on macOS opens that page at + * startup (`agents_window` in src-tauri/src/lib.rs); elsewhere they run in + * the page that shows them. */ +export function agentsRunApart() { + return ( + isTauri() && + typeof navigator !== "undefined" && + /^Mac/.test(navigator.platform) + ); +} diff --git a/src/features/agents/connections.ts b/src/features/agents/connections.ts index 89b4aa1bef..6b6913f0a1 100644 --- a/src/features/agents/connections.ts +++ b/src/features/agents/connections.ts @@ -34,6 +34,24 @@ type Entry = { binding?: RunnerBinding | undefined; }; +/** How often a failed connection to an unselected community is tried again. */ +const RETRY_MS = 30_000; +/** When each community's failed connection was last tried again, by whom + * ever asked, so two owners of one session do not each try it. */ +const retried = new WeakMap(); + +/** Tries the community's connection again if it failed, at most every + * `RETRY_MS`. */ +export function retryFailed(relay: RelayData, now = Date.now()) { + if ( + relay.snapshot().status !== "error" || + now - (retried.get(relay) ?? 0) < RETRY_MS + ) + return; + retried.set(relay, now); + relay.retry(); +} + export type ConnectionsHost = Readonly<{ /** The community's relay, while the viewer is in it. */ open(origin: string): RelayData | undefined; @@ -100,6 +118,12 @@ export class RunnerConnections { ([origin, entry]) => [origin, entry.binding] as const, ); } + /** Tries again each community whose connection failed but the selected + * one, which the screens try again. */ + reconnect(selected: string | undefined) { + for (const [origin, entry] of this.bound) + if (origin !== selected) retryFailed(entry.relay); + } clear() { for (const origin of [...this.bound.keys()]) this.unbind(origin); this.watched.clear(); diff --git a/src/features/agents/event-port.test.ts b/src/features/agents/event-port.test.ts new file mode 100644 index 0000000000..a38928964b --- /dev/null +++ b/src/features/agents/event-port.test.ts @@ -0,0 +1,92 @@ +import { expect, it } from "vitest"; +import { eventPort, type Events, type Frame } from "./event-port"; + +/** Tauri events between webviews, delivered when the test says, each batch + * in reverse, and as JSON. */ +function bus() { + const handlers = new Set<(event: { payload: Frame }) => void>(); + let queue: Frame[] = []; + const events: Events = { + emitTo: async (_webview, _event, frame) => { + queue.push(JSON.parse(JSON.stringify(frame)) as Frame); + }, + listen: async (_event, handler) => { + handlers.add(handler); + return () => handlers.delete(handler); + }, + }; + return { + events, + /** Delivers what was sent, then what that sent, until nothing is. */ + async settle() { + for (let round = 0; round < 20; round += 1) { + for (let tick = 0; tick < 5; tick += 1) await Promise.resolve(); + const batch = queue; + queue = []; + if (!batch.length) continue; + for (const frame of batch.reverse()) + for (const handler of [...handlers]) handler({ payload: frame }); + } + }, + }; +} + +function received(port: ReturnType) { + const messages: unknown[] = []; + port.onmessage = ({ data }) => messages.push(data); + return messages; +} + +it("delivers each message once, in the order sent, though frames arrive out of order", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const main = eventPort("main", "agents", events); + const heard = received(main); + await settle(); + for (const n of [1, 2, 3]) agents.postMessage({ n }); + await settle(); + expect(heard).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]); +}); + +it("holds what a page sends until the other page is up", async () => { + const { events, settle } = bus(); + const main = eventPort("main", "agents", events); + main.postMessage("early"); + await settle(); + const agents = eventPort("agents", "main", events); + const heard = received(agents); + await settle(); + expect(heard).toEqual(["early"]); +}); + +it("talks to a reloaded page afresh, and drops the old page's late frames", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const heardByAgents = received(agents); + const before = eventPort("main", "agents", events); + await settle(); + // The old page's last message is still on its way when the page reloads. + before.postMessage("stale"); + before.close(); + const after = eventPort("main", "agents", events); + const heardByMain = received(after); + await settle(); + after.postMessage("fresh"); + agents.postMessage("hello again"); + await settle(); + expect(heardByAgents).toEqual(["fresh"]); + expect(heardByMain).toEqual(["hello again"]); +}); + +it("ignores frames for another webview", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const main = eventPort("main", "agents", events); + const huddle = eventPort("huddle", "agents", events); + const heard = received(huddle); + await settle(); + agents.postMessage("for main"); + await settle(); + expect(heard).toEqual([]); + main.close(); +}); diff --git a/src/features/agents/event-port.ts b/src/features/agents/event-port.ts new file mode 100644 index 0000000000..cdac555ea0 --- /dev/null +++ b/src/features/agents/event-port.ts @@ -0,0 +1,144 @@ +// A port between pages in two webviews of this app, over the native agents +// port (src-tauri/src/agents_port.rs), which passes each frame from one +// page's webview to the other's. The link (link.ts) needs each message once +// and in the order sent: a call's answer must not overtake the state push +// before it. Separate sends can arrive out of order, so each frame is +// numbered and held until its turn. +// +// Either page may load first, and either may reload. Each page load is a +// session. A page announces its session once it listens, and answers a new +// peer's announcement with its own; until it knows the peer's session it +// holds what it sends. Frames name the session they are for, so a frame +// meant for a page that has since reloaded is dropped. +import { Channel, invoke } from "@tauri-apps/api/core"; +import type { LinkPort } from "./link"; + +const EVENT = "buzz://agents-port"; +type UnlistenFn = () => void; + +export type Frame = Readonly< + { + /** The webviews it is for and from. */ + to: string; + by: string; + /** The sending page load. */ + from: string; + } & ( + | { kind: "ready" } + | { kind: "message"; for: string; seq: number; data: unknown } + ) +>; + +export type Events = Readonly<{ + emitTo(webview: string, event: string, payload: Frame): Promise; + listen( + event: string, + handler: (event: { payload: Frame }) => void, + ): Promise; +}>; + +/** The native agents port: it knows which webview a frame comes from, and + * sends it to the other one. */ +export const agentsPortEvents: Events = { + emitTo: (_webview, _event, frame) => invoke("agents_port_send", { frame }), + async listen(_event, handler) { + const frames = new Channel(); + let open = true; + frames.onmessage = (payload) => { + if (open) handler({ payload }); + }; + await invoke("agents_port_listen", { frames }); + return () => { + open = false; + }; + }, +}; + +/** A port from this page, in webview `from`, to the page in webview `to`. + * Messages are JSON: a value JSON cannot carry does not arrive as sent. + * What this page sends before the peer's page is up waits for it; what it + * sent to a peer page that reloaded is lost with that page. */ +export function eventPort( + from: string, + to: string, + events: Events = agentsPortEvents, +): LinkPort { + const session = crypto.randomUUID(); + let closed = false; + let peer: string | undefined; + let sent = 0; + let next = 0; + const waiting: unknown[] = []; + const held = new Map(); + // Peer pages that have since reloaded: their late frames are stale. + const retired = new Set(); + const emit = (frame: Frame) => + events.emitTo(to, EVENT, frame).catch((error: unknown) => { + console.error("Agents port message was not sent", error); + }); + const send = (data: unknown) => { + if (peer) + void emit({ + to, + by: from, + from: session, + kind: "message", + for: peer, + seq: sent++, + data, + }); + }; + const port: LinkPort = { + onmessage: null, + postMessage(data) { + if (closed) return; + if (peer) send(data); + else waiting.push(data); + }, + close() { + closed = true; + waiting.length = 0; + held.clear(); + void unlisten.then((stop) => stop()); + }, + }; + // A new peer page: what was sent to the last one is gone with it. + const meet = (other: string) => { + if (peer) retired.add(peer); + peer = other; + sent = 0; + next = 0; + held.clear(); + for (const data of waiting.splice(0)) send(data); + }; + const unlisten = events + .listen(EVENT, ({ payload: frame }) => { + if (closed || frame.to !== from || frame.by !== to) return; + if (frame.from !== peer && retired.has(frame.from)) return; + if (frame.kind === "ready") { + if (frame.from === peer) return; + meet(frame.from); + void emit({ to, by: from, from: session, kind: "ready" }); + return; + } + if (frame.for !== session) return; + // It knows this page, so its announcement is on the way or was missed. + if (frame.from !== peer) meet(frame.from); + held.set(frame.seq, frame.data); + while (held.has(next) && !closed) { + const data = held.get(next); + held.delete(next); + next += 1; + port.onmessage?.(new MessageEvent("message", { data })); + } + }) + .then((stop) => { + if (!closed) void emit({ to, by: from, from: session, kind: "ready" }); + return stop; + }) + .catch((error: unknown): UnlistenFn => { + console.error("Agents port is not listening", error); + return () => {}; + }); + return port; +} diff --git a/src/features/agents/lanes.test.ts b/src/features/agents/lanes.test.ts new file mode 100644 index 0000000000..45deaf0bcf --- /dev/null +++ b/src/features/agents/lanes.test.ts @@ -0,0 +1,49 @@ +import { expect, it, vi } from "vitest"; +import { lanes } from "./lanes"; +import { link } from "./link"; + +it("carries a link on each lane of one port, apart from the others", async () => { + const channel = new MessageChannel(); + const close = vi.spyOn(channel.port1, "close"); + const one = lanes(channel.port1, ["agents", "page"]); + const other = lanes(channel.port2, ["agents", "page"]); + const agents = link(one.agents, { + topics: { + shared: () => ({ get: () => "agents", subscribe: () => () => {} }), + }, + }); + const page = link(one.page, { + topics: { + shared: () => ({ get: () => "page", subscribe: () => () => {} }), + }, + }); + const agentsPeer = link(other.agents); + const pagePeer = link(other.page); + agentsPeer.watch("shared", null, () => {}); + pagePeer.watch("shared", null, () => {}); + await vi.waitFor(() => { + expect(agentsPeer.read("shared")).toBe("agents"); + expect(pagePeer.read("shared")).toBe("page"); + }); + // The port stays open while a lane is. + agents.close(); + expect(close).not.toHaveBeenCalled(); + page.close(); + expect(close).toHaveBeenCalledTimes(1); + agentsPeer.close(); + pagePeer.close(); +}); + +it("holds what reaches a lane before it listens, as a MessagePort does", async () => { + const channel = new MessageChannel(); + const one = lanes(channel.port1, ["agents"]); + const other = lanes(channel.port2, ["agents"]); + one.agents.postMessage("early"); + await new Promise((resolve) => setTimeout(resolve, 10)); + const heard: unknown[] = []; + other.agents.onmessage = ({ data }) => heard.push(data); + one.agents.postMessage("later"); + await vi.waitFor(() => expect(heard).toEqual(["early", "later"])); + one.agents.close(); + other.agents.close(); +}); diff --git a/src/features/agents/lanes.ts b/src/features/agents/lanes.ts new file mode 100644 index 0000000000..77d189df22 --- /dev/null +++ b/src/features/agents/lanes.ts @@ -0,0 +1,63 @@ +// Several links over one port. The native port between two webviews has one +// end per page, so each link between those pages takes a named lane of it. +import type { LinkPort } from "./link"; + +type Lane = Readonly<{ lane: string; data: unknown }>; + +const isLane = (message: unknown): message is Lane => + typeof message === "object" && + message !== null && + typeof (message as { lane?: unknown }).lane === "string" && + "data" in message; + +/** A port for each of `names` over `port`. Each delivers what the same lane + * of the other end sends, once and in order, as `port` does; what arrives + * before the lane listens waits for it, as for a `MessagePort`. `port` + * closes once every lane has. */ +export function lanes( + port: LinkPort, + names: readonly Name[], +): Readonly> { + const open = new Set(names); + const ports = {} as Record; + const waiting = new Map(); + const deliver = (name: Name, data: unknown) => + ports[name].onmessage?.(new MessageEvent("message", { data })); + const flush = (name: Name) => { + const held = waiting.get(name) ?? []; + while (held.length && ports[name].onmessage) deliver(name, held.shift()); + }; + for (const name of names) { + let listener: LinkPort["onmessage"] = null; + waiting.set(name, []); + ports[name] = { + get onmessage() { + return listener; + }, + set onmessage(next) { + listener = next; + // Later, as a MessagePort starts, and before what arrives next. + if (next && waiting.get(name)?.length) + queueMicrotask(() => flush(name)); + }, + postMessage(data) { + if (open.has(name)) port.postMessage({ lane: name, data }); + }, + close() { + if (!open.delete(name)) return; + listener = null; + waiting.delete(name); + if (open.size === 0) port.close(); + }, + }; + } + port.onmessage = (event) => { + const message: unknown = event.data; + if (!isLane(message) || !open.has(message.lane as Name)) return; + const name = message.lane as Name; + const held = waiting.get(name); + if (ports[name].onmessage && !held?.length) deliver(name, message.data); + else held?.push(message.data); + }; + return ports; +} diff --git a/src/features/agents/page-link.ts b/src/features/agents/page-link.ts new file mode 100644 index 0000000000..a0938a2e19 --- /dev/null +++ b/src/features/agents/page-link.ts @@ -0,0 +1,39 @@ +// The link between the screens' page and the page that runs agents, beside +// the Manager's and the Runner's: what the screens' page has that the agents +// page follows, and the local work the agents page shows back. +import type { FollowedCommunities } from "../communities/page"; +import type { Json, Link, Served } from "./link"; + +/** What the screens' page tells the agents page. */ +export type PageState = Readonly<{ + /** Where agent storage is: the device's files, which the screens' page + * loaded first, or localStorage, which both pages share. */ + storage: "files" | "local"; + communities: FollowedCommunities; +}>; + +/** The screens' page's side: `state` once it has it, then each change. */ +export function servePage( + state: Readonly<{ + snapshot(): PageState | undefined; + subscribe(listener: () => void): () => void; + }>, +): Served { + return { + topics: { + "page.state": () => ({ + get: () => (state.snapshot() ?? null) as Json, + subscribe: state.subscribe, + }), + }, + }; +} + +/** The agents page's side: hears each state the screens' page sends, from + * the first. */ +export function followPage(page: Link, listener: (state: PageState) => void) { + return page.watch("page.state", null, () => { + const state = page.read("page.state") as PageState | null | undefined; + if (state) listener(state); + }); +} diff --git a/src/features/agents/runner-link.test.ts b/src/features/agents/runner-link.test.ts index 582ee9ce72..475f59c2ff 100644 --- a/src/features/agents/runner-link.test.ts +++ b/src/features/agents/runner-link.test.ts @@ -163,6 +163,14 @@ function fakeRunner() { _action: SessionControlAction, ) => {}, ), + protect: vi.fn( + async ( + _pubkey: string, + _revision: number, + _provider: string, + _policy: unknown, + ) => {}, + ), // Saves the config, and the type it is shown with, at once. retype: vi.fn( async (key: string, type: string, config: unknown, _publish: boolean) => { @@ -257,6 +265,7 @@ function split(identities: Member[] = []) { async (_key: string, _type: string, _config: unknown) => {}, ), finished: vi.fn((_key: string) => {}), + refresh: vi.fn(async () => {}), }; const sharesChanged = new Set<() => void>(); const manager = managerSide(channel.port2, { @@ -277,6 +286,7 @@ function split(identities: Member[] = []) { templates: asks.templates, publishType: asks.publishType, finished: asks.finished, + refresh: asks.refresh, shared: (plugin) => shares.get(plugin), subscribeShared(listener) { sharesChanged.add(listener); @@ -430,6 +440,7 @@ it("does not take what the Runner decided for a page that reloaded as an answer" templates: async () => [], publishType: async () => {}, finished: () => {}, + refresh: async () => {}, shared: () => undefined, subscribeShared: () => () => {}, }); diff --git a/src/features/agents/runner-link.ts b/src/features/agents/runner-link.ts index 0138e4500a..62d1ee6e04 100644 --- a/src/features/agents/runner-link.ts +++ b/src/features/agents/runner-link.ts @@ -27,7 +27,12 @@ import type { Teammate } from "./team-prompt"; * type's, and that a run of it ended. */ type ManagerAsks = Pick< RunnerManager, - "createAgent" | "updateAgent" | "templates" | "publishType" | "finished" + | "createAgent" + | "updateAgent" + | "templates" + | "publishType" + | "finished" + | "refresh" >; type WorkKey = Pick; @@ -303,6 +308,7 @@ export function runnerSide( | "edit" | "session" | "retype" + | "protect" | "retypes" | "types" | "subscribeTypes" @@ -461,6 +467,17 @@ export function runnerSide( await runner.retype(key, type, config, publish); return null; }, + "agents.protect": async (args) => { + const { pubkey, revision, provider, policy } = + args as unknown as Readonly<{ + pubkey: string; + revision: number; + provider: string; + policy: unknown; + }>; + await runner.protect(pubkey, revision, provider, policy); + return null; + }, "agents.resume": (keys) => { runner.resumeInterrupted(keys as readonly AgentKey[]); return null; @@ -517,9 +534,16 @@ export function runnerSide( console.warn("The Manager was not told a run ended", error), ); }, + refresh: async () => { + await ask("manager.refresh", null); + }, }; return Object.assign(side, asks); } + +/** How long the Manager waits for the Runner to decide before it goes on. */ +const DECIDED_MS = 10_000; + /** The Manager's side: it sends the Runner the viewer's memberships, and * shows what the Runner decided and the records it keeps. An agent or record * keeps its identity until the Runner sends a change to it. */ @@ -620,6 +644,11 @@ export function managerSide( host.finished(key as AgentKey); return null; }, + // Settles once the memberships it read are sent. + "manager.refresh": async () => { + await host.refresh(); + return null; + }, "plugin.call": async (args) => { const { plugin, @@ -743,10 +772,26 @@ export function managerSide( revision++; for (const listener of [...listeners]) listener(); }, - /** Settles once the Runner decided from the memberships sent so far. */ + /** Settles once the Runner decided from the memberships sent so far, + * or after `DECIDED_MS` without an answer from a Runner page that is not + * up. */ decided() { if (closed || state.revision >= revision) return Promise.resolve(); - return new Promise((resolve) => waits.add({ revision, resolve })); + return new Promise((resolve) => { + const wait = { + revision, + resolve() { + clearTimeout(timer); + resolve(); + }, + }; + const timer = setTimeout(() => { + waits.delete(wait); + console.warn("The agents page has not answered; going on without it"); + resolve(); + }, DECIDED_MS); + waits.add(wait); + }); }, /** The agents running here, in every community. */ running: () => state.running, @@ -850,6 +895,21 @@ export function managerSide( key, }); }, + /** Asks the Runner to save `pubkey`'s protection policy once its + * sessions are retired. */ + async protect( + pubkey: string, + revision: number, + provider: string, + policy: unknown, + ) { + await side.call("agents.protect", { + pubkey, + revision, + provider, + policy: policy as Json, + }); + }, /** Asks the Runner to give the agent type `type`, starting with * `config`, and with `publish` to save its synced settings as that * type's first; settles once it is saved, and shown here. */ diff --git a/src/features/agents/runner-service.test.ts b/src/features/agents/runner-service.test.ts new file mode 100644 index 0000000000..8b6b681071 --- /dev/null +++ b/src/features/agents/runner-service.test.ts @@ -0,0 +1,60 @@ +import { Context } from "@deepseek-ai/cordis"; +import { expect, it, vi } from "vitest"; +import { link } from "./link"; +import { createLocalAgentActivity } from "./local-activity"; +import { AgentsRunnerService } from "./runner-service"; +import type { AgentCommunities } from "./service"; +import { memoryStorage } from "./test-fakes"; + +it("runs without the Manager, which hears its agent types over the link", async () => { + const ctx = new Context(); + ctx.provide("pluginStatus", { + isActive: () => true, + subscribe: () => () => {}, + }); + const plugin = ctx.extend({ + pluginOwner: Object.freeze({ id: "example", revision: "one" }), + }); + const channel = new MessageChannel(); + const communities = { + snapshot: () => ({ viewer: undefined }), + subscribe: () => () => {}, + open: () => undefined, + } as unknown as AgentCommunities; + const runner = new AgentsRunnerService(ctx, channel.port1, { + communities, + native: undefined, + storage: memoryStorage(), + localActivity: createLocalAgentActivity(), + decisions: undefined, + }); + const manager = link(channel.port2); + manager.watch("runner.types", null, () => {}); + runner.register(plugin, { + id: "echo", + title: "Echo", + defaults: () => ({ config: {} }), + run: () => {}, + }); + await vi.waitFor(() => + expect(manager.read("runner.types")).toEqual([ + expect.objectContaining({ key: "example/echo", title: "Echo" }), + ]), + ); + // A hosted type is the Manager's. + expect(() => + runner.register(plugin, { + id: "hosted", + title: "Hosted", + defaults: () => ({ config: {} }), + remote: { + provision: async () => ({}), + activate: async () => {}, + profile: async () => {}, + remove: async () => {}, + }, + } as never), + ).toThrow("A hosted agent type runs elsewhere"); + manager.close(); + await ctx.fiber.dispose(); +}); diff --git a/src/features/agents/runner-service.ts b/src/features/agents/runner-service.ts new file mode 100644 index 0000000000..90ba441801 --- /dev/null +++ b/src/features/agents/runner-service.ts @@ -0,0 +1,305 @@ +// The Runner: it runs agents, owns the agent storage they keep, and has what +// plugins' agent entries add for them. It reaches the Manager only over its +// link, so a page can build it without the screens. +import { Service, type Context } from "@deepseek-ai/cordis"; +import type {} from "../navigation/service"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import { createContributions } from "../../plugins/contributions"; +import type { Decisions } from "../decisions/service"; +import { AgentsRunner } from "./agents-runner"; +import { originOf } from "./connections"; +import type { AttentionSource } from "./attention-sources"; +import { + createAgentHooks, + dispatchAgentHooks, + type AgentHook, + type AgentHookContext, +} from "./hooks"; +import { bindKeepAwake } from "./keep-awake"; +import type { RuntimeLink } from "./exposed"; +import type { LinkPort } from "./link"; +import type { LocalAgentActivity } from "./local-activity"; +import type { AgentKey } from "./membership"; +import type { AgentsNative } from "./native"; +import { TIMEOUT_LIMIT_MS } from "./runner"; +import { routed, runnerSide } from "./runner-link"; +import { RunnerPlugins } from "./runner-plugins"; +import type { + Agent, + AgentCommunities, + AgentType, + AgentWake, + RegisteredAgentType, +} from "./service"; +import type { AgentProtection } from "../agent-protection/service"; +import { PRESENTED } from "./shown-types"; +import type { AgentStore } from "./state"; +import { setAttention, type AgentRecord } from "./store"; + +/** How often each runner is woken, to start any timer that is due. */ +const TIMER_TICK_MS = 5_000; + +/** A new agent's settings: its type's defaults. */ +export function defaults(pubkey: string, type: RegisteredAgentType) { + const { config, attention = {} } = type.defaults(); + let record: AgentRecord = { pubkey, attention: {}, config }; + for (const [slug, value] of Object.entries(attention)) + record = setAttention(record, slug, value); + return record; +} + +/** Throws unless `type` is a whole agent type: one that runs here, or a + * hosted one. */ +export function checkType(type: AgentType) { + if ( + !/^[a-z0-9][a-z0-9._-]*$/.test(type.id ?? "") || + !type.title || + typeof type.defaults !== "function" + ) + throw new Error("Agent types need an id, a title and defaults"); + if ( + type.timeoutMs !== undefined && + !( + Number.isInteger(type.timeoutMs) && + type.timeoutMs >= 1 && + type.timeoutMs <= TIMEOUT_LIMIT_MS + ) + ) + throw new Error( + `Agent type timeoutMs must be a whole number from 1 to ${TIMEOUT_LIMIT_MS}`, + ); + const { remote } = type; + if ( + remote && + (type.run || + type.models || + type.timeoutMs !== undefined || + (["provision", "activate", "profile", "remove"] as const).some( + (step) => typeof remote[step] !== "function", + )) + ) + throw new Error( + "A hosted agent type has provision, activate, profile and remove, and no run, models or timeoutMs", + ); + // A type that runs here may run in another page than its UI. + const shown = remote + ? undefined + : (PRESENTED.find((field) => type[field] !== undefined) ?? + (type.sessionControls?.scope && "sessionControls.scope") ?? + (type.provider?.DialogBody && "provider.DialogBody") ?? + (type.provider?.DefaultControl && "provider.DefaultControl")); + if (shown) + throw new Error( + `An agent type that runs here gives ${shown} from its plugin's present`, + ); +} + +export type RunnerOptions = Readonly<{ + communities: AgentCommunities; + native: AgentsNative | undefined; + storage: AgentStore; + localActivity: LocalAgentActivity; + decisions: Decisions | undefined; + /** The protection a protected agent's run checks its actions with, from + * the agentProtection service of this page. */ + protection?: (agent: Agent) => AgentProtection | undefined; +}>; + +/** The Runner, with its side of the link on `port`. What a plugin calls + * takes that plugin's `ctx`. */ +export class AgentsRunnerService { + /** The agent types that run here. */ + private readonly types; + private readonly hooks; + private readonly runner: AgentsRunner; + private readonly plugins: RunnerPlugins; + private readonly link: ReturnType; + + constructor(ctx: Context, port: LinkPort, options: RunnerOptions) { + const { communities, storage, native, localActivity, decisions } = options; + this.types = createContributions(ctx); + this.hooks = createAgentHooks(ctx); + this.runner = new AgentsRunner(storage, native, localActivity, decisions, { + open: (origin) => communities.open(origin), + types: () => this.types.snapshot(), + defaults, + // What the Runner asks of the Manager crosses the link. + createAgent: (agent, input) => this.link.createAgent(agent, input), + updateAgent: (agent, input) => this.link.updateAgent(agent, input), + templates: (agent) => this.link.templates(agent), + publishType: (key, type, config) => + this.link.publishType(key, type, config), + finished: (key) => this.link.finished(key), + refresh: () => this.link.refresh(), + navigation: ctx.navigation, + protection: (agent) => options.protection?.(agent), + }); + this.plugins = new RunnerPlugins(ctx, this.runner); + this.link = runnerSide(port, this.runner); + if (native) + ctx.effect(() => + bindKeepAwake(this.runner.runners, (hold) => native.keepAwake(hold)), + ); + ctx.effect(() => { + const stops = [ + this.types.subscribe(() => { + this.hooks.validate(this.types.snapshot()); + this.runner.typesChanged(); + }), + communities.subscribe(() => + this.runner.communitiesChanged(communities.snapshot().viewer), + ), + ]; + // Wakes each runner, which starts any timer that is due. + const tick = setInterval(() => { + this.runner.runners.tick(); + this.runner.connections.reconnect( + originOf(communities.snapshot().selected ?? ""), + ); + this.runner.presence.announce(); + this.runner.runners.catchUp(true); + }, TIMER_TICK_MS); + return () => { + clearInterval(tick); + for (const stop of stops) stop(); + // Nothing the link sends now may start runners again. + this.link.close(); + this.runner.clear(); + }; + }); + } + + register(ctx: Context, type: AgentType) { + checkType(type); + if (type.remote) throw new Error("A hosted agent type runs elsewhere"); + this.types.register(ctx, type as AgentType); + } + registerHook(ctx: Context, hook: AgentHook) { + const adapter = this.types + .snapshot() + .find((type) => type.key === hook.harness); + if (adapter && !adapter.hooks?.events.includes(hook.event)) + throw new Error( + `Hook event ${hook.event} is not supported by ${hook.harness}`, + ); + this.hooks.register(ctx, hook); + } + dispatchHook( + ctx: Context, + harness: string, + event: string, + input: unknown, + context: AgentHookContext, + options: { accepts(output: unknown): boolean; milliseconds: number }, + ) { + const owner = ctx.pluginOwner; + if (!owner || !harness.startsWith(`${owner.id}/`)) + throw new Error( + "Plugins can dispatch hooks only for their own agent types", + ); + return dispatchAgentHooks( + this.hooks.handlers(harness, event), + input, + context, + options.accepts, + options.milliseconds, + ); + } + /** The link to what `ctx`'s plugin's UI shares. */ + shared(ctx: Context) { + const plugin = ctx.pluginOwner?.id; + if (!plugin) throw new Error("Only a plugin reaches what its UI shares"); + const reached = routed(this.link, "plugin", { plugin }); + ctx.effect(() => () => reached.close()); + return reached as RuntimeLink; + } + addTool(ctx: Context, tool: AddedTool) { + this.plugins.addTool(ctx, tool); + } + addAttentionSource(ctx: Context, source: AttentionSource) { + this.plugins.addAttentionSource(ctx, source); + } + attentionProviders = () => this.plugins.attentionProviders(); + tools = () => this.plugins.tools(); + wake(ctx: Context, key: AgentKey, wake: AgentWake) { + const plugin = ctx.pluginOwner?.id; + if (!plugin) throw new Error("Only a plugin wakes an agent"); + this.runner.wake(key, wake, { plugin }); + } + handle(ctx: Context, key: AgentKey) { + const owner = ctx.pluginOwner; + return owner && this.runner.runners.handle(key, owner.id); + } + note(key: AgentKey, channelId: string, text: string, root?: string) { + return this.runner.runners.note(key, channelId, text, root); + } + relay = (key: AgentKey) => this.runner.relay(key); + core = (key: AgentKey, signal?: AbortSignal) => + this.runner.coreMemory.core(key, signal); + /** What this page's agentProtection reads: the agents running here, and + * the types that run here. */ + running = () => this.runner.running(); + agentTypes = () => this.types.snapshot(); + /** Saves `pubkey`'s protection policy once its sessions are retired. */ + protect = ( + pubkey: string, + revision: number, + provider: string, + policy: unknown, + ) => this.runner.protect(pubkey, revision, provider, policy); +} + +/** `ctx.agents` in a page that runs only the Runner: what plugins' agent + * entries use. The members the screens use are the Manager's page's. */ +export class RunnerAgents extends Service { + constructor( + ctx: Context, + private readonly runner: AgentsRunnerService, + ) { + super(ctx, "agents"); + } + register(type: AgentType) { + this.runner.register(this.ctx, type as unknown as AgentType); + } + registerHook(hook: AgentHook) { + this.runner.registerHook(this.ctx, hook); + } + dispatchHook( + harness: string, + event: string, + input: unknown, + context: AgentHookContext, + options: { accepts(output: unknown): boolean; milliseconds: number }, + ) { + return this.runner.dispatchHook( + this.ctx, + harness, + event, + input, + context, + options, + ); + } + shared() { + return this.runner.shared(this.ctx); + } + addTool(tool: AddedTool) { + this.runner.addTool(this.ctx, tool); + } + addAttentionSource(source: AttentionSource) { + this.runner.addAttentionSource(this.ctx, source); + } + attentionProviders = () => this.runner.attentionProviders(); + tools = () => this.runner.tools(); + wake(key: AgentKey, wake: AgentWake) { + this.runner.wake(this.ctx, key, wake); + } + handle(key: AgentKey) { + return this.runner.handle(this.ctx, key); + } + note(key: AgentKey, channelId: string, text: string, root?: string) { + return this.runner.note(key, channelId, text, root); + } + relay = (key: AgentKey) => this.runner.relay(key); + core = (key: AgentKey, signal?: AbortSignal) => this.runner.core(key, signal); +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 928ce63247..065aef36e3 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -6,7 +6,6 @@ import { type ProtectionBinding, } from "../agent-protection/service"; import { createLocalAgentActivity } from "./local-activity"; -import { bindKeepAwake } from "./keep-awake"; import { applySynced, isSynced, resolveSynced } from "./agent-start"; import { type CommunityConfig, @@ -31,19 +30,13 @@ import type { Expected, SnapshotRead } from "./agent-config-artifacts"; import type { revertTemplate, SettingsCheck } from "./agent-config-history"; import type { RevertOutcome } from "./AgentConfigHistory"; import { nativeRelayInfo } from "../relay/native"; -import { - createAgentHooks, - dispatchAgentHooks, - type AgentHook, - type AgentHookContext, -} from "./hooks"; +import type { AgentHook, AgentHookContext } from "./hooks"; import { AgentConfigStore, type SettingsSchema, type Subject, } from "./agent-config"; import type { Mode, Synced } from "./agent-start"; -import { TIMEOUT_LIMIT_MS } from "./runner"; // 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 @@ -77,7 +70,7 @@ import { removeRelayAgent, } from "./relay-removal"; import { relayOrigin } from "../communities/destination"; -import type { Communities } from "../communities/service"; +import type { ClientSnapshot, Communities } from "../communities/service"; import { prepareRemoteAgentAuthorization, type NipOaAuthorization, @@ -121,23 +114,21 @@ import { } from "./native"; import { agentKey, pubkeyOf, type AgentKey } from "./membership"; import { EMPTY, type Member } from "./records"; -import { AgentsRunner, type RunnerStatus } from "./agents-runner"; -import { RunnerPlugins } from "./runner-plugins"; +import type { RunnerStatus } from "./agents-runner"; +import { AgentsRunnerService, checkType, defaults } from "./runner-service"; export { teamPrompt } from "./team-prompt"; import type { Teammate } from "./team-prompt"; -import { managerSide, routed, runnerSide, type TypeInfo } from "./runner-link"; +import { managerSide, type TypeInfo } from "./runner-link"; import type { RuntimeLink } from "./exposed"; -import type { Served } from "./link"; -import { PRESENTED, shownType } from "./shown-types"; +import type { LinkPort, Served } from "./link"; +import { shownType } from "./shown-types"; type ManagerSide = ReturnType; -import { originOf } from "./connections"; -import { AgentPresence } from "./presence"; +import { originOf, retryFailed } from "./connections"; import { readArchiveUnchecked, readTeams, readUnfinished, - setAttention, writeArchiveUnchecked, writeTeams, writeUnfinished, @@ -1282,26 +1273,19 @@ declare module "@deepseek-ai/cordis" { /** Bounds one cleanup attempt, whose relay steps otherwise wait on the outbox. */ const CLEANUP_TIMEOUT_MS = 60_000; const TIMER_TICK_MS = 5_000; -/** How often a failed connection to an unselected community is tried again. */ +/** How often a failed config support check is tried again. */ const RETRY_MS = 30_000; -/** A new agent's settings: its type's defaults. */ -function defaults(pubkey: string, type: RegisteredAgentType) { - const { config, attention = {} } = type.defaults(); - let record: AgentRecord = { pubkey, attention: {}, config }; - for (const [slug, value] of Object.entries(attention)) - record = setAttention(record, slug, value); - return record; -} const message = (error: unknown) => String(error instanceof Error ? error.message : error); export { fromOwner } from "./owner-attestation"; /** What Agents needs of the communities: the viewer, the selected one, and * a connection to any joined one without selecting it. */ -export type AgentCommunities = Pick< - Communities, - "snapshot" | "subscribe" | "open" ->; +export type AgentCommunities = Readonly<{ + snapshot(): Readonly>; + subscribe: Communities["subscribe"]; + open: Communities["open"]; +}>; type Binding = { scope: string; origin: string; @@ -1319,8 +1303,6 @@ type Bound = { relay: RelayData; stop: () => void; binding?: Binding | undefined; - /** When its failed connection was last tried again. */ - retried?: number; }; /** A saved identity, which is the agent's membership in one community, with * that membership's key. */ @@ -1379,16 +1361,8 @@ const wssOf = (origin: string) => relayOrigin(origin).replace(/^https:/, "wss:"); export class AgentsService extends Service implements Agents { - private readonly hookContributions; registerHook(hook: AgentHook) { - const adapter = this.contributions - .snapshot() - .find((type) => type.key === hook.harness); - if (adapter && !adapter.hooks?.events.includes(hook.event)) - throw new Error( - `Hook event ${hook.event} is not supported by ${hook.harness}`, - ); - this.hookContributions.register(this.ctx, hook); + this.here().registerHook(this.ctx, hook); } dispatchHook( harness: string, @@ -1397,19 +1371,16 @@ export class AgentsService extends Service implements Agents { context: AgentHookContext, options: { accepts(output: unknown): boolean; milliseconds: number }, ) { - const owner = this.ctx.pluginOwner; - if (!owner || !harness.startsWith(`${owner.id}/`)) - throw new Error( - "Plugins can dispatch hooks only for their own agent types", - ); - return dispatchAgentHooks( - this.hookContributions.handlers(harness, event), + return this.here().dispatchHook( + this.ctx, + harness, + event, input, context, - options.accepts, - options.milliseconds, + options, ); } + /** The hosted agent types; the Runner keeps those that run here. */ private readonly contributions; private readonly tabContributions; private readonly attentionControls; @@ -1451,12 +1422,9 @@ export class AgentsService extends Service implements Agents { private state: AgentsSnapshot; /** By origin, so each agent listens in its own community, selected or not. */ private bound = new Map(); - /** Runs the agents, and owns the agent storage they keep. */ - private readonly runner: AgentsRunner; - /** The tools and attention sources plugins add for agents: the Runner's. */ - private readonly plugins: RunnerPlugins; - /** The Runner's side of the link between them. */ - private readonly runnerLink: ReturnType; + /** Runs the agents, and owns the agent storage they keep, with what + * plugins' agent entries add for them, when it runs in this page. */ + private readonly runner: AgentsRunnerService | undefined; /** The Manager's side: what it sends the Runner, and what the Runner * decided, as the link last sent it. */ private readonly decided: ManagerSide; @@ -1509,36 +1477,6 @@ export class AgentsService extends Service implements Agents { private shown = new Map(); /** Each bound community's config channels, while config sync is on. */ private readonly communityConfigs = new Map(); - /** Says which agents this copy runs in each community. */ - private readonly presence = new AgentPresence({ - running: () => { - const online = new Map>(); - for (const agent of this.state.running) { - const origin = originOf(agent.relay); - if (origin && this.bindingOf(agent) && this.decided.runs(agent.key)) - online.set( - origin, - (online.get(origin) ?? new Set()).add(agent.pubkey), - ); - } - return online; - }, - communities: () => - [...this.bound].map( - ([origin, entry]) => [origin, entry.binding] as const, - ), - // Each pubkey said online there is one membership's, the one there. - yielded: (origin, pubkey) => { - const key = this.identities.find( - (identity) => - identity.pubkey === pubkey && - !identity.deleted && - originOf(identity.relay) === origin, - )?.key; - return !!key && this.decided.yielded(key); - }, - }); - private readonly protection: AgentProtectionService; constructor( ctx: Context, @@ -1555,10 +1493,12 @@ export class AgentsService extends Service implements Agents { /** Which artifact types an agent has synced, from its config channel * binding. Unset, every agent starts `local`. */ private readonly bindingFor?: (agent: Agent) => ConfigBinding, + /** The Manager's end of the link to the Runner in another page. Unset, + * the Runner runs in this page. */ + runnerPort?: LinkPort, ) { super(ctx, "agents"); this.contributions = createContributions(ctx); - this.hookContributions = createAgentHooks(ctx); this.tabContributions = createContributions(ctx); this.attentionControls = createContributions<{ id: string }>(ctx); this.configStore = new AgentConfigStore(storage); @@ -1588,68 +1528,29 @@ export class AgentsService extends Service implements Agents { policy, ); }, - update: async (pubkey, revision, provider, policy) => { - const setProtection = this.native?.setProtection?.bind(this.native); - if (!setProtection) - throw new Error("Native agent protection is unavailable"); - const members = this.identities.filter( - (agent) => agent.pubkey === pubkey && !agent.deleted, - ); - const agent = members[0]; - if (!agent) throw new Error("No such agent on this device"); - const type = this.types().find((type) => type.key === agent.type); - if (!type?.retire) - throw new Error( - "This runtime cannot safely change protection while Buzz is open", - ); - await this.runner.runners.retype(agent.key, agent.name, async () => { - for (const member of members) await type.retire?.(member.key); - await setProtection(pubkey, revision, provider, policy); - await this.refresh(); - }); - }, + // The Runner retires its sessions, then saves the policy. + update: (pubkey, revision, provider, policy) => + this.decided.protect(pubkey, revision, provider, policy), }, ); - // A claim names the protection the agent is saved with; a refused one - // may mean it changed. - const custody = native && { - ...native, - claim: async (pubkey: string) => { - try { - return await native.claim( - pubkey, - this.identities.find( - (agent) => agent.pubkey === pubkey && !agent.deleted, - )?.protectionRevision ?? 0, - ); - } catch (error) { - void this.refresh().catch(() => undefined); - throw error; - } - }, - }; this.shares = createContributions(ctx); - this.runner = new AgentsRunner(storage, custody, localActivity, decisions, { - changed: () => this.update(), - open: (origin) => this.communities.open(origin), - types: () => this.contributions.snapshot(), - defaults, - // What the Runner asks of the Manager crosses the link. - createAgent: (agent, input) => this.runnerLink.createAgent(agent, input), - updateAgent: (agent, input) => this.runnerLink.updateAgent(agent, input), - templates: (agent) => this.runnerLink.templates(agent), - publishType: (key, type, config) => - this.runnerLink.publishType(key, type, config), - finished: (key) => this.runnerLink.finished(key), - navigation: ctx.navigation, - protection: (agent) => this.protection.bind(agent), - }); // The Runner decides from the memberships the Manager sends over the - // link, and the screens show what it sends back. Only JSON crosses, as it - // will between pages. - const channel = new MessageChannel(); - this.runnerLink = runnerSide(channel.port1, this.runner); - this.decided = managerSide(channel.port2, { + // link, and the screens show what it sends back. Only JSON crosses, in + // this page as between pages. + let port = runnerPort; + if (!port) { + const channel = new MessageChannel(); + this.runner = new AgentsRunnerService(ctx, channel.port1, { + communities, + native, + storage, + localActivity, + decisions, + protection: (agent) => this.protection.bind(agent), + }); + port = channel.port2; + } + this.decided = managerSide(port, { members: () => ({ loaded: this.loaded, identities: this.identities, @@ -1781,16 +1682,12 @@ export class AgentsService extends Service implements Agents { if (identity) await this.publishType(identity, type, config as object); }, finished: (key) => this.retakeType(key), + refresh: () => this.refresh(), shared: (plugin) => this.shares.snapshot().find((entry) => entry.pluginId === plugin) ?.served, subscribeShared: (listener) => this.shares.subscribe(listener), }); - this.plugins = new RunnerPlugins(ctx, this.runner); - if (native) - ctx.effect(() => - bindKeepAwake(this.runner.runners, (hold) => native.keepAwake(hold)), - ); this.unfinished = Object.freeze(readUnfinished(storage)); this.unarchiving = readArchiveUnchecked(storage); this.teamList = Object.freeze(readTeams(storage)); @@ -1811,12 +1708,12 @@ export class AgentsService extends Service implements Agents { }); ctx.effect(() => { const stops = [ - this.contributions.subscribe(() => { - this.hookContributions.validate(this.contributions.snapshot()); - this.runner.typesChanged(); - this.update(); - this.notify(); - }), + ...[this.contributions, this.presentations].map((registry) => + registry.subscribe(() => { + this.update(); + this.notify(); + }), + ), // Every agent's Attention turns on or off with its controls. this.attentionControls.subscribe(() => { this.update(); @@ -1824,11 +1721,6 @@ export class AgentsService extends Service implements Agents { }), // The Runner holds Attention until startup is over. this.ctx.pluginStatus.subscribe(() => this.decided.changed()), - this.presentations.subscribe(() => { - this.runner.typesChanged(); - this.update(); - this.notify(); - }), // A new state object, so views reading `addedTabs()` render again. this.tabContributions.subscribe(() => { this.state = Object.freeze({ ...this.state }); @@ -1850,7 +1742,6 @@ export class AgentsService extends Service implements Agents { this.updates = updates; this.notify(); } - this.runner.communitiesChanged(viewer); this.watch(); // Synced teams are the signed-in account's. if (viewer !== this.teamsFor) this.teamsChanged(); @@ -1862,24 +1753,14 @@ export class AgentsService extends Service implements Agents { this.watch(); void this.refresh(); this.updateClassifier(); - // Wakes each runner, which starts any timer that is due. - const tick = setInterval(() => { - this.runner.runners.tick(); - this.reconnect(); - this.presence.announce(); - this.runner.runners.catchUp(true); - }, TIMER_TICK_MS); + const tick = setInterval(() => this.reconnect(), TIMER_TICK_MS); return () => { clearInterval(tick); // A list still being read is stale: it must not reopen communities. this.listed++; for (const stop of stops) stop(); for (const origin of [...this.bound.keys()]) this.unbind(origin); - // Nothing the link sends now may start runners again. - this.runnerLink.close(); this.decided.close(); - this.runner.clear(); - this.presence.clear(); // Clear what is shown without reconciling, which would make new runners. this.state = Object.freeze({ ...this.state, @@ -1894,48 +1775,11 @@ export class AgentsService extends Service implements Agents { } register(type: AgentType) { - if ( - !/^[a-z0-9][a-z0-9._-]*$/.test(type.id ?? "") || - !type.title || - typeof type.defaults !== "function" - ) - throw new Error("Agent types need an id, a title and defaults"); - if ( - type.timeoutMs !== undefined && - !( - Number.isInteger(type.timeoutMs) && - type.timeoutMs >= 1 && - type.timeoutMs <= TIMEOUT_LIMIT_MS - ) - ) - throw new Error( - `Agent type timeoutMs must be a whole number from 1 to ${TIMEOUT_LIMIT_MS}`, - ); - const { remote } = type; - if ( - remote && - (type.run || - type.models || - type.timeoutMs !== undefined || - (["provision", "activate", "profile", "remove"] as const).some( - (step) => typeof remote[step] !== "function", - )) - ) - throw new Error( - "A hosted agent type has provision, activate, profile and remove, and no run, models or timeoutMs", - ); - // A type that runs here may run in another page than its UI. - const shown = remote - ? undefined - : (PRESENTED.find((field) => type[field] !== undefined) ?? - (type.sessionControls?.scope && "sessionControls.scope") ?? - (type.provider?.DialogBody && "provider.DialogBody") ?? - (type.provider?.DefaultControl && "provider.DefaultControl")); - if (shown) - throw new Error( - `An agent type that runs here gives ${shown} from its plugin's present`, - ); - this.contributions.register(this.ctx, type as unknown as AgentType); + const registered = type as unknown as AgentType; + checkType(registered); + // A type that runs here is the Runner's; the Manager keeps hosted ones. + if (registered.remote) this.contributions.register(this.ctx, registered); + else this.here().register(this.ctx, registered); } present( typeId: string, @@ -1959,11 +1803,12 @@ export class AgentsService extends Service implements Agents { this.shares.register(this.ctx, { id: "shared", served }); } shared() { - const plugin = this.ctx.pluginOwner?.id; - if (!plugin) throw new Error("Only a plugin reaches what its UI shares"); - const reached = routed(this.runnerLink, "plugin", { plugin }); - this.ctx.effect(() => () => reached.close()); - return reached as RuntimeLink; + return this.here().shared(this.ctx); + } + /** The Runner, for what plugins' agent entries use; they run in its page. */ + private here() { + if (!this.runner) throw new Error("Agent entries run in the agents page"); + return this.runner; } /** Whether a plugin presents this type but the Runner has not sent it yet: * the plugin is on and the type is on its way over the link. */ @@ -2009,10 +1854,7 @@ export class AgentsService extends Service implements Agents { known.runtime?.close(); this.presentedTypes.delete(info); } - const result = Object.freeze([ - ...local, - ...registered.filter((type) => type.remote), - ]); + const result = Object.freeze([...local, ...registered]); this.merged = { infos, registered, shown, result }; return result; }; @@ -2032,13 +1874,13 @@ export class AgentsService extends Service implements Agents { /** Whether a plugin that shows Attention is on, so Attention may run. */ private attentionShown = () => this.attentionControls.snapshot().length > 0; addTool(tool: AddedTool) { - this.plugins.addTool(this.ctx, tool); + this.here().addTool(this.ctx, tool); } addAttentionSource(source: import("./attention-sources").AttentionSource) { - this.plugins.addAttentionSource(this.ctx, source); + this.here().addAttentionSource(this.ctx, source); } - attentionProviders = () => this.plugins.attentionProviders(); - tools = () => this.plugins.tools(); + attentionProviders = () => this.here().attentionProviders(); + tools = () => this.here().tools(); presentWatch = (watch: import("./attention").SourceWatch, title: string) => { // Whichever page runs the source, its UI entry presents it here. const shown = this.sourcePresentations @@ -2056,9 +1898,7 @@ export class AgentsService extends Service implements Agents { } }; wake(key: AgentKey, wake: AgentWake) { - const plugin = this.ctx.pluginOwner?.id; - if (!plugin) throw new Error("Only a plugin wakes an agent"); - this.runner.wake(key, wake, { plugin }); + this.here().wake(this.ctx, key, wake); } snapshot = () => this.state; subscribe = (listener: () => void) => { @@ -2084,15 +1924,13 @@ export class AgentsService extends Service implements Agents { find = (pubkey: string) => this.state.agents.find((agent) => agent.pubkey === pubkey); handle(key: AgentKey) { - const owner = this.ctx.pluginOwner; - return owner && this.runner.runners.handle(key, owner.id); + return this.here().handle(this.ctx, key); } note(key: AgentKey, channelId: string, text: string, root?: string) { - return this.runner.runners.note(key, channelId, text, root); + return this.here().note(key, channelId, text, root); } - relay = (key: AgentKey) => this.runner.relay(key); - core = (key: AgentKey, signal?: AbortSignal) => - this.runner.coreMemory.core(key, signal); + relay = (key: AgentKey) => this.here().relay(key); + core = (key: AgentKey, signal?: AbortSignal) => this.here().core(key, signal); private localCache: | { @@ -4463,14 +4301,7 @@ export class AgentsService extends Service implements Agents { const selected = originOf(this.communities.snapshot().selected ?? ""); const now = Date.now(); for (const [origin, entry] of this.bound) - if ( - origin !== selected && - entry.relay.snapshot().status === "error" && - now - (entry.retried ?? 0) >= RETRY_MS - ) { - entry.retried = now; - entry.relay.retry(); - } + if (origin !== selected) retryFailed(entry.relay, now); for (const [origin, { binding }] of this.bound) if ( typeof binding?.support === "object" && @@ -4639,7 +4470,6 @@ export class AgentsService extends Service implements Agents { } else if (held !== this.held || types !== this.shownTypes) this.notify(); this.held = held; this.shownTypes = types; - this.presence.announce(); for (const [origin, config] of this.communityConfigs) void config.attach( running diff --git a/src/features/agents/state.test.ts b/src/features/agents/state.test.ts index 252d6403eb..87681d047b 100644 --- a/src/features/agents/state.test.ts +++ b/src/features/agents/state.test.ts @@ -1,5 +1,11 @@ import { expect, it, vi } from "vitest"; -import { fileStore, loadAgentStorage, MOVED_KEY } from "./state"; +import { + agentStorage, + fileStore, + loadAgentStorage, + MOVED_KEY, + readAgentStorage, +} from "./state"; import { memoryStorage } from "./test-fakes"; import { SyncedTeams, teamSources } from "./team-config"; @@ -214,3 +220,16 @@ it("puts back a key whose write failed, so it is not taken as saved and can be s expect(native.saved.get("a")).toBe("new"); expect(storage.get("a")).toBe("new"); }); + +it("reads agent storage another page loaded, without moving localStorage, and fails rather than use another copy", async () => { + const native = files({ [MOVED_KEY]: "1", "buzz.agent.v1:aa": "kept" }); + const storage = await readAgentStorage(native); + expect(storage.get("buzz.agent.v1:aa")).toBe("kept"); + expect(agentStorage()).toBe(storage); + await storage.set("buzz.agent.v1:aa", "changed"); + expect(native.saved.get("buzz.agent.v1:aa")).toBe("changed"); + expect(native.set).toHaveBeenCalledTimes(1); + const unreadable = files(); + unreadable.load.mockRejectedValue(new Error("No access")); + await expect(readAgentStorage(unreadable)).rejects.toThrow("No access"); +}); diff --git a/src/features/agents/state.ts b/src/features/agents/state.ts index d874236473..3017ce58c3 100644 --- a/src/features/agents/state.ts +++ b/src/features/agents/state.ts @@ -155,6 +155,25 @@ export async function loadAgentStorage( } } +/** Whether agent storage is the device's files, as `loadAgentStorage` read + * them, rather than localStorage. */ +export const agentStorageInFiles = () => loaded !== undefined; + +/** Reads agent storage that another page of this app already loaded, and + * moved localStorage into: this page moves nothing. It rejects if the files + * cannot be read, so this page never writes a copy the other does not. */ +export async function readAgentStorage( + files: Native | undefined = isTauri() ? native : undefined, +): Promise { + if (!files) throw new Error("Agent storage files need the desktop app"); + loaded = fileStore( + new Map(Object.entries(await files.load())), + files.set, + files.add, + ); + return loaded; +} + /** Agent storage held in memory and saved through `save`, one write at a * time, in order. A failed write puts back the key's saved value, so the * copy never keeps what is not saved and the same value can be set again. */ diff --git a/src/features/communities/page.test.ts b/src/features/communities/page.test.ts new file mode 100644 index 0000000000..80ae245356 --- /dev/null +++ b/src/features/communities/page.test.ts @@ -0,0 +1,103 @@ +import { Context } from "@deepseek-ai/cordis"; +import { afterEach, expect, it, vi } from "vitest"; +import type { ReadTransport } from "../relay/transport"; +import { keypair } from "../relay/testing"; +import { communityDestination } from "./destination"; +import { pageCommunities } from "./page"; + +const relays = vi.hoisted(() => ({ provide: vi.fn() })); +vi.mock("../relay/service", async (original) => { + const actual = await original(); + relays.provide.mockImplementation(actual.provideRelay); + return { ...actual, provideRelay: relays.provide }; +}); + +const roots: Context[] = []; +afterEach(async () => { + for (const ctx of roots.splice(0)) await ctx.fiber.dispose(); + relays.provide.mockClear(); +}); + +const alpha = communityDestination("wss://alpha.example").id; +const beta = communityDestination("wss://beta.example").id; +function setup() { + const viewer = keypair().pubkey; + const relay = keypair().pubkey; + const connected: string[] = []; + const ctx = new Context(); + roots.push(ctx); + const communities = pageCommunities(ctx, async (id) => { + connected.push(id); + return { + viewer, + relayAuthor: relay, + scope: id, + media: () => undefined, + query: async () => [], + } satisfies ReadTransport; + }); + const changes = vi.fn(); + communities.subscribe(changes); + return { viewer, communities, changes, connected }; +} + +it("opens nothing until the page that shows the communities sends them", () => { + const { communities } = setup(); + expect(communities.snapshot()).toEqual({ memberships: [], selected: null }); + expect(communities.open(alpha)).toBeUndefined(); + // Signed out there: no viewer to connect as. + communities.follow({ memberships: [alpha], selected: alpha }); + expect(communities.open(alpha)).toBeUndefined(); +}); + +it("connects a joined community for its agents with a session that keeps nothing on the device", async () => { + const { viewer, communities, changes, connected } = setup(); + communities.follow({ viewer, memberships: [alpha, beta], selected: beta }); + expect(changes).toHaveBeenCalledTimes(1); + expect(communities.snapshot()).toEqual({ + viewer, + memberships: [alpha, beta], + selected: beta, + }); + const relay = communities.open("wss://Alpha.example/"); + expect(relay).toBeDefined(); + expect(communities.open(alpha)).toBe(relay); + expect(communities.open("wss://gamma.example")).toBeUndefined(); + await vi.waitFor(() => expect(connected).toEqual([alpha])); + // No presence, nothing restored, and its sends and history in memory only. + const [, , presence, , , saved, , options] = + relays.provide.mock.calls[0] ?? []; + expect({ presence, saved, options }).toEqual({ + presence: undefined, + saved: undefined, + options: { ephemeral: true }, + }); +}); + +it("lets go of a community that is left, and of every one when the viewer changes", () => { + const { viewer, communities } = setup(); + communities.follow({ viewer, memberships: [alpha, beta], selected: alpha }); + const first = communities.open(alpha); + const second = communities.open(beta); + communities.follow({ viewer, memberships: [beta], selected: null }); + expect(communities.open(alpha)).toBeUndefined(); + expect(communities.open(beta)).toBe(second); + communities.follow({ viewer, memberships: [alpha, beta], selected: null }); + expect(communities.open(alpha)).not.toBe(first); + communities.follow({ + viewer: keypair().pubkey, + memberships: [alpha, beta], + selected: null, + }); + expect(communities.open(beta)).not.toBe(second); +}); + +it("tells its listeners only of what changed, and never selects a community it is not in", () => { + const { viewer, communities, changes } = setup(); + communities.follow({ viewer, memberships: [alpha], selected: alpha }); + communities.follow({ viewer, memberships: [alpha], selected: alpha }); + expect(changes).toHaveBeenCalledTimes(1); + communities.follow({ viewer, memberships: [alpha], selected: beta }); + expect(changes).toHaveBeenCalledTimes(2); + expect(communities.snapshot().selected).toBeNull(); +}); diff --git a/src/features/communities/page.ts b/src/features/communities/page.ts new file mode 100644 index 0000000000..07df206350 --- /dev/null +++ b/src/features/communities/page.ts @@ -0,0 +1,126 @@ +// FOUNDATION: Communities as a page that shows nothing follows them. The +// page that shows them owns the device's record: its memberships, selection +// and viewer. It sends them here each time they change. This page connects +// to the joined communities for its agents with sessions of its own that +// keep nothing on the device, so the other page stays the one owner of what +// the device saves, shows and clears. +import { Context } from "@deepseek-ai/cordis"; +import type { LocalAgentSource } from "../agents/local-activity"; +import type { LocalAgents } from "../agents/local"; +import type { IdentityNames } from "../identity-names/service"; +import { provideRelay, type RelayData } from "../relay/service"; +import type { ReadTransport } from "../relay/transport"; +import { communityDestination } from "./destination"; + +/** What this page follows of the communities. */ +export type FollowedCommunities = Readonly<{ + viewer?: string; + /** The joined communities' ids. */ + memberships: readonly string[]; + selected: string | null; +}>; + +export function pageCommunities( + ctx: Context, + connect: (id: string, signal: AbortSignal) => Promise, + options: Readonly<{ + identityNames?: IdentityNames; + agentChoices?: LocalAgents; + localAgents?: LocalAgentSource; + }> = {}, +) { + let state: FollowedCommunities = Object.freeze({ + memberships: [], + selected: null, + }); + let disposed = false; + const listeners = new Set<() => void>(); + // Each session owns a scope so leaving can dispose exactly that one. + const sessions = new Map< + string, + Readonly<{ relay: RelayData; scope: Context }> + >(); + const release = (id: string) => { + const session = sessions.get(id); + sessions.delete(id); + return session?.scope.fiber.dispose().catch((error: unknown) => { + console.warn(`Couldn't dispose the agents' session for ${id}`, error); + }); + }; + const idOf = (id: string) => { + try { + return communityDestination(id).id; + } catch { + return undefined; + } + }; + ctx.effect(() => () => { + disposed = true; + listeners.clear(); + return Promise.all([...sessions.keys()].map(release)).then(() => {}); + }); + return { + snapshot: () => state, + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + /** A joined community's session, for the viewer that joined it; it lives + * until the community is left or the viewer changes. */ + open(id: string): RelayData | undefined { + const key = idOf(id); + if (disposed || !key || !state.viewer || !state.memberships.includes(key)) + return undefined; + let session = sessions.get(key); + if (!session) { + const scope = new Context(); + session = Object.freeze({ + scope, + relay: provideRelay( + scope, + (signal) => connect(key, signal), + undefined, + options.identityNames, + options.agentChoices, + undefined, + options.localAgents, + { ephemeral: true }, + ), + }); + sessions.set(key, session); + } + return session.relay; + }, + /** What the page that shows the communities has now. */ + follow(next: FollowedCommunities) { + if (disposed) return; + const memberships = [ + ...new Set(next.memberships.flatMap((id) => idOf(id) ?? [])), + ]; + const selected = + next.selected === null ? null : (idOf(next.selected) ?? null); + const followed: FollowedCommunities = Object.freeze({ + ...(next.viewer !== undefined && { viewer: next.viewer }), + memberships: Object.freeze(memberships), + selected: selected && memberships.includes(selected) ? selected : null, + }); + if ( + followed.viewer === state.viewer && + followed.selected === state.selected && + followed.memberships.length === state.memberships.length && + followed.memberships.every( + (id, index) => state.memberships[index] === id, + ) + ) + return; + const viewerChanged = followed.viewer !== state.viewer; + state = followed; + for (const id of [...sessions.keys()]) + if (viewerChanged || !memberships.includes(id)) void release(id); + for (const listener of [...listeners]) listener(); + }, + }; +} +export type PageCommunities = ReturnType; diff --git a/src/features/relay/service.ts b/src/features/relay/service.ts index 4b7395ed4d..bc858a9607 100644 --- a/src/features/relay/service.ts +++ b/src/features/relay/service.ts @@ -4,6 +4,7 @@ import type { LocalAgents } from "../agents/local"; import type { IdentityNames } from "../identity-names/service"; import type { PresenceActivity } from "../presence/activity"; import type { Context } from "@deepseek-ai/cordis"; +import type { OutboxStorage } from "./outbox"; import { createRelaySession, type RelaySession } from "./session"; import { createHeadPersistence } from "./persistence"; import { @@ -49,17 +50,26 @@ declare module "@deepseek-ai/cordis" { } } +/** Sends that live only as long as the page. */ +const PAGE_OUTBOX: OutboxStorage = { load: () => [], save() {} }; + /** App-owned session. Plugins receive read models, never a signer or transport. - * Reconnection replaces the session and invalidates all old async work. */ + * Reconnection replaces the session and invalidates all old async work. + * + * An `ephemeral` session is for a page that shows nothing, beside the page + * that does: it keeps its sends and history in memory only, and restores + * nothing, so it never adds to or clears what the other page saves. */ export function provideRelay( ctx: Context, connect?: (signal: AbortSignal) => Promise, presenceActivity?: PresenceActivity, identityNames?: IdentityNames, agentChoices?: LocalAgents, - resume?: { viewer: string; scope: string }, + saved?: { viewer: string; scope: string }, localAgents?: LocalAgentSource, + { ephemeral = false }: { ephemeral?: boolean } = {}, ) { + const resume = ephemeral ? undefined : saved; let disposed = false; let generation = 0; let controller: AbortController | undefined; @@ -261,10 +271,14 @@ export function provideRelay( "selected-channel", undefined, ), - persistence: createHeadPersistence( - transport.viewer, - transportOrigin(transport), - ), + ...(ephemeral + ? { outboxStorage: PAGE_OUTBOX } + : { + persistence: createHeadPersistence( + transport.viewer, + transportOrigin(transport), + ), + }), }); await successor.restore(); if ( diff --git a/src/features/relay/startup-service.test.ts b/src/features/relay/startup-service.test.ts index f76b00c0d9..7c5e3e1b52 100644 --- a/src/features/relay/startup-service.test.ts +++ b/src/features/relay/startup-service.test.ts @@ -8,6 +8,11 @@ import { provideRelay } from "./service"; const storage = vi.hoisted(() => ({ create: vi.fn() })); vi.mock("./persistence", () => ({ createHeadPersistence: storage.create })); +const journal = vi.hoisted(() => ({ create: vi.fn() })); +vi.mock("./outbox", async (original) => ({ + ...(await original()), + browserOutboxStorage: journal.create, +})); const roots: Context[] = []; afterEach(async () => { for (const ctx of roots.splice(0)) await ctx.fiber.dispose(); @@ -268,3 +273,43 @@ it("disconnect during a coalesced cache clear does not reconnect", async () => { expect(data.snapshot().status).toBe("disconnected"); expect(connections).toHaveLength(1); }); +it("an ephemeral session saves, restores and clears nothing on the device", async () => { + const { source } = setup(); + storage.create.mockClear(); + journal.create.mockReset(); + journal.create.mockImplementation(() => ({ load: () => [], save() {} })); + const writing: ReadTransport = { + ...source, + writer: { + kinds: [9], + sign: async () => { + throw new Error("unused"); + }, + publish: async () => {}, + }, + }; + const open = (ephemeral: boolean) => { + const ctx = new Context(); + roots.push(ctx); + return provideRelay( + ctx, + async () => writing, + undefined, + undefined, + undefined, + { viewer: source.viewer, scope: "primary" }, + undefined, + { ephemeral }, + ); + }; + const page = open(true); + await vi.waitFor(() => expect(page.snapshot().status).toBe("ready")); + await page.clearCache(); + expect(storage.create).not.toHaveBeenCalled(); + expect(journal.create).not.toHaveBeenCalled(); + // The same session, kept on the device, uses both. + const kept = open(false); + await vi.waitFor(() => expect(kept.snapshot().status).toBe("ready")); + expect(storage.create).toHaveBeenCalled(); + expect(journal.create).toHaveBeenCalled(); +}); diff --git a/src/plugins/manager.ts b/src/plugins/manager.ts index fcc71a55bc..96b5dd0d96 100644 --- a/src/plugins/manager.ts +++ b/src/plugins/manager.ts @@ -1,6 +1,6 @@ import { createPluginStorage, type PluginStorage } from "./storage"; import * as React from "react"; -import { PluginRuntime } from "./runtime"; +import { type PluginParts, PluginRuntime } from "./runtime"; import type { Context } from "@deepseek-ai/cordis"; import type { PluginManifest, PluginModule } from "./api"; import { createModuleLoader } from "./modules"; @@ -38,6 +38,7 @@ export function createPluginManager( { bundled, builtIn = [], + parts = "both", storage = createPluginStorage(() => bundled.map(({ manifest, enabledByDefault = true }) => ({ manifest, @@ -54,6 +55,8 @@ export function createPluginManager( /** Host-owned contributions, independent of optional-plugin storage. */ builtIn?: readonly BundledPlugin[]; storage?: PluginStorage | undefined; + /** Which part of each plugin this page runs. */ + parts?: PluginParts; }, ) { ctx.provide("react", React); @@ -85,6 +88,7 @@ export function createPluginManager( () => configuration.status === "loading" || (configuration.status === "recovery" && !configuration.canReset), + parts, ); let busy = false; let catalogPending = false; diff --git a/src/plugins/runtime.test.mjs b/src/plugins/runtime.test.mjs index 8a58b9cfc9..37ec488eda 100644 --- a/src/plugins/runtime.test.mjs +++ b/src/plugins/runtime.test.mjs @@ -29,7 +29,7 @@ function deferred() { }); return { promise, resolve }; } -function harness(load, timeoutMs) { +function harness(load, timeoutMs, parts) { let states = {}; const waiters = new Set(); const root = new Context(); @@ -37,6 +37,8 @@ function harness(load, timeoutMs) { root, async (plugin) => ({ inject: ["pages"], ...(await load(plugin)) }), timeoutMs, + undefined, + parts, ); runtime.subscribe(() => { states = runtime.snapshot(); @@ -156,6 +158,45 @@ test("an agent entry has its own dependencies and the plugin waits for both part } }); +test("a page runs only the part of each plugin it is for", async () => { + const started = []; + const load = async (info) => ({ + apply() { + started.push(`${info.manifest.id} ui`); + }, + ...(info.manifest.id === "example.agent" && { + agent: { + apply(ctx) { + assert.equal(ctx.pluginOwner.id, "example.agent"); + started.push(`${info.manifest.id} agent`); + }, + }, + }), + }); + const plugins = [plugin("example.agent"), plugin("example.page")]; + const active = (s) => + plugins.every(({ manifest }) => s[manifest.id]?.status === "active"); + const agents = harness(load, undefined, "agent"); + const screens = harness(load, undefined, "ui"); + try { + agents.runtime.reconcile(plugins); + screens.runtime.reconcile(plugins); + await agents.wait(active); + await screens.wait(active); + assert.deepEqual(started.sort(), [ + "example.agent agent", + "example.agent ui", + "example.page ui", + ]); + // Nothing of example.page runs in the agents page, yet it is active + // there, so what the screens' page adds as it is admitted. + assert.equal(agents.runtime.isActive("example.page", "one"), true); + } finally { + await agents.dispose(); + await screens.dispose(); + } +}); + test("a failed agent entry fails its plugin and stops its other part", async () => { const h = harness(async () => ({ apply(ctx) { diff --git a/src/plugins/runtime.ts b/src/plugins/runtime.ts index 7a20e2fe5e..0670ba16dd 100644 --- a/src/plugins/runtime.ts +++ b/src/plugins/runtime.ts @@ -21,6 +21,11 @@ type Entry = { predecessor: Promise; }; +/** Which parts of each plugin this page runs: both, its UI entry, or its + * agent entry. A plugin with nothing to run here is active once loaded, so + * what another page adds under its identity is admitted here. */ +export type PluginParts = "both" | "ui" | "agent"; + // Cordis 4.0.2 erases its FiberState const enum from the published JavaScript. // Keep the version-specific values here, at the lifecycle adapter boundary. const cordisState = { active: 2, failed: 3, disposed: 4 } as const; @@ -35,6 +40,7 @@ export class PluginRuntime { private readonly load: (plugin: PluginInfo) => Promise; private readonly listeners = new Set<() => void>(); private readonly timeoutMs: number; + private readonly parts: PluginParts; constructor( root: Context, @@ -42,9 +48,11 @@ export class PluginRuntime { timeoutMs = 10_000, /** Whether the manager is still reading which plugins to run. */ configuring = () => false, + parts: PluginParts = "both", ) { this.root = root; this.load = load; + this.parts = parts; root.provide("pluginStatus", { isActive: this.isActive, subscribe: this.subscribe, @@ -161,10 +169,14 @@ export class PluginRuntime { }), }); const id = entry.plugin.manifest.id; - // Both parts run in this page for now; the plugin is active once both are. + // The plugin is active once each part this page runs is. const parts = [ - { name: id, inject: module.inject, apply: module.apply }, - ...(module.agent ? [{ name: `${id}/agent`, ...module.agent }] : []), + ...(this.parts !== "agent" + ? [{ name: id, inject: module.inject, apply: module.apply }] + : []), + ...(module.agent && this.parts !== "ui" + ? [{ name: `${id}/agent`, ...module.agent }] + : []), ]; entry.fibers = parts.map(({ name, inject, apply }) => scope.plugin({ diff --git a/vite.config.ts b/vite.config.ts index ca28165464..d3bd5461d5 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -159,6 +159,7 @@ export default defineConfig(async ({ command, mode }) => { input: { main: fileURLToPath(new URL("index.html", import.meta.url)), huddle: fileURLToPath(new URL("huddle.html", import.meta.url)), + agents: fileURLToPath(new URL("agents.html", import.meta.url)), }, }, },