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/src-tauri/build.rs b/src-tauri/build.rs index 2d602d2237..072baa65ed 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -111,6 +111,8 @@ fn main() { "media_copy_image", "media_stream_base", "get_os_idle_seconds", + "agents_port_listen", + "agents_port_send", "workspace_pick_folder", "repositories_folder_validate", "plugin_import_folder", diff --git a/src-tauri/capabilities/agents.json b/src-tauri/capabilities/agents.json new file mode 100644 index 0000000000..70a10d6df3 --- /dev/null +++ b/src-tauri/capabilities/agents.json @@ -0,0 +1,102 @@ +{ + "$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-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-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 c38b62903c..92eeac8964 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -161,6 +161,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 08954df231..16a7a4a9c4 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, Published, Removal, Secret, CREDENTIALS_BUSY, @@ -88,11 +89,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 @@ -140,8 +143,8 @@ 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; let host = self.clone(); @@ -150,13 +153,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())?; @@ -268,9 +271,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. @@ -530,12 +534,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] @@ -561,21 +566,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 @@ -590,7 +596,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); } @@ -1204,22 +1210,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!( @@ -1231,7 +1245,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); diff --git a/src-tauri/src/browser_permissions_tests.rs b/src-tauri/src/browser_permissions_tests.rs index 112cf50ef3..697b3571b3 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 = tauri::WebviewWindowBuilder::new(&app, "huddle", WebviewUrl::default()) .build() .unwrap(); + // The hidden page that runs agents. + let agents = tauri::WebviewWindowBuilder::new(&app, "agents", WebviewUrl::default()) + .build() + .unwrap(); let application_commands = [ "identity_restore", @@ -122,6 +126,8 @@ fn native_command_permissions_isolate_webviews() { "media_copy_image", "media_stream_base", "get_os_idle_seconds", + "agents_port_listen", + "agents_port_send", "plugin_import_folder", "plugin_update_check", "plugin_update_preview", @@ -215,6 +221,7 @@ fn native_command_permissions_isolate_webviews() { assert!(invoke(&main, "relay_agent_history_decode", local_origin).is_err()); for command in application_commands { assert!(invoke(&main, 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(), command == "media_download", @@ -233,6 +240,38 @@ fn native_command_permissions_isolate_webviews() { ); } assert!(invoke(&main, "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/host_mcp.rs b/src-tauri/src/host_mcp.rs index b75019f333..1f9a033ba4 100644 --- a/src-tauri/src/host_mcp.rs +++ b/src-tauri/src/host_mcp.rs @@ -24,7 +24,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}; @@ -105,6 +107,7 @@ impl Shared { struct Entry { plugin: String, + page: Page, shared: Arc, } @@ -115,9 +118,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 { @@ -130,20 +133,20 @@ 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, ) -> 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 { @@ -175,6 +178,7 @@ impl HostMcpServers { handle, Entry { plugin, + page, shared: shared.clone(), }, ); @@ -215,10 +219,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(); } @@ -478,14 +493,15 @@ 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, revision: String, on_event: Channel, ) -> 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 643c05012b..7c738b4252 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) + .serve(servers.page("main"), "plugin".into(), channel) .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) + .serve(servers.page("main"), "plugin".into(), channel) .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) + .serve(servers.page("main"), "plugin".into(), channel) .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) + .serve(servers.page("main"), "plugin".into(), channel) .unwrap(); initialize(&servers, &info, &mut events).await; // Whoever claimed the session, two clients hold the token. @@ -240,7 +240,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) + .serve(servers.page("main"), "plugin".into(), channel) .unwrap(); let claim = || claim(&servers, &info); // The page answered with an error. @@ -284,7 +284,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) + .serve(servers.page("main"), "plugin".into(), channel) .unwrap(); let named = initialize(&servers, &info, &mut events).await; let call = r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#; @@ -305,7 +305,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) + .serve(servers.page("main"), "plugin".into(), channel) .unwrap(); let request = post(&info, &info.token, "", INITIALIZE); let client = tokio::spawn({ @@ -340,7 +340,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) + .serve(servers.page("main"), "plugin".into(), channel) .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. @@ -363,9 +363,13 @@ 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) + .serve(page.clone(), "plugin".into(), events_channel) + .unwrap(); + let (other_channel, _other_events) = channel(); + let other = servers + .serve(servers.page("agents"), "plugin".into(), other_channel) .unwrap(); let request = post( &info, @@ -378,11 +382,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).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 3d62548031..6d12187911 100644 --- a/src-tauri/src/host_process.rs +++ b/src-tauri/src/host_process.rs @@ -1,10 +1,11 @@ //! 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 +//! 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: there is no sandbox. use crate::host_command::{effective_path, resolve_program}; +use crate::page::{Page, PageLoads}; use crate::{with_manager, PluginManager}; use buzzodz_plugins::HostProcess; use serde::{Deserialize, Serialize}; @@ -60,6 +61,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, @@ -74,9 +76,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 { @@ -97,15 +99,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, @@ -113,7 +115,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 { @@ -138,6 +140,7 @@ impl HostProcesses { handle, Entry { plugin, + page: page.clone(), group: child.id(), stdin: child .stdin @@ -205,7 +208,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 { @@ -251,22 +254,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() @@ -278,6 +280,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 { @@ -292,6 +305,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, @@ -303,7 +317,7 @@ pub(crate) async fn plugin_host_process_spawn( on_event: Channel, mcp: Option, ) -> Result { - let page = processes.page(); + let page = processes.page(webview.label()); let plugin = id.clone(); let declared = with_manager(manager, move |manager| { manager diff --git a/src-tauri/src/host_process/tests.rs b/src-tauri/src/host_process/tests.rs index 827e417624..46e9eb25f1 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 fd51fc4006..d52836522f 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1,3 +1,4 @@ +mod agents_port; mod archive; use archive::relay_archive; mod browser; @@ -54,6 +55,7 @@ mod microphone; mod notifications; mod os_idle; +mod page; mod repositories_folder; mod window_controls; mod window_state; @@ -546,6 +548,8 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen media_copy_image, media_stream_base, get_os_idle_seconds, + agents_port::agents_port_listen, + agents_port::agents_port_send, workspace_pick_folder, repositories_folder_validate, plugin_import_folder, @@ -792,6 +796,11 @@ pub fn run() { // `just desktop` builds step aside for a packaged app. !tauri::is_dev(), )); + // 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"))] @@ -833,6 +842,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) @@ -875,21 +885,23 @@ 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()); + } 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::().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 { @@ -916,6 +928,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 6394118ad4..b93ba4154d 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -922,6 +922,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); @@ -950,8 +952,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; @@ -976,13 +984,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); } } @@ -999,13 +1029,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 e9ca9171df..dff703133d 100644 --- a/src-tauri/src/relay/tests.rs +++ b/src-tauri/src/relay/tests.rs @@ -3365,7 +3365,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(); @@ -3389,7 +3389,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()); } } @@ -3398,7 +3398,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(); @@ -3449,7 +3449,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) @@ -3515,7 +3517,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..88296a11a5 --- /dev/null +++ b/src/agents.ts @@ -0,0 +1,12 @@ +// FOUNDATION: Startup of the page that runs agents. It has no screens. +import { createAgentsHost } from "./app/agents-host"; +import { loadAgentStorage } from "./features/agents/state"; + +// Agent storage is read before the service, since its reads are synchronous. +void loadAgentStorage().finally(() => { + const host = createAgentsHost(); + if (import.meta.hot) + import.meta.hot.dispose(() => { + void host.dispose().catch(console.error); + }); +}); diff --git a/src/app/agents-host.ts b/src/app/agents-host.ts new file mode 100644 index 0000000000..dd7c3b4c7b --- /dev/null +++ b/src/app/agents-host.ts @@ -0,0 +1,86 @@ +// FOUNDATION: Compose the page that runs agents apart from the screens: the +// plugins' agent entries and the agents service, served to 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 { createLocalAgents } from "../features/agents/local"; +import { createLocalAgentActivity } from "../features/agents/local-activity"; +import { type AgentsPort, serveAgents } from "../features/agents/port"; +import { AgentsService } from "../features/agents/service"; +import type { ToolOwner } from "../features/agents/tool-port"; +import { pageCommunities } from "../features/communities/page"; +import { DecisionsService } from "../features/decisions/service"; +import { HostService } from "../features/host/service"; +import { IdentityNamesService } from "../features/identity-names/service"; +import { connectNativeTransport } from "../features/relay/native"; +import { createPluginManager } from "../plugins/manager"; +import { withTimeout } from "../plugins/timeout"; + +export function createAgentsHost( + port: AgentsPort = eventPort(AGENTS_WEBVIEW, "main"), +) { + const ctx = new Context(); + new HostService(ctx); + const plugins = createPluginManager(ctx, { + bundled: bundledPlugins, + parts: "agent", + }); + const localAgents = createLocalAgents(); + const localActivity = createLocalAgentActivity(); + const decisions = new DecisionsService(ctx); + const identityNames = new IdentityNamesService(ctx, localAgents.agents); + const communities = pageCommunities(ctx, connectNativeTransport, { + identityNames, + agentChoices: localAgents.agents, + localAgents: localActivity.reader, + }); + const agents = new AgentsService( + ctx, + communities, + undefined, + undefined, + undefined, + localActivity, + undefined, + decisions, + ); + ctx.effect(() => + localAgents.bind({ + snapshot: agents.local, + subscribe: agents.subscribe, + refresh: () => agents.refresh(), + }), + ); + /** Runs `add` as `owner`'s, as the screens' page's plugin runtime names + * it, until the returned stop. */ + const as = (owner: ToolOwner, add: (plugin: Context) => void) => { + const fiber = ctx + .extend({ pluginOwner: Object.freeze({ ...owner }) }) + .plugin({ inject: ["agents"], apply: add }); + return () => void fiber.dispose(); + }; + const stop = serveAgents(agents, port, { + addTool: (tool, owner) => + as(owner, (plugin) => plugin.agents.addTool(tool)), + register: (type, owner) => + as(owner, (plugin) => plugin.agents.register(type)), + local: localAgents.agents, + activity: localActivity.reader, + follow: communities.follow, + }); + let disposal: Promise | undefined; + return { + agents, + dispose() { + stop(); + 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 9fc686611a..c24e1d48f9 100644 --- a/src/app/services.ts +++ b/src/app/services.ts @@ -23,7 +23,11 @@ import { import { AccountActionsService } from "../features/account-actions/service"; import { ShortcutsService } from "../features/shortcuts/service"; import { DecisionsService } from "../features/decisions/service"; -import { AgentsService } from "../features/agents/service"; +import { AgentsService, type AgentsClient } from "../features/agents/service"; +import { AGENTS_WEBVIEW, agentsRunApart } from "../features/agents/apart"; +import { eventPort } from "../features/agents/event-port"; +import { connectAgents } from "../features/agents/port"; +import { ScreensAgents } from "../features/agents/screens"; import { createShortcutBindings } from "../features/shortcuts/preferences"; import { ConversationService } from "../features/conversation/service"; import { createAppearance } from "../shared/theme/service"; @@ -46,12 +50,18 @@ export function createServices() { const updates = createUpdates(); const ctx = new Context(); new HostService(ctx); + // Agents run in their own page where the app opens one; this page then + // runs only the plugins' UI entries, and reaches agents over a port. + const apart = agentsRunApart() + ? connectAgents(eventPort("main", AGENTS_WEBVIEW)) + : undefined; const plugins = createPluginManager(ctx, { bundled: bundledPlugins, builtIn: conversationFeatures, + parts: apart ? "ui" : "both", }); const localAgents = createLocalAgents(); - const localActivity = createLocalAgentActivity(); + const localActivity = apart ? undefined : createLocalAgentActivity(); const navigationHost = provideNavigation(ctx); const navigation = navigationHost.navigation; const browser = new BrowserService(ctx); @@ -72,7 +82,7 @@ export function createServices() { localAgents.agents, identity?.ready, identity ? connectNativeTransport : undefined, - localActivity.reader, + apart?.activity ?? localActivity?.reader, ); ctx.provide("communityReader", { snapshot: communities.snapshot, @@ -88,23 +98,42 @@ export function createServices() { const adminAccess = createAdminAccess(relay, communities, nativeRelayStaff()); ctx.provide("adminAccess", adminAccess); ctx.effect(() => () => adminAccess.dispose()); - const agents = new AgentsService( - ctx, - communities, - undefined, - undefined, - undefined, - localActivity, - undefined, - decisions, - ); - ctx.effect(() => - localAgents.bind({ - snapshot: agents.local, - subscribe: agents.subscribe, - refresh: () => agents.refresh(), - }), - ); + let agents: AgentsClient; + if (apart) { + new ScreensAgents(ctx, apart, communities.open); + agents = apart; + ctx.effect(() => localAgents.bind(apart.local)); + const follow = () => { + const { viewer, memberships, selected } = communities.snapshot(); + apart.follow({ + ...(viewer !== undefined && { viewer }), + memberships: memberships.map((membership) => membership.id), + selected, + }); + }; + follow(); + ctx.effect(() => communities.subscribe(follow)); + ctx.effect(() => () => apart.close()); + } else { + const service = new AgentsService( + ctx, + communities, + undefined, + undefined, + undefined, + localActivity, + undefined, + decisions, + ); + agents = service; + ctx.effect(() => + localAgents.bind({ + snapshot: service.local, + subscribe: service.subscribe, + refresh: () => service.refresh(), + }), + ); + } const notifications = new NotificationsService( ctx, navigation, 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/event-port.ts b/src/features/agents/event-port.ts index ffbe3732dd..02b558abfd 100644 --- a/src/features/agents/event-port.ts +++ b/src/features/agents/event-port.ts @@ -1,17 +1,20 @@ -// A port between pages in two webviews of this app, over Tauri events. The -// agents protocol needs each message once and in the order sent: a call's -// answer must not overtake the state push before it. Separate emits can -// arrive out of order, so each frame is numbered and held until its turn. +// 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 agents protocol 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 { emitTo, listen, type UnlistenFn } from "@tauri-apps/api/event"; +import { Channel, invoke } from "@tauri-apps/api/core"; import type { AgentsPort } from "./port"; const EVENT = "buzz://agents-port"; +type UnlistenFn = () => void; export type Frame = Readonly< { @@ -34,6 +37,23 @@ export type Events = Readonly<{ ): 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 @@ -41,7 +61,7 @@ export type Events = Readonly<{ export function eventPort( from: string, to: string, - events: Events = { emitTo, listen }, + events: Events = agentsPortEvents, ): AgentsPort { const session = crypto.randomUUID(); let closed = false; diff --git a/src/features/agents/port.test.ts b/src/features/agents/port.test.ts index 19cc2bf202..0a573ca9ae 100644 --- a/src/features/agents/port.test.ts +++ b/src/features/agents/port.test.ts @@ -684,3 +684,31 @@ it("does not show a type whose handles cannot cross", async () => { expect(error).toHaveBeenCalled(); error.mockRestore(); }); + +it("tells the service the screens' page's communities, and tells a reloaded service again", async () => { + const events = jsonEvents(); + const agents = fakeAgents() as unknown as Agents; + const heard: unknown[] = []; + const follow = (communities: unknown) => heard.push(communities); + const client = connectAgents(eventPort("main", "agents", events)); + const communities = { + viewer: "v", + memberships: ["wss://alpha.example"], + selected: "wss://alpha.example", + }; + client.follow(communities); + const first = serveAgents(agents, eventPort("agents", "main", events), { + follow, + }); + await vi.waitFor(() => expect(heard).toEqual([communities])); + // The agents page reloads: the new one hears them without a change. + first(); + const second = serveAgents(agents, eventPort("agents", "main", events), { + follow, + }); + stops.push(() => { + client.close(); + second(); + }); + await vi.waitFor(() => expect(heard).toEqual([communities, communities])); +}); diff --git a/src/features/agents/port.ts b/src/features/agents/port.ts index 6920b96c27..52bc3fa55c 100644 --- a/src/features/agents/port.ts +++ b/src/features/agents/port.ts @@ -15,6 +15,7 @@ import { type SessionState, } from "./service"; import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { FollowedCommunities } from "../communities/page"; import type { LocalAgents, LocalAgentsSnapshot } from "./local"; import type { LocalAgentSource, LocalWork } from "./local-activity"; import type { Reply } from "../../buzz-mcp/tools"; @@ -136,7 +137,8 @@ type ToService = }> | Readonly<{ kind: "refresh-local"; id: number }> | Readonly<{ kind: "dismiss-work"; work: WorkKey }> - | Readonly<{ kind: "interrupt-work"; id: number; work: WorkKey }>; + | Readonly<{ kind: "interrupt-work"; id: number; work: WorkKey }> + | Readonly<{ kind: "communities"; communities: FollowedCommunities }>; /** A `MessagePort`, or a link to a page in another webview. It delivers each * message once, in the order sent. */ @@ -161,6 +163,9 @@ export type AgentsPortClient = AgentsClient & local: LocalAgents; /** The service's local agents' work, for status rows and Stop. */ activity: LocalAgentSource; + /** Tells the service the communities this page has now: the viewer, + * the joined ones and the selected one. */ + follow(communities: FollowedCommunities): void; }>; const placeKey = (key: string, { channelId, root }: SessionPlace) => @@ -222,7 +227,9 @@ function stateOf( * tool or hosted type the screens' page lists is added, as its plugin's, * with `options.addTool` or `options.register` until it withdraws it; * without them, it is not added. `local` and `activity` are the service's - * local agents and their work, for the screens' page's names and status. */ + * local agents and their work, for the screens' page's names and status. + * `follow` hears the communities the screens' page has, each time they + * change. */ export function serveAgents( agents: Agents, port: AgentsPort, @@ -231,6 +238,7 @@ export function serveAgents( register?: (type: AgentType, owner: ToolOwner) => () => void; local?: LocalAgents; activity?: LocalAgentSource; + follow?: (communities: FollowedCommunities) => void; }> = {}, ) { const { local, activity } = options; @@ -337,6 +345,10 @@ export function serveAgents( answer(data.id, () => local?.refresh()); return; } + if (data.kind === "communities") { + options.follow?.(data.communities); + return; + } if (data.kind === "dismiss-work") { activity?.dismissError?.(data.work); return; @@ -477,6 +489,7 @@ export function connectAgents(port: AgentsPort) { let closed = false; const listeners = new Set<() => void>(); const asked = new Set(); + let followed: FollowedCommunities | undefined; const waiting = new Map< number, Readonly<{ resolve(value: unknown): void; reject(error: Error): void }> @@ -490,6 +503,7 @@ export function connectAgents(port: AgentsPort) { asked.clear(); tools.relist(); hosted.relist(); + if (followed) post({ kind: "communities", communities: followed }); } if (state !== prior) for (const listener of listeners) listener(); return; @@ -612,6 +626,12 @@ export function connectAgents(port: AgentsPort) { present: types.present, addTool: tools.add, register: hosted.register, + follow: (communities: FollowedCommunities) => { + // The screens' page's communities change for much the service ignores. + if (JSON.stringify(communities) === JSON.stringify(followed)) return; + followed = communities; + if (!closed) post({ kind: "communities", communities }); + }, local: Object.freeze({ snapshot: () => state.local, subscribe, diff --git a/src/features/agents/screens.test.ts b/src/features/agents/screens.test.ts new file mode 100644 index 0000000000..1f936cbed6 --- /dev/null +++ b/src/features/agents/screens.test.ts @@ -0,0 +1,89 @@ +import { Context } from "@deepseek-ai/cordis"; +import { afterEach, expect, it, vi } from "vitest"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import { communityDestination } from "../communities/destination"; +import type { RelayData } from "../relay/service"; +import { connectAgents, serveAgents } from "./port"; +import { ScreensAgents } from "./screens"; +import type { Agents, AgentsSnapshot } from "./service"; +import type { ToolOwner } from "./tool-port"; + +const stops: (() => unknown)[] = []; +afterEach(async () => { + for (const stop of stops.splice(0).reverse()) await stop(); +}); +const key = "a".repeat(64); +const snapshot: AgentsSnapshot = { + status: "ready", + classifier: "available", + classifierKey: "saved", + agents: [], + running: [{ key, relay: "wss://alpha.example" }] as never, + interrupted: [], + teams: [], +}; +function setup() { + const agents = { + snapshot: () => snapshot, + subscribe: () => () => {}, + types: () => [], + memberships: () => [], + joinable: () => [], + interruptible: () => false, + creationRequests: () => [], + updateRequests: () => [], + } as unknown as Agents; + const added = new Map(); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(agents, port1, { + addTool: (tool, owner) => { + added.set(tool.name, owner); + return () => added.delete(tool.name); + }, + }); + const client = connectAgents(port2); + const ctx = new Context(); + const community = {} as RelayData; + const open = vi.fn(() => community); + new ScreensAgents(ctx, client, open); + stops.push( + stop, + () => client.close(), + () => ctx.fiber.dispose(), + ); + return { ctx, client, added, open, community }; +} +const tool: AddedTool = { + name: "files_copy", + description: "Copy a file.", + inputSchema: { type: "object" }, + call: async () => "done", +}; + +it("lists a UI entry's tool under its plugin until the plugin stops", async () => { + const { ctx, added } = setup(); + const owner = Object.freeze({ id: "example", revision: "1" }); + const plugin = ctx.extend({ pluginOwner: owner }).plugin({ + inject: ["agents"], + apply: (scope: Context) => scope.agents.addTool(tool), + }); + await vi.waitFor(() => expect(added.get("files_copy")).toEqual(owner)); + await plugin.dispose(); + await vi.waitFor(() => expect(added.size).toBe(0)); +}); + +it("connects a running agent's community from this page, and refuses what only running agents use", async () => { + const { ctx, client, open, community } = setup(); + await vi.waitFor(() => expect(client.snapshot().running).toHaveLength(1)); + expect(ctx.agents.relay(key)).toBe(community); + expect(open).toHaveBeenCalledWith( + communityDestination("wss://alpha.example").url, + ); + expect(ctx.agents.relay("b".repeat(64))).toBeUndefined(); + expect(() => + ctx.agents.wake(key, { channelId: "c", text: "t" } as never), + ).toThrow(/agent entry/); + expect(() => ctx.agents.storage).toThrow(/agent entry/); + // Outside a plugin, nothing names whose tool it would be. + expect(() => ctx.agents.addTool(tool)).toThrow(/Only a plugin/); +}); diff --git a/src/features/agents/screens.ts b/src/features/agents/screens.ts new file mode 100644 index 0000000000..abdeffdfa6 --- /dev/null +++ b/src/features/agents/screens.ts @@ -0,0 +1,154 @@ +// The agents service as the screens' page has it when agents run in another +// page. What screens read and call goes over the agents port. What a +// plugin's UI entry adds, its tools, hosted types and what its types show, +// is listed with the service under the plugin's identity until the plugin +// stops. What only running agents use belongs in the plugin's agent entry, +// which runs where the agents do. +import { type Context, Service } from "@deepseek-ai/cordis"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { RelayData } from "../relay/service"; +import { communityDestination } from "../communities/destination"; +import type { connectAgents } from "./port"; +import type { Remote } from "./handles"; +import type { AgentKey } from "./membership"; +import type { + AgentPresentation, + Agents, + AgentsClient, + AgentType, + RegisteredAgentType, +} from "./service"; + +type Client = ReturnType; + +const apart = (what: string) => + new Error( + `${what} is for running agents, which run in another page: use it from the plugin's agent entry.`, + ); + +export class ScreensAgents extends Service implements Agents { + // What screens read and call, as the client has it. + declare readonly snapshot: AgentsClient["snapshot"]; + declare readonly subscribe: AgentsClient["subscribe"]; + declare readonly resumeInterrupted: AgentsClient["resumeInterrupted"]; + declare readonly dismissInterrupted: AgentsClient["dismissInterrupted"]; + declare readonly find: AgentsClient["find"]; + declare readonly create: AgentsClient["create"]; + declare readonly creationRequests: AgentsClient["creationRequests"]; + declare readonly dismissCreation: AgentsClient["dismissCreation"]; + declare readonly approveCreation: AgentsClient["approveCreation"]; + declare readonly updateRequests: AgentsClient["updateRequests"]; + declare readonly dismissUpdate: AgentsClient["dismissUpdate"]; + declare readonly approveUpdate: AgentsClient["approveUpdate"]; + declare readonly finish: AgentsClient["finish"]; + declare readonly portable: AgentsClient["portable"]; + declare readonly port: AgentsClient["port"]; + declare readonly portTeam: AgentsClient["portTeam"]; + declare readonly save: AgentsClient["save"]; + declare readonly previewRetype: AgentsClient["previewRetype"]; + declare readonly retype: AgentsClient["retype"]; + declare readonly remove: AgentsClient["remove"]; + declare readonly memberships: AgentsClient["memberships"]; + declare readonly forget: AgentsClient["forget"]; + declare readonly joinable: AgentsClient["joinable"]; + declare readonly join: AgentsClient["join"]; + declare readonly leave: AgentsClient["leave"]; + declare readonly interruptible: AgentsClient["interruptible"]; + declare readonly sessionState: AgentsClient["sessionState"]; + declare readonly session: AgentsClient["session"]; + declare readonly interrupt: AgentsClient["interrupt"]; + declare readonly unarchive: AgentsClient["unarchive"]; + declare readonly classifier: AgentsClient["classifier"]; + declare readonly setClassifierKey: AgentsClient["setClassifierKey"]; + declare readonly clearClassifierKey: AgentsClient["clearClassifierKey"]; + declare readonly saveTeam: AgentsClient["saveTeam"]; + declare readonly removeTeam: AgentsClient["removeTeam"]; + private readonly client: Client; + private readonly open: (community: string) => RelayData | undefined; + constructor( + ctx: Context, + client: Client, + open: (community: string) => RelayData | undefined, + ) { + super(ctx, "agents"); + const { + present: _present, + addTool: _addTool, + register: _register, + local: _local, + activity: _activity, + follow: _follow, + close: _close, + types: _types, + ...screens + } = client; + Object.assign(this, screens); + this.client = client; + this.open = open; + } + private owner(what: string) { + const owner = this.ctx.pluginOwner; + if (!owner) throw new Error(`Only a plugin ${what}`); + return owner; + } + types = () => + this.client.types() as unknown as readonly RegisteredAgentType[]; + register(type: AgentType) { + const withdraw = this.client.register( + type as unknown as AgentType, + this.owner("registers agent types"), + ); + this.ctx.effect(() => withdraw); + } + present( + typeId: string, + present: (exposed: Remote) => AgentPresentation, + ) { + const { id } = this.owner("presents agent types"); + const stop = this.client.present( + `${id}/${typeId}`, + present as unknown as Parameters[1], + ); + this.ctx.effect(() => stop); + } + addTool(tool: AddedTool) { + const withdraw = this.client.addTool(tool, this.owner("adds agent tools")); + this.ctx.effect(() => withdraw); + } + /** This page's own connection to the running agent's community. */ + relay(key: AgentKey) { + const agent = this.client + .snapshot() + .running.find((running) => running.key === key); + if (!agent) return undefined; + try { + return this.open(communityDestination(agent.relay).url); + } catch { + return undefined; + } + } + addAttentionSource(): never { + throw apart("addAttentionSource"); + } + attentionProviders(): never { + throw apart("attentionProviders"); + } + tools(): never { + throw apart("tools"); + } + wake(): never { + throw apart("wake"); + } + handle(): never { + throw apart("handle"); + } + core(): never { + throw apart("core"); + } + note(): never { + throw apart("note"); + } + get storage(): never { + throw apart("storage"); + } +} diff --git a/src/main.tsx b/src/main.tsx index c190cc5971..fee0d0019c 100644 --- a/src/main.tsx +++ b/src/main.tsx @@ -5,6 +5,7 @@ import { StrictMode } from "react"; import { createRoot } from "react-dom/client"; import { createServices } from "./app/services"; import { loadAgentStorage } from "./features/agents/state"; +import { agentsRunApart } from "./features/agents/apart"; import { App } from "./app/App"; import "@fontsource-variable/inter/wght.css"; import "@fontsource-variable/inter/wght-italic.css"; @@ -20,8 +21,9 @@ if (!container) { } // Agent storage is read before any service, since its reads are synchronous. -// It settles, with localStorage at worst, so the window always starts. -void loadAgentStorage().finally(() => { +// It settles, with localStorage at worst, so the window always starts. Where +// agents run in their own page, that page reads it, and this one never does. +void (agentsRunApart() ? Promise.resolve() : loadAgentStorage()).finally(() => { const services = createServices(); const root = createRoot(container); function HostApp() { diff --git a/src/plugins/manager.ts b/src/plugins/manager.ts index 3aabda94f5..2452710220 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); @@ -78,6 +81,8 @@ export function createPluginManager( ), storage.readModule, ), + undefined, + parts, ); let configuration: ConfigurationState = { status: "loading" }; let busy = false; diff --git a/src/plugins/runtime.test.mjs b/src/plugins/runtime.test.mjs index 1a86b7e432..60b24bf6fc 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,7 @@ function harness(load, timeoutMs) { root, async (plugin) => ({ inject: ["pages"], ...(await load(plugin)) }), timeoutMs, + parts, ); runtime.subscribe(() => { states = runtime.snapshot(); @@ -135,6 +136,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 304bb5a301..970f01dcf3 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,14 +40,17 @@ 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, load: (plugin: PluginInfo) => Promise, timeoutMs = 10_000, + parts: PluginParts = "both", ) { this.root = root; this.load = load; + this.parts = parts; root.provide("pluginStatus", { isActive: this.isActive, subscribe: this.subscribe, @@ -153,10 +161,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 a0b36b7e01..792b14afe8 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -97,6 +97,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)), }, }, },