From 770aeefc6a62ea8ce8c0679cb4dcc1552942d5a5 Mon Sep 17 00:00:00 2001 From: Dusty Date: Mon, 26 Oct 2026 17:02:29 -0700 Subject: [PATCH 1/8] desktop: Compose owns the server and worker; give containers their own addresses SERVICES now includes server and worker, and Stop names the openbot profile they sit behind. The settings point every Bot at server:3001, give the server host.docker.internal addresses for local Intelligence (the browser keeps the localhost ones), and give an installed harness its container address. The macOS Podman overlay publishes the server's two ports on IPv4 only. --- desktop/src-tauri/src/env.rs | 174 +++++++++++++++++-- desktop/src-tauri/src/macos-podman-ports.yml | 4 + desktop/src-tauri/src/main.rs | 2 +- desktop/src-tauri/src/stack.rs | 58 ++++++- 4 files changed, 221 insertions(+), 17 deletions(-) diff --git a/desktop/src-tauri/src/env.rs b/desktop/src-tauri/src/env.rs index d6848b56a1..c475c862f0 100644 --- a/desktop/src-tauri/src/env.rs +++ b/desktop/src-tauri/src/env.rs @@ -226,6 +226,37 @@ fn bind_loopbacks(port: u16) -> std::io::Result> { Ok(held) } +/** +A loopback address as a container on this machine reaches it, or empty if it is not loopback. + +`localhost`, `*.localhost`, `127.x.y.z` and `[::1]` become `host.docker.internal`, keeping the +scheme, port and path. Anything else is already reachable from a container as written, and empty is +how `docker-compose.yml` is told to use it unchanged. +*/ +pub fn container_address(address: &str) -> String { + let address = address.trim(); + let Ok(mut url) = reqwest::Url::parse(address) else { + return String::new(); + }; + let host = url.host_str().unwrap_or_default(); + let loopback = host == "localhost" + || host.ends_with(".localhost") + || host + .trim_start_matches('[') + .trim_end_matches(']') + .parse::() + .is_ok_and(|ip| ip.is_loopback()); + if !loopback || url.set_host(Some("host.docker.internal")).is_err() { + return String::new(); + } + // The parser writes an empty path as `/`; given without one, it is returned without one. + let moved = url.to_string(); + match moved.strip_suffix('/') { + Some(bare) if url.path() == "/" && !address.ends_with('/') => bare.to_string(), + _ => moved, + } +} + #[cfg(test)] mod port_tests { use super::*; @@ -596,6 +627,23 @@ pub fn compose( "INTELLIGENCE_GATEWAY_WS_URL".into(), intelligence.gateway_ws_url.trim().to_string(), ); + /* + * The same two addresses as the server's container reaches them. + * + * Local Intelligence answers on this machine's loopback, which inside a container is the + * container. `host.docker.internal` is the machine. Only the server is moved: the browser opens + * the gateway too, from the address in `/info`, and the browser is on the host, so + * `docker-compose.yml` hands it the address above unchanged. Empty for an Intelligence somewhere + * else, which the container reaches at the address it was given. + */ + env.insert( + "INTELLIGENCE_CONTAINER_API_URL".into(), + container_address(&intelligence.api_url), + ); + env.insert( + "INTELLIGENCE_CONTAINER_GATEWAY_WS_URL".into(), + container_address(&intelligence.gateway_ws_url), + ); // Explicit desktop selection must override a CLI key left in the checkout's .env. // Supply both names for consumers that still read the legacy variable directly. for key in ["CPK_INTELLIGENCE_API_KEY", "INTELLIGENCE_API_KEY"] { @@ -685,6 +733,9 @@ pub fn compose( } .into(), ); + // Empty unless a harness runs here, so the address a remote Bot was given is the one used and a + // harness picked on an earlier start does not leave its container address behind. + env.insert("PICKED_HARNESS_CONTAINER_URL".into(), String::new()); if let Some(picked) = harness { match picked { PickedHarness::Installed { @@ -700,15 +751,20 @@ pub fn compose( env.insert("PICKED_HARNESS_PORT".into(), port.to_string()); env.insert("PICKED_HARNESS_NAME".into(), name.clone()); let run_path = run_path.trim(); + let run_path = if run_path.is_empty() || run_path.starts_with('/') { + run_path.to_string() + } else { + format!("/{run_path}") + }; env.insert( "PICKED_HARNESS_URL".into(), - if run_path.is_empty() { - format!("http://127.0.0.1:{host_port}") - } else if run_path.starts_with('/') { - format!("http://127.0.0.1:{host_port}{run_path}") - } else { - format!("http://127.0.0.1:{host_port}/{run_path}") - }, + format!("http://127.0.0.1:{host_port}{run_path}"), + ); + // The server is a container on the harness's network, so it dials the service by + // name on the image's own port, not the host port published for everything else. + env.insert( + "PICKED_HARNESS_CONTAINER_URL".into(), + format!("http://agent-harness:{port}{run_path}"), ); /* * The kind, as the package spells it. @@ -772,12 +828,11 @@ pub fn compose( ); env.extend(ports.settings()); + // The server is a container on the Bots' network now, so they reach it by name and on its own + // port, whichever host port this start published it on. env.insert( "OPENBOT_TOOL_URL".into(), - format!( - "http://host.docker.internal:{}/api/agent-tools/call", - ports.server - ), + "http://server:3001/api/agent-tools/call".into(), ); // The whole deployment is on this machine, so the server must be allowed to talk to it. @@ -2135,8 +2190,8 @@ HTTPS_PROXY=http://proxy:8080 ); assert_eq!( value_of(&written, "OPENBOT_TOOL_URL"), - "http://host.docker.internal:3999/api/agent-tools/call", - "container tools must call the selected API port" + "http://server:3001/api/agent-tools/call", + "Bots call the server container by name, on its own port, whatever the host port" ); } @@ -2168,6 +2223,99 @@ HTTPS_PROXY=http://proxy:8080 settings["PICKED_HARNESS_URL"], "http://127.0.0.1:52106/ag-ui" ); + // The server container dials the service by name, on the image's port. + assert_eq!( + settings["PICKED_HARNESS_CONTAINER_URL"], + "http://agent-harness:4206/ag-ui" + ); + } + + #[test] + fn a_remote_harness_and_no_harness_leave_no_container_address() { + let remote = compose( + &intelligence(), + &Model::default(), + &engine_status(None), + &Ports::default(), + &pinned(), + Some(&PickedHarness::RemoteAgUi { + url: "https://agents.example/run".into(), + name: "Remote".into(), + remote_agent_id: String::new(), + }), + &BTreeMap::new(), + ); + assert_eq!(remote["PICKED_HARNESS_URL"], "https://agents.example/run"); + assert_eq!(remote["PICKED_HARNESS_CONTAINER_URL"], ""); + + let none = compose( + &intelligence(), + &Model::default(), + &engine_status(None), + &Ports::default(), + &pinned(), + None, + &BTreeMap::new(), + ); + assert_eq!(none["PICKED_HARNESS_CONTAINER_URL"], ""); + } + + /// Local Intelligence answers on loopback; the server's container reaches it through the host. + #[test] + fn local_intelligence_gets_a_container_address_and_hosted_does_not() { + for (given, expected) in [ + ( + "http://localhost:18082", + "http://host.docker.internal:18082", + ), + ("ws://localhost:14401", "ws://host.docker.internal:14401"), + ("ws://localhost:14401/", "ws://host.docker.internal:14401/"), + ( + "http://127.0.0.1:8082/api", + "http://host.docker.internal:8082/api", + ), + ("http://[::1]:8082", "http://host.docker.internal:8082"), + ( + "http://cpk.localhost:8080", + "http://host.docker.internal:8080", + ), + ( + " http://localhost:18082 ", + "http://host.docker.internal:18082", + ), + ("https://api.intelligence.copilotkit.ai", ""), + ("wss://realtime.intelligence.copilotkit.ai", ""), + ("http://localhost.evil.test:8082", ""), + ("not a url", ""), + ("", ""), + ] { + assert_eq!(container_address(given), expected, "{given}"); + } + + let env = compose( + &Intelligence { + api_url: "http://localhost:18082".into(), + gateway_ws_url: "ws://localhost:14401".into(), + api_key: "synthetic".into(), + }, + &Model::default(), + &engine_status(None), + &Ports::default(), + &pinned(), + None, + &BTreeMap::new(), + ); + // The host's own addresses stay as given: the browser reads the gateway's from these. + assert_eq!(env["INTELLIGENCE_API_URL"], "http://localhost:18082"); + assert_eq!(env["INTELLIGENCE_GATEWAY_WS_URL"], "ws://localhost:14401"); + assert_eq!( + env["INTELLIGENCE_CONTAINER_API_URL"], + "http://host.docker.internal:18082" + ); + assert_eq!( + env["INTELLIGENCE_CONTAINER_GATEWAY_WS_URL"], + "ws://host.docker.internal:14401" + ); } } diff --git a/desktop/src-tauri/src/macos-podman-ports.yml b/desktop/src-tauri/src/macos-podman-ports.yml index 20b0ae6804..f3f8a86f52 100644 --- a/desktop/src-tauri/src/macos-podman-ports.yml +++ b/desktop/src-tauri/src/macos-podman-ports.yml @@ -20,3 +20,7 @@ services: agent-harness: ports: !override - "127.0.0.1:${PICKED_HARNESS_PORT:-4202}:${PICKED_HARNESS_PORT:-4202}" + server: + ports: !override + - "127.0.0.1:${APP_PORT:-3010}:3001" + - "127.0.0.1:${SERVER_PORT:-3001}:3001" diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs index ed0abdb898..3e344a06d9 100644 --- a/desktop/src-tauri/src/main.rs +++ b/desktop/src-tauri/src/main.rs @@ -11565,7 +11565,7 @@ fn main() { ["compose","-f","docker-compose.yml","config","--format","json"] => println!("{{\"services\":{{\"supervisor\":{{\"environment\":{{\"COMPUTER_NAMESPACE\":\"fixture\"}}}}}}}}"), ["compose","-f","docker-compose.yml","stop","supervisor"] => (), ["ps","--quiet","--filter",_,"--filter",_] => (), - ["compose","-f","docker-compose.yml","--profile","harness","down"] => { + ["compose","-f","docker-compose.yml","--profile","harness","--profile","openbot","down"] => { if cwd.join("fail-down").exists() { eprintln!("synthetic down refusal");std::process::exit(73); } if fs::read_to_string(cwd.join("fixture-containers-running")).ok().as_deref()==Some(&identity) { fs::remove_file(cwd.join("fixture-containers-running")).unwrap(); diff --git a/desktop/src-tauri/src/stack.rs b/desktop/src-tauri/src/stack.rs index 038a0a3383..6758b4b997 100644 --- a/desktop/src-tauri/src/stack.rs +++ b/desktop/src-tauri/src/stack.rs @@ -26,7 +26,24 @@ use crate::problem::Problem; /// The services Compose owns. `migrate` is deliberately absent: it is run once, to completion, /// rather than raised, and treating it as a long-lived service makes it look like a crash loop. -const SERVICES: [&str; 3] = ["postgres", "supervisor", "agent-computer"]; +/// +/// `server` and `worker` are OpenBot itself, from the published server image: the API, the app it +/// serves, and the routines sweep. They used to be Bun processes this app started on the host after +/// downloading Bun and running `bun install`, which on a fresh Windows machine took longer than +/// everything else in the install put together. Now they are containers like the rest, and the +/// restart policy that keeps them up is Docker's. +const SERVICES: [&str; 5] = [ + "postgres", + "supervisor", + "agent-computer", + "server", + "worker", +]; + +/// The Compose profile `server` and `worker` sit behind, so `scripts/start.sh`, whose server is a +/// host process, never starts a second one. Named on every command rather than only on `up`, so +/// `down` stops them and `config` describes them. +const OPENBOT_PROFILE: &str = "openbot"; /** The Bots that ship with OpenBot, which only run on an API key. @@ -1273,9 +1290,20 @@ pub fn down(engine: &Address, root: &Path) -> Result<(), String> { * * Named unconditionally rather than only when a harness was picked: this has to stop what an * earlier run started, and whether that run picked one is not something a Stop can know. + * + * `openbot` for the same reason: `server` and `worker` sit behind it, and a Stop that left them + * running would leave the app answering a window that says OpenBot is stopped. */ let output = compose_command(engine, root, &Secrets::new()) - .args(["-f", "docker-compose.yml", "--profile", "harness", "down"]) + .args([ + "-f", + "docker-compose.yml", + "--profile", + "harness", + "--profile", + OPENBOT_PROFILE, + "down", + ]) .output() .map_err(|error| format!("could not stop the stack: {error}"))?; @@ -6190,7 +6218,8 @@ fn main() { } } } - Some("compose") if actual == ["compose", "--profile", "harness", "down"] => { + Some("compose") + if actual == ["compose", "--profile", "harness", "--profile", "openbot", "down"] => { if race { std::fs::write(root.join("supervisor.stopped"), "").unwrap(); std::fs::write(root.join("stack.down"), "").unwrap(); @@ -8077,6 +8106,29 @@ fn main() { assert!(picked.contains(&"agent-harness")); } + /// Stop has to name both profiles, or what sits behind them keeps running: the one Bot the + /// person picked, and OpenBot's own server and worker. + #[test] + fn stopping_names_the_harness_and_openbot_profiles() { + let source = include_str!("stack.rs"); + let down = &source[source.find("pub fn down(").unwrap()..]; + let down = &down[..down.find("\n}\n").unwrap()]; + assert_eq!( + down.matches(r#""--profile","#).count(), + 2, + "compose down without a profile leaves that profile's services running" + ); + assert!( + down.contains(r#""harness","#), + "agent-harness keeps running" + ); + assert!( + down.contains("OPENBOT_PROFILE,"), + "server and worker keep running" + ); + assert_eq!(OPENBOT_PROFILE, "openbot"); + } + #[test] fn the_published_ports_are_read_off_a_real_listing() { // Verbatim from `compose ps --format '{{.Ports}}'` against a running deployment. From d39b004f95a40089910ec8168efd239f3f975661 Mon Sep 17 00:00:00 2001 From: Dusty Date: Mon, 26 Oct 2026 17:15:34 -0700 Subject: [PATCH 2/8] desktop: wait for the server container instead of starting Bun on the host Start now hands everything to Compose and waits for the `server` container to answer `/health` on both the app and API host ports, for up to 180 s. Docker's restart policy supervises the server and worker, so the host launch, readiness, pid recording and restart loop are gone. - The desktop host token is minted before `up` and reaches Compose, which names it for the server alone; it is still never persisted. - COMPUTER_NETWORK is `_default` (new `stack::project_name`), so Bot computers join the server's network. On Linux OPENBOT_SERVER_USER is the deployment folder's uid:gid. - `.openbot/` is created before `up`, since its bind mount does not create it. - App and server ports count as ours when our container publishes them; the pre-spawn port check is gone. - Ownership for showing, restoring and asking OpenBot is now "this root's server container publishes the ports" (`stack::server_publishes`), not recorded host pids. - Status lists server and worker as containers; Restart runs `compose restart server worker`. - What is left of the supervisor polls the server for refused connections (`watch_connections`). - Start no longer requires the host Bun or packages. - Old Bun host processes from earlier installs are still stopped on Start and Stop, which is how an upgrade frees 3001/3010. Also brings tests left stale by the previous commit up to date (server/worker in the up list, `--profile openbot` on down). --- desktop/src-tauri/src/desktop_page.rs | 4 +- desktop/src-tauri/src/desktop_status.rs | 24 +- desktop/src-tauri/src/env.rs | 32 +- desktop/src-tauri/src/main.rs | 2259 ++++---------------- desktop/src-tauri/src/preparation.rs | 26 +- desktop/src-tauri/src/stack.rs | 336 +-- desktop/src-tauri/src/stop_ipc_tests.rs | 20 +- desktop/src-tauri/tests/fixtures/engine.rs | 11 +- desktop/src/StatusWindow.tsx | 6 +- 9 files changed, 728 insertions(+), 1990 deletions(-) diff --git a/desktop/src-tauri/src/desktop_page.rs b/desktop/src-tauri/src/desktop_page.rs index 8f4357766c..7b544ef459 100644 --- a/desktop/src-tauri/src/desktop_page.rs +++ b/desktop/src-tauri/src/desktop_page.rs @@ -49,8 +49,8 @@ async fn status( } #[tauri::command] -fn restart(app: tauri::AppHandle) -> Result<(), String> { - crate::restart_openbot(app) +async fn restart(app: tauri::AppHandle) -> Result<(), String> { + crate::restart_openbot(app).await } /// Stops everything, as the menu's Stop did. The page asking goes with it; the window shows setup. diff --git a/desktop/src-tauri/src/desktop_status.rs b/desktop/src-tauri/src/desktop_status.rs index 03c547d24d..b6d3b57d97 100644 --- a/desktop/src-tauri/src/desktop_status.rs +++ b/desktop/src-tauri/src/desktop_status.rs @@ -2,8 +2,8 @@ //! //! Containers are told apart by the names their owners give them: Compose names OpenBot's //! `openbot--` (its project is `openbot`), and k3d names an Intelligence stack's -//! `k3d--`. Memory is Docker's own figure for each container, and the resident size -//! of OpenBot's host processes, so the total is what the machine is actually spending. +//! `k3d--`. Memory is Docker's own figure for each container, OpenBot's own server +//! and worker among them, so the total is what the machine is actually spending. use serde::Serialize; @@ -62,18 +62,6 @@ pub fn containers(stats: &str, stack: Option<&str>) -> (Vec, Vec) { (openbot, intelligence) } -/// `ps -o pid=,rss= -p …`: each process's resident size, in bytes (ps reports kibibytes). -pub fn resident(ps: &str) -> Vec<(u32, u64)> { - ps.lines() - .filter_map(|line| { - let mut fields = line.split_whitespace(); - let pid = fields.next()?.parse().ok()?; - let kib: u64 = fields.next()?.parse().ok()?; - Some((pid, kib * 1024)) - }) - .collect() -} - #[cfg(test)] mod tests { use super::*; @@ -133,12 +121,4 @@ mod tests { ); assert_eq!(intelligence[0].memory_bytes, None); } - - #[test] - fn resident_sizes_are_kibibytes_per_pid() { - assert_eq!( - resident(" 123 2048\n 456 10\nnonsense\n"), - vec![(123, 2_097_152), (456, 10_240)] - ); - } } diff --git a/desktop/src-tauri/src/env.rs b/desktop/src-tauri/src/env.rs index c475c862f0..be12c59632 100644 --- a/desktop/src-tauri/src/env.rs +++ b/desktop/src-tauri/src/env.rs @@ -137,6 +137,8 @@ impl Ports { /// Hold the probes until every port is chosen so allocations cannot collide with each other. /// Existing containers from this exact Compose project are reusable, never foreign listeners. /// `browser` asks for a port for the person's browser, kept from the last run where it is free. + /// The app and the API are among them: both are host ports of the one `server` container, so + /// a running OpenBot keeps its own ports across a Start rather than being moved off them. pub fn available( self, ours: &std::collections::HashSet, @@ -175,8 +177,8 @@ impl Ports { })) }; Ok(Self { - app: choose(self.app, false)?, - server: choose(self.server, false)?, + app: choose(self.app, true)?, + server: choose(self.server, true)?, postgres: choose(self.postgres, true)?, computer: choose(self.computer, true)?, bot: choose(self.bot, true)?, @@ -350,19 +352,17 @@ mod port_tests { } #[test] - fn own_published_container_ports_are_reused_but_do_not_authorize_host_port_reuse() { + fn own_published_container_ports_are_reused_but_a_foreign_listener_is_not() { let owned = TcpListener::bind("127.0.0.1:0").unwrap(); let port = owned.local_addr().unwrap().port(); let preferred = Ports { postgres: port, - server: port, ..Ports::default() }; let chosen = preferred .available(&HashSet::from([port]), None, false) .unwrap(); assert_eq!(chosen.postgres, port); - assert_ne!(chosen.server, port); assert_ne!( preferred .available(&HashSet::new(), None, false) @@ -372,6 +372,28 @@ mod port_tests { ); } + #[test] + fn the_app_and_api_ports_our_server_container_publishes_are_kept() { + // The one `server` container publishes both, so both are held while it runs. + let app = TcpListener::bind("127.0.0.1:0").unwrap(); + let server = TcpListener::bind("127.0.0.1:0").unwrap(); + let preferred = Ports { + app: app.local_addr().unwrap().port(), + server: server.local_addr().unwrap().port(), + ..Ports::default() + }; + let ours = HashSet::from([preferred.app, preferred.server]); + let chosen = preferred.available(&ours, None, false).unwrap(); + assert_eq!( + (chosen.app, chosen.server), + (preferred.app, preferred.server) + ); + // Held by somebody else, they are moved, as they always were. + let moved = preferred.available(&HashSet::new(), None, false).unwrap(); + assert_ne!(moved.app, preferred.app); + assert_ne!(moved.server, preferred.server); + } + #[test] fn invalid_saved_ports_fail_instead_of_probing_an_unrelated_default() { let root = crate::test_support::temp_root("invalid-local-ports"); diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs index 3e344a06d9..c2adf08499 100644 --- a/desktop/src-tauri/src/main.rs +++ b/desktop/src-tauri/src/main.rs @@ -19,7 +19,7 @@ use openbot_desktop_lib::{ install, local_browser, local_eval, local_eval_state, preparation, problem::{LeftoverReset, Problem}, provider, pull_metrics, quiet, recorder_external, recorder_host, recorder_install, stack, - supervise, telemetry, tray, windows as win, + telemetry, tray, windows as win, }; const QUIT_CLEANUP_NOTICE_FILE: &str = ".openbot-quit-cleanup-notice"; @@ -37,22 +37,20 @@ use tauri::{Emitter, Manager}; struct Shell { /// Session-only local folder authority; shutdown retires it before stopping the API server. host_access: Mutex>, - /// Named, because a restart policy that cannot say which process died cannot start it again. - children: Mutex>, /// Native Linux engine API owned by this desktop session, kept through Stop/Start. #[cfg(target_os = "linux")] podman_api: Mutex, /// Which run is the current one. /// - /// Stopping and starting again inside two seconds would otherwise leave the previous watcher - /// alive beside the new one, both answering the same death, and a process restarted twice is - /// one process and one orphan holding a port. + /// Stopping and starting again inside two seconds would otherwise leave the previous run's + /// connection watcher alive beside the new one, both sending the window back to setup for the + /// same refusal, and a recovery marked by the old run would outlive the run that cleared it. generation: std::sync::atomic::AtomicU64, /// Cancellation of pending Start is independent of the hosts still serving the previous run. start_generation: std::sync::atomic::AtomicU64, /// Stop serializes with synchronous startup side effects, never with an async wait. startup: Mutex<()>, - /// A cancelled attempt must finish returning its unpublished children before another starts. + /// A cancelled attempt must finish unwinding before another starts. starting: std::sync::atomic::AtomicBool, /// An explicit Stop must survive setup navigation; a new app session may resume again. stopped_in_session: std::sync::atomic::AtomicBool, @@ -1454,10 +1452,7 @@ fn reset_leftover_database_with( } let attempt = StartAttempt::begin(shell)?; let _startup = attempt.lock_current()?; - if shell.containers.lock().unwrap().is_some() - || shell.root.lock().unwrap().is_some() - || !shell.children.lock().unwrap().is_empty() - { + if shell.containers.lock().unwrap().is_some() || shell.root.lock().unwrap().is_some() { return Err(Problem::plain("OpenBot still owns running services. Choose Stop OpenBot before resetting leftover data.")); } if shell @@ -1531,7 +1526,7 @@ async fn reset_leftover_database( ) } -/// Write the `.env`, raise the containers, migrate, then start the three host processes. +/// Write the `.env`, raise the containers, migrate, then wait for the server container to answer. #[tauri::command] #[allow( clippy::too_many_arguments, @@ -1720,7 +1715,15 @@ async fn start_stack_inner( attempt.require_current()?; } - let (logs, bun, mut secrets, ports, local_browser) = { + // Minted before Compose, because the server is a container now and Compose is how it gets its + // environment. Never persisted: it goes into `secrets` only after they are written to the + // credential store, and the Compose file names it for the `server` service alone, so model + // workers, Bots and the worker cannot impersonate the native approval transport. + use base64::Engine as _; + let host_token = + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(rand::random::<[u8; 32]>()); + + let (found, secrets, ports) = { let _startup = attempt.lock_current()?; // Belt and braces: a fetch that reported success and left something out is still not a @@ -1749,7 +1752,7 @@ async fn start_stack_inner( #[cfg(target_os = "linux")] let status = found.status(); acquire::prepare_for_compose(&found)?; - let bun = preparation::require(&root, Some(&requested_harness), &found)?; + preparation::require(&root, Some(&requested_harness), &found)?; // An installation from before engines were saved is pinned to the one its images were // just found on. Best effort: Start does not fail over a hint for next time. if saved.is_none() { @@ -1781,7 +1784,14 @@ async fn start_stack_inner( *shell.leftover_database.lock().unwrap() = leftover_offer(problem, &root, &found); })?; - // Only this deployment's recorded hosts are reclaimed; its existing containers are reusable. + /* + * Only this deployment's recorded hosts are reclaimed; its existing containers are reusable. + * + * THIS IS ALSO THE UPGRADE. An installation from before the server was a container still + * has its Bun server, app and worker on the host, recorded in `host-pids.json` and holding + * the very ports the `server` container is about to publish. Stopping them here, and waiting + * for their sockets to close, is what lets that `up` bind them. + */ let previous_ports = openbot_env::Ports::read(&root).map_err(|error| { Problem::with("OpenBot could not read its local ports.", error.to_string()) })?; @@ -1847,6 +1857,40 @@ async fn start_stack_inner( settings.insert("OPENBOT_ORGANIZATION_AUTH_URL".into(), authority); } openbot_env::explain_browser_absence(&mut settings, browser_absent); + /* + * Each Bot's computer on the network the server container is on. + * + * The supervisor publishes a computer on a loopback port of the host unless this names a + * network to attach it to instead, and a loopback port on the host is not one the server + * container can open. `default` is the network the server shares with the Bots, under the + * project name Compose is about to give it. See `stack::project_name`. + */ + settings.insert( + "COMPUTER_NETWORK".into(), + format!("{}_default", stack::project_name(&root)), + ); + /* + * On Linux the server runs as the person, not as the image's user. + * + * The model sign-in file is theirs and readable by nobody else, and a bind mount on Linux + * keeps the host's numbers, so the image's 1001 could not read it. The deployment folder is + * the one this app made for them, so its owner is who that is, with no call out to `id`. + * Docker Desktop on macOS and Windows maps ownership itself, and there the default stays. + */ + #[cfg(target_os = "linux")] + { + use std::os::unix::fs::MetadataExt as _; + let owner = std::fs::metadata(&root).map_err(|error| { + Problem::with( + "OpenBot could not read its installation folder.", + format!("{}: {error}", root.display()), + ) + })?; + settings.insert( + "OPENBOT_SERVER_USER".into(), + format!("{}:{}", owner.uid(), owner.gid()), + ); + } seed_learning_default( &root, &mut settings, @@ -1886,13 +1930,32 @@ async fn start_stack_inner( report(&app, "env", true, "settings written, credentials stored"); // Compose gives inherited environment precedence over .env. Pin this run's chosen ports. secrets.extend(ports.settings()); - for key in ["PICKED_HARNESS_PORT", "OPENBOT_TOOL_URL"] { + for key in [ + "PICKED_HARNESS_PORT", + "OPENBOT_TOOL_URL", + "COMPUTER_NETWORK", + "OPENBOT_SERVER_USER", + ] { if let Some(value) = settings.get(key) { secrets.insert(key.into(), value.clone()); } } - // Set before Bun imports the runtime, and retained for supervised restarts. + // Handed to Compose with the rest; a service receives only what the Compose file names. secrets.extend(desktop_telemetry::runtime_env(&app)); + // After the credential store has been written, so the token never is. See where it is + // minted above. + secrets.insert("OPENBOT_DESKTOP_HOST_TOKEN".into(), host_token.clone()); + + // Compose's bind mount of the sign-in folder does not create it, because Docker would make + // it as root on Linux and this app could then not write a sign-in into it. So it is made + // here, as the person, before anything asks for it. + let sign_in_folder = root.join(".openbot"); + std::fs::create_dir_all(&sign_in_folder).map_err(|error| { + Problem::with( + "OpenBot could not prepare its sign-in folder.", + format!("{}: {error}", sign_in_folder.display()), + ) + })?; // Installation already verified these images. Start only raises the local containers; // its no-pull policy sends missing assets back to the installation step. @@ -1932,53 +1995,55 @@ async fn start_stack_inner( report(&app, "services", false, detail); })?; - // Before spawning: if these are still held, whatever answers later is not ours. - let mut hosts = vec![("API server", ports.server), ("app", ports.app)]; - if let Some(port) = ports.browser { - hosts.push(("browser", port)); - } - if let Some(problem) = stack::port_already_taken(&hosts) { - report(&app, "ports", false, problem.clone()); - return Err(problem.into()); - } - - let logs = root.join(".logs"); - (logs, bun, secrets, ports, local_browser) + (found, secrets, ports) }; - // Never persisted or passed to Compose. Only the server process receives this credential; - // model workers and frontend processes cannot impersonate the native approval transport. - use base64::Engine as _; - let host_token = - base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(rand::random::<[u8; 32]>()); - secrets.insert("OPENBOT_DESKTOP_HOST_TOKEN".into(), host_token.clone()); - - let logs_for_wait = logs.clone(); - let generation = start_host_processes( - &attempt, - &root, - &logs, - &bun, - &secrets, - local_browser.as_ref(), - |name| report(&app, name, true, "started"), - move |started| { - stack::wait_until_answering( - started, - &logs_for_wait, - &stack::Ready { - api: ports.server, - app: ports.app, - }, - std::time::Duration::from_secs(180), - ) - }, - ) + /* + * The server container answering, on both of the host ports it was given. + * + * No port check before this, as there was before spawning host processes: these ports are now + * published by this deployment's own `server` container, which `Ports::available` already + * counts as ours, and anybody else on them was avoided when the ports were chosen. Docker's + * restart policy keeps it up from here, so there is nothing to supervise after it answers. + */ + report_running(&app, "answering", "waiting for OpenBot to answer", None); + let ticket = attempt.generation; + let waiting = app.clone(); + let ready = stack::Ready { + api: ports.server, + app: ports.app, + }; + let answered = tauri::async_runtime::spawn_blocking(move || { + stack::wait_until_healthy(&ready, std::time::Duration::from_secs(180), || { + waiting + .state::() + .start_generation + .load(std::sync::atomic::Ordering::SeqCst) + == ticket + }) + }) .await - .inspect_err(|problem| report(&app, "answering", false, problem.said.clone()))?; + .unwrap_or_else(|error| Err(format!("the wait did not run: {error}"))); + // A Stop during the wait is the reason it ended, whatever the wait itself says. + attempt.require_current()?; + if let Err(error) = answered { + // What the server said while it was not answering, from Compose's log, with every + // credential this run handed out taken out of it. + let log = stack::service_log(&found, &root, "server", 40); + let detail = if log.is_empty() { + "The server container wrote nothing to its log.".to_string() + } else { + format!("Last from the server container (up to 40 lines):\n{log}") + }; + let problem = stack::startup_problem(Problem::with(error, detail), &secrets); + report(&app, "answering", false, problem.said.clone()); + return Err(problem); + } // Stop must not finish between accepting readiness and reporting a successful Start. let _startup = attempt.lock_current()?; - // Only a stack that answered successfully acquires a restart policy. + *shell.root.lock().unwrap() = Some(root.clone()); + clear_recovery_required(&shell, &root); + let generation = shell.generation.load(std::sync::atomic::Ordering::SeqCst); let address = shell .containers .lock() @@ -1988,7 +2053,7 @@ async fn start_stack_inner( .ok_or_else(|| Problem::plain("The local container runtime is unavailable."))?; let config = host_access::HostAccessConfig::new( format!("http://127.0.0.1:{}", ports.server), - host_token, + host_token.clone(), address, deployment::reference(&root, "agent-computer")?, vec![root.clone()], @@ -2007,16 +2072,7 @@ async fn start_stack_inner( ) })?; preparation::save_selected_root(&config, &root)?; - supervise_host_processes( - app.clone(), - root, - logs, - bun, - secrets, - generation, - ports, - local_browser, - ); + watch_connections(app.clone(), root, host_token, generation, ports.server); report(&app, "answering", true, "the API and the app are answering"); Ok(()) @@ -2053,8 +2109,8 @@ fn shutdown_root(shell: &Shell, fallback_root: &Path) -> PathBuf { root } -/// Take the whole stack down: the host processes, anything left over, the containers, and the -/// local Intelligence stack if OpenBot created it. +/// Take the whole stack down: any host processes an installation from before the server was a +/// container left running, the containers, and the local Intelligence stack if OpenBot created it. /// /// One implementation, because there are three ways to ask for it (the button, the menu bar, and /// quitting) and a person who used one of them and got a different amount of stopping would be @@ -2463,102 +2519,16 @@ fn down_containers_on_quit(shell: &Shell, root: &Path) -> Result<(), String> { down_owned_containers(shell, root) } -/// Reclaim held replacements before consulting durable inventory. Keep the handles and pidfile -/// if any phase fails, so the next Stop or Start can retry with the same ownership evidence. -fn cleanup_host_children( - root: &Path, - children: &mut Vec<(&'static str, std::process::Child)>, - cleanup: C, -) -> Result -where - C: FnOnce(&Path) -> Result, -{ - let held = stack::stop_host_children(root, children)?; - stop_held_process_handles(children)?; - let recorded = cleanup(root)?; - Ok(held + recorded) -} - -fn stop_held_process_handles( - children: &mut Vec<(&'static str, std::process::Child)>, -) -> Result<(), Problem> { - for (name, child) in children.iter_mut() { - let failure = |error| { - Problem::with( - "OpenBot could not stop one of its host processes.", - format!("could not finish stopping held {name}: {error}"), - ) - }; - if child.try_wait().map_err(failure)?.is_none() { - child.kill().map_err(failure)?; - } - child.wait().map_err(failure)?; - } - children.clear(); - Ok(()) -} - -fn cleanup_after_host_recording_failure( - shell: &Shell, - root: &Path, - children: &mut Vec<(&'static str, std::process::Child)>, - recording: Problem, - cleanup: C, - force_handles: F, -) -> Result -where - C: FnOnce(&Path, &mut Vec<(&'static str, std::process::Child)>) -> Result, - F: FnOnce(&mut Vec<(&'static str, std::process::Child)>) -> Result<(), Problem>, -{ - shell - .generation - .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - let cleanup = cleanup(root, children); - let failure = match cleanup { - Ok(_) => { - *shell.root.lock().unwrap() = None; - return Err(recording); - } - Err(cleanup) => cleanup, - }; - let forced = force_handles(children); - let mut detail = recording.detail.unwrap_or_default(); - if !detail.is_empty() { - detail.push('\n'); - } - detail.push_str(&problem_detail(failure)); - match forced { - Ok(()) => { - *shell.root.lock().unwrap() = None; - } - Err(forced) => { - detail.push('\n'); - detail.push_str(&problem_detail(forced)); - } - } - Err(Problem::with(recording.said, detail)) -} - -#[cfg(test)] -fn retire_host_processes(shell: &Shell, root: &Path, cleanup: C) -> Result -where - C: FnOnce(&Path) -> Result, -{ - // Invalidate before waiting for a restart that already owns the lock. That restart either - // observes retirement before spawning, or publishes its handle before cleanup can proceed. - shell - .generation - .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - let _startup = shell.startup.lock().unwrap(); - cleanup_host_state(shell, root, cleanup) -} - +/// Stop what this session holds and then what the deployment's records name. The records are +/// how an installation from before the server was a container still has Bun processes on the +/// host to stop; a current one has none, and the cleanup finds nothing. Keep failed cleanup +/// owned for a later Stop. fn cleanup_host_state(shell: &Shell, root: &Path, cleanup: C) -> Result where C: FnOnce(&Path) -> Result, { // Also runs for startup replacement and failure recovery, so a retired authorization session - // cannot leave a job running beside a new server. Keep failed cleanup owned for a later Stop. + // cannot leave a job running beside a new server. let host_access_result = { let mut slot = shell.host_access.lock().unwrap(); let result = slot.as_ref().map_or(Ok(()), |broker| { @@ -2569,14 +2539,13 @@ where } result }; - let mut children = shell.children.lock().unwrap(); let selected = shell .root .lock() .unwrap() .clone() .unwrap_or_else(|| root.to_path_buf()); - let result = cleanup_host_children(&selected, &mut children, cleanup); + let result = cleanup(&selected); let result = match (host_access_result, result) { (Ok(()), result) => result, (Err(error), Ok(_)) => Err(Problem::with( @@ -2594,176 +2563,6 @@ where result } -/// The initial host launch, including ownership handoff on every outcome. -#[allow( - clippy::too_many_arguments, - reason = "The launch carries what every host process needs and what this start decided about the person's browser." -)] -async fn start_host_processes( - attempt: &StartAttempt<'_>, - root: &Path, - logs: &Path, - bun: &Path, - secrets: &stack::Secrets, - local_browser: Option<&local_browser::LocalBrowser>, - mut report_started: R, - wait: W, -) -> Result -where - R: FnMut(&'static str), - W: FnOnce(&mut Vec<(&'static str, std::process::Child)>) -> Result<(), String> + Send + 'static, -{ - let started = { - let _startup = attempt.lock_current()?; - let mut started = Vec::new(); - for process in stack::host_processes(local_browser.is_some()) { - if attempt.require_current().is_err() { - return finish_host_start_locked(attempt, root, started, Ok(())); - } - let child = - match stack::spawn_host_process(process, root, logs, bun, secrets, local_browser) { - Ok(child) => child, - Err(error) => { - return finish_host_start_locked( - attempt, - root, - started, - Err(format!("could not start {}: {error}", process.name)), - ); - } - }; - started.push((process.name, child)); - report_started(process.name); - } - // Stop and Quit need durable ownership while readiness is still waiting, especially on - // Windows where a deployment directory alone cannot authorize terminating a process. - if let Err(recording) = stack::record_host_processes( - root, - &started - .iter() - .map(|(name, child)| (*name, child.id())) - .collect::>(), - ) { - let shell = attempt.shell; - let mut children = shell.children.lock().unwrap(); - children.extend(started); - *shell.root.lock().unwrap() = Some(root.to_path_buf()); - return cleanup_after_host_recording_failure( - shell, - root, - &mut children, - recording, - |root, children| cleanup_host_children(root, children, stack::stop_processes_under), - stop_held_process_handles, - ); - } - if attempt.require_current().is_err() { - return finish_host_start_locked(attempt, root, started, Ok(())); - } - started - }; - // A failed blocking task must not drop the only handles either. The caller retains the - // vector while readiness borrows it; even a panic returns every child to the same cleanup. - let owned = std::sync::Arc::new(Mutex::new(started)); - let waiting = std::sync::Arc::clone(&owned); - let outcome = tauri::async_runtime::spawn_blocking(move || { - let mut started = waiting.lock().unwrap(); - wait(&mut started) - }) - .await - .unwrap_or_else(|error| Err(format!("the wait did not run: {error}"))); - let started = std::mem::take( - &mut *owned - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner), - ); - let readiness_failure = outcome.as_ref().err().cloned(); - finish_host_start(attempt, root, started, outcome).map_err(|problem| { - // Preserve cancellation and lifecycle failures. Only the actual readiness error gets - // the startup headline; cleanup details remain attached and are redacted with it. - if readiness_failure.as_deref() == Some(problem.said.as_str()) { - stack::startup_problem(problem, secrets) - } else { - problem - } - }) -} - -fn finish_host_start( - attempt: &StartAttempt<'_>, - root: &Path, - started: Vec<(&'static str, std::process::Child)>, - outcome: Result<(), String>, -) -> Result { - let _startup = attempt.shell.startup.lock().unwrap(); - finish_host_start_locked(attempt, root, started, outcome) -} - -fn finish_host_start_locked( - attempt: &StartAttempt<'_>, - root: &Path, - mut started: Vec<(&'static str, std::process::Child)>, - outcome: Result<(), String>, -) -> Result { - let shell = attempt.shell; - if let Err(cancelled) = attempt.require_current() { - // These handles were never published. Stop may already have finished, so this attempt - // must reap them itself. A concurrent initial Start is excluded until its ticket drops. - let cleaned = cleanup_host_children(root, &mut started, stack::stop_processes_under); - return match cleaned { - Ok(_) => Err(cancelled), - Err(cleanup) => { - let mut detail = problem_detail(cleanup); - if let Err(forced) = stop_held_process_handles(&mut started) { - detail.push('\n'); - detail.push_str(&problem_detail(forced)); - shell.children.lock().unwrap().extend(started); - *shell.root.lock().unwrap() = Some(root.to_path_buf()); - } - Err(Problem::with(cancelled.said, detail)) - } - }; - } - let mut children = shell.children.lock().unwrap(); - children.extend(started); - *shell.root.lock().unwrap() = Some(root.to_path_buf()); - if let Err(original) = outcome { - shell - .generation - .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - // The initial failure is the reason Start failed, even if its cleanup also needs help. - return match cleanup_host_children(root, &mut children, stack::stop_processes_under) { - Ok(_) => { - *shell.root.lock().unwrap() = None; - Err(original.into()) - } - Err(cleanup) => Err(Problem::with(original, problem_detail(cleanup))), - }; - } - // Keep handles only while the recording failure is being cleaned up. Reporting Start failure - // while leaving the just-spawned host processes alive would recreate the orphan this ownership - // record exists to prevent. - if let Err(recording) = stack::record_host_processes( - root, - &children - .iter() - .map(|(name, child)| (*name, child.id())) - .collect::>(), - ) { - return cleanup_after_host_recording_failure( - shell, - root, - &mut children, - recording, - |root, children| cleanup_host_children(root, children, stack::stop_processes_under), - stop_held_process_handles, - ); - } - attempt.require_current()?; - clear_recovery_required(shell, root); - Ok(shell.generation.load(std::sync::atomic::Ordering::SeqCst)) -} - fn require_no_exited_compose_services( found: &engine::Address, root: &Path, @@ -2805,9 +2604,9 @@ where C: FnOnce(&Path) -> Result, { attempt.require_current()?; - // Preflight can fail while the previous run still owns live hosts. Retire that watcher only - // when replacement actually begins reclaiming them, before taking the children lock, so a - // restart already in progress hands its child to this cleanup and never adopts the new run. + // Preflight can fail while the previous run is still serving. Retire its connection watcher + // only when replacement actually begins, so a refusal it finds now cannot be published + // against the run that replaces it. attempt .shell .generation @@ -3267,7 +3066,7 @@ fn show_openbot_on( if recovery_required_or_pending_quit_notice(&shell, &root) && !organization_recovery { return Err("Part of OpenBot needs recovery. Try starting OpenBot once more.".into()); } - let url = owned_app_url(&root, ports).ok_or_else(|| { + let url = owned_app_url(&shell, &root, ports).ok_or_else(|| { format!("OpenBot could not verify its app on port {port} belongs to this installation. Try starting OpenBot again.") })?; let authority = @@ -3747,7 +3546,7 @@ fn show_status_in_app(app: &tauri::AppHandle) -> bool { } let Some(mut url) = ports_for_shell(app) .ok() - .and_then(|ports| owned_app_url(&root, &ports)) + .and_then(|ports| owned_app_url(&app.state::(), &root, &ports)) .and_then(|url| url.parse::().ok()) else { return false; @@ -3772,9 +3571,11 @@ fn show_status_in_app(app: &tauri::AppHandle) -> bool { pub(crate) struct DesktopStatus { /// OpenBot answers on its own address, whoever started it. openbot_running: bool, - /// This session of the desktop app holds OpenBot's processes, so its supervisor can restart - /// them. False for an OpenBot left running by an earlier session that ended without its Quit. + /// This session of the desktop app started the OpenBot that is running, so it can restart it. + /// False for an OpenBot left running by an earlier session that ended without its Quit. openbot_held: bool, + /// Always empty now: OpenBot's server and worker are containers and are listed with them. + /// Kept so a Status window from before that still reads the shape it expects. processes: Vec, containers: Vec, /// The local Intelligence stack recorded for this installation, if any. @@ -3810,14 +3611,7 @@ fn read_desktop_status( ) -> DesktopStatus { let shell = app.state::(); let root = cleanup_root(&shell, &stack::default_root()); - let pids: Vec<(&'static str, u32)> = shell - .children - .lock() - .unwrap() - .iter() - .map(|(name, child)| (*name, child.id())) - .collect(); - let processes = resident_parts(&pids); + let held = shell.root.lock().unwrap().is_some() && shell.containers.lock().unwrap().is_some(); let stack_id = local_eval_state::read(&root, None).map(|state| state.stack_id().as_str().to_owned()); let found = engine::detect(); @@ -3854,11 +3648,11 @@ fn read_desktop_status( let ports = ports_for_shell(app).ok(); let answering = ports .as_ref() - .is_some_and(|ports| owned_app_url(&root, ports).is_some()); + .is_some_and(|ports| owned_app_url(&shell, &root, ports).is_some()); DesktopStatus { - openbot_running: !pids.is_empty() || answering, - openbot_held: !pids.is_empty(), - processes, + openbot_running: held || answering, + openbot_held: held, + processes: Vec::new(), containers, intelligence_stack: stack_id, intelligence_containers, @@ -3874,57 +3668,34 @@ fn read_desktop_status( } } -/// Each host process by name, with its resident memory where `ps` could read it. -fn resident_parts(pids: &[(&'static str, u32)]) -> Vec { - let sizes = if pids.is_empty() { - Vec::new() - } else { - let list = pids - .iter() - .map(|(_, pid)| pid.to_string()) - .collect::>() - .join(","); - quiet::command("ps") - .args(["-o", "pid=,rss=", "-p", &list]) - .output() - .ok() - .map(|output| desktop_status::resident(&String::from_utf8_lossy(&output.stdout))) - .unwrap_or_default() - }; - pids.iter() - .map(|(name, pid)| desktop_status::Part { - name: (*name).to_owned(), - memory_bytes: sizes - .iter() - .find(|(found, _)| found == pid) - .map(|(_, bytes)| *bytes), - }) - .collect() -} - /** -Restart OpenBot's host processes: each is ended, and the supervisor that keeps them alive starts it -again, as it would after a crash. The containers, its database among them, keep running, so nothing -is lost and the restart takes seconds. +Restart OpenBot itself: the server and the worker containers, through Compose. The other +containers, its database among them, keep running, so nothing is lost and the restart takes seconds. +Only an OpenBot this session started, whose engine it knows. */ #[tauri::command] -fn restart_openbot(app: tauri::AppHandle) -> Result<(), String> { +async fn restart_openbot(app: tauri::AppHandle) -> Result<(), String> { + // Compose takes seconds to restart a container. Off Tauri's threads, as Stop is. + tauri::async_runtime::spawn_blocking(move || restart_openbot_now(&app)) + .await + .map_err(|error| format!("the restart did not run: {error}"))? +} + +fn restart_openbot_now(app: &tauri::AppHandle) -> Result<(), String> { let shell = app.state::(); - let mut children = shell.children.lock().unwrap(); - if children.is_empty() { + let root = shell.root.lock().unwrap().clone(); + let address = shell + .containers + .lock() + .unwrap() + .as_ref() + .filter(|owned| Some(&owned.root) == root.as_ref()) + .map(|owned| owned.address.clone()); + let (Some(root), Some(address)) = (root, address) else { return Err("OpenBot is not running.".into()); - } - for (name, child) in children.iter_mut() { - // SAFETY: the pid is this app's own child, held and not yet reaped, so it cannot be reused. - #[cfg(unix)] - let ended = unsafe { libc::kill(child.id() as i32, libc::SIGTERM) } == 0; - #[cfg(not(unix))] - let ended = child.kill().is_ok(); - if !ended { - eprintln!("[status] could not restart {name}"); - } - } - Ok(()) + }; + stack::restart_openbot(&address, &root) + .map_err(|error| format!("OpenBot could not restart: {error}")) } /// Stop, as the menu used to offer it: everything, Intelligence included if OpenBot started it. @@ -4005,18 +3776,37 @@ fn already_running_now(app: &tauri::AppHandle, root: &str) let shell = app.state::(); let _startup = shell.startup.lock().unwrap(); !recovery_required_or_pending_quit_notice(&shell, &root) - && openbot_env::Ports::read(&root).is_ok_and(|ports| already_running_at(&root, &ports)) + && openbot_env::Ports::read(&root) + .is_ok_and(|ports| already_running_at(&shell, &root, &ports)) } -fn already_running_at(root: &Path, ports: &openbot_env::Ports) -> bool { - owned_app_url(root, ports).is_some() +fn already_running_at(shell: &Shell, root: &Path, ports: &openbot_env::Ports) -> bool { + owned_app_url(shell, root, ports).is_some() } -/// Neither an owned API nor an answering app port alone authorizes showing a deployment. -fn owned_app_url(root: &Path, ports: &openbot_env::Ports) -> Option { - if !already_running_on(root, ports.server, stack::recorded_server_owns_port) - || !stack::recorded_process_owns_port(root, "app", ports.app).unwrap_or(false) - { +/// Neither an owned API nor an answering app port alone authorizes showing a deployment: this +/// root's own `server` container has to be the one publishing both. +fn owned_app_url(shell: &Shell, root: &Path, ports: &openbot_env::Ports) -> Option { + let owned = shell + .containers + .lock() + .unwrap() + .as_ref() + .filter(|owned| owned.root == root) + .map(|owned| owned.address.clone()); + app_url_published_by(owned.as_ref(), root, ports) +} + +/// [`owned_app_url`] without a window: the Recorder's native host is a process of its own, started +/// by the browser, so it holds no engine and asks whichever one answers. +fn app_url_published_by( + owned: Option<&engine::Address>, + root: &Path, + ports: &openbot_env::Ports, +) -> Option { + if !already_running_on(root, ports.server, |root, server| { + server_container_publishes(owned, root, &[server, ports.app]) + }) { return None; } stack::app_url(ports.app) @@ -4134,12 +3924,44 @@ fn recorder_host_environment() -> recorder_host::HostEnvironment { app_url: Box::new(move || { let root = root.as_deref()?; let ports = openbot_env::Ports::read(root).ok()?; - owned_app_url(root, &ports) + app_url_published_by(None, root, &ports) }), organization_cookie, } } +/// Whether `root`'s `server` container publishes `ports`, asked of the engine this session raised +/// it on when there is one, and otherwise of whichever engine answers: an OpenBot left running by +/// an earlier session is still this deployment's, and its restart policy kept it up. +#[cfg(not(test))] +fn server_container_publishes( + owned: Option<&engine::Address>, + root: &Path, + ports: &[u16], +) -> Result { + let address = match owned { + Some(address) => address.clone(), + None => { + let status = engine::detect(); + match status.address.filter(|_| status.responding) { + Some(address) => address, + None => return Ok(false), + } + } + }; + stack::server_publishes(&address, root, ports) +} + +/// Tests never ask a real engine. They say which roots' server containers publish which ports. +#[cfg(test)] +fn server_container_publishes( + _owned: Option<&engine::Address>, + root: &Path, + ports: &[u16], +) -> Result { + Ok(tests::published_by_server(root, ports)) +} + /// What stopped the stack, if anything did, and forget it once it has been read. /// /// Cleared on reading so a failure from an hour ago does not greet somebody who has since fixed it. @@ -4230,7 +4052,9 @@ async fn ask_the_bot_inner(root: PathBuf, question: String) -> Result Option { candidates.into_iter().find(|path| path.exists()) } -/// Watch the three host processes and start one again when it dies. +/// Mark this run as needing its connection refreshed, and send the window to setup to do it. /// -/// The policy is in `supervise.rs`; this is the loop that applies it. It ends when the stack is -/// stopped, which is what clearing the root means, so stopping does not race a restart. +/// False when the run it was found for is no longer the current one, which a Stop or a new Start +/// in the meantime makes it. fn publish_connection_failure( app: &tauri::AppHandle, root: &Path, @@ -5738,39 +5562,32 @@ fn publish_connection_failure( Ok(true) } -#[allow( - clippy::too_many_arguments, - reason = "Retain selected ports alongside the supervised run's identity and credentials." -)] -fn supervise_host_processes( +/** +Watch a running OpenBot for a connection it reports refused, and send the window back to setup to +refresh it. + +What is left of the host-process supervisor. Docker's restart policy keeps the server and the worker +up now, so nothing here starts anything. The server still knows, as only it can, when the model, the +Intelligence or the organization sign-in it was given has been refused, and it tells the desktop +through the token minted for this run. The watcher ends with the run it was started for: a Stop or +another Start moves the generation on, and clearing the root says there is no run at all. +*/ +fn watch_connections( app: tauri::AppHandle, root: PathBuf, - logs: PathBuf, - bun: PathBuf, - // Carried rather than fetched again on each restart. A restart happens when something is - // already wrong, and a credential prompt at that moment is the worst time to ask for one. - secrets: stack::Secrets, + // Carried rather than fetched again: it was minted for this run and is kept nowhere else. + token: String, generation: u64, - ports: openbot_env::Ports, - // The person's browser this start runs, carried so a restart starts the same one. - local_browser: Option, + server_port: u16, ) -> std::thread::JoinHandle<()> { std::thread::spawn(move || { - eprintln!( - "[watch] supervising {} host processes", - stack::host_processes(local_browser.is_some()).count() - ); - let mut watches: Vec = stack::HOST_PROCESSES - .iter() - .map(|process| supervise::Watch::new(process.name)) - .collect(); - let connection_client = desktop_connection::client() - .map_err(|error| { - eprintln!("[watch] {error}"); - }) - .ok(); - let mut connection_notice_sent = false; - let mut connection_poll_failed = false; + let Ok(client) = desktop_connection::client().map_err(|error| { + eprintln!("[watch] {error}"); + }) else { + return; + }; + let mut notice_sent = false; + let mut poll_failed = false; loop { std::thread::sleep(std::time::Duration::from_secs(2)); @@ -5782,174 +5599,33 @@ fn supervise_host_processes( return; } - if connection_notice_sent && !recovery_required(&shell, &root) { - connection_notice_sent = false; - } - if !connection_notice_sent { - if let (Some(client), Some(token)) = ( - &connection_client, - secrets.get("OPENBOT_DESKTOP_HOST_TOKEN"), - ) { - match desktop_connection::poll(client, ports.server, token) { - Ok(Some(connection)) => { - match publish_connection_failure(&app, &root, generation, connection) { - Ok(published) => connection_notice_sent = published, - Err(error) => { - eprintln!("[watch] could not show connection refresh: {error}") - } - } - } - Ok(None) => connection_poll_failed = false, - Err(error) => { - if !connection_poll_failed { - eprintln!("[watch] {error}"); - } - connection_poll_failed = true; - } - } - } + if notice_sent && !recovery_required(&shell, &root) { + notice_sent = false; } - - // Which ones have died. Collected rather than acted on under the lock, because a - // restart waits, and waiting while holding the children is how Stop would block on a - // backoff nobody asked it to sit through. - let dead: Vec<&'static str> = { - let mut children = shell.children.lock().unwrap(); - let mut dead = Vec::new(); - for (name, child) in children.iter_mut() { - if let Ok(Some(_)) = child.try_wait() { - dead.push(*name); - } - } - dead - }; - - if !dead.is_empty() { - eprintln!("[watch] dead: {dead:?}"); + if notice_sent { + continue; } - for name in dead { - let Some(watch) = watches.iter_mut().find(|watch| watch.name == name) else { - continue; - }; - if !watch.should_restart(std::time::Instant::now()) { - // A restore already probing may finish first; publish recovery and its setup - // navigation together after it, so that late probe cannot undo this transition. - let _startup = shell.startup.lock().unwrap(); - // Let go of it. A dead child left in the list is found dead again two seconds - // later, and forever after: the count climbs past what actually happened, the - // window is sent back to the setup screen on a loop, and the giving up that was - // supposed to stop a hot laptop becomes one. - { - let mut children = shell.children.lock().unwrap(); - if shell.generation.load(std::sync::atomic::Ordering::SeqCst) != generation - || shell.root.lock().unwrap().as_deref() != Some(root.as_path()) - { - return; + match desktop_connection::poll(&client, server_port, &token) { + Ok(Some(connection)) => { + match publish_connection_failure(&app, &root, generation, connection) { + Ok(published) => notice_sent = published, + Err(error) => { + eprintln!("[watch] could not show connection refresh: {error}") } - children.retain(|(held, _)| *held != name); } - - let reason = watch.gave_up(); - mark_recovery_required(&shell, &root, generation); - report(&app, name, false, reason.clone()); - /* - * Both registers here too. `gave_up` names the process and quotes the tail of - * its log, which is the developer half; the person needs to know a piece of - * OpenBot stopped and that starting again is the thing to try. - */ - *shell.last_failure.lock().unwrap() = - Some(openbot_desktop_lib::problem::Problem::with( - format!( - "Part of OpenBot ({name}) stopped and could not be started again. \ - Try starting OpenBot once more." - ), - reason, - )); - // Back to the setup screen. By now the window is showing OpenBot, and OpenBot - // is not running: leaving it there is a window that lies. - let _ = show_setup(app.clone()); - continue; - } - report( - &app, - name, - false, - format!("{name} stopped. Starting it again."), - ); - std::thread::sleep(supervise::backoff(watch.restarts - 1)); - - // Asked again after the backoff: a stop, or another start, may have happened while - // this was waiting, and starting a process into either is how an orphan is made. - if shell.generation.load(std::sync::atomic::Ordering::SeqCst) != generation - || shell.root.lock().unwrap().is_none() - { - return; } - let Some(process) = stack::HOST_PROCESSES - .iter() - .find(|process| process.name == name) - else { - continue; - }; - match restart_host_process_with(&shell, &root, name, generation, || { - stack::spawn_host_process( - process, - &root, - &logs, - &bun, - &secrets, - local_browser.as_ref(), - ) - }) { - Ok(true) => report(&app, name, true, "started again"), - Ok(false) => return, - Err(problem) => { - report(&app, name, false, problem.said.clone()); - *shell.last_failure.lock().unwrap() = Some(problem); + Ok(None) => poll_failed = false, + Err(error) => { + if !poll_failed { + eprintln!("[watch] {error}"); } + poll_failed = true; } } } }) } -/// The same lock covers generation validation, launch, publication, and owned cleanup on every -/// platform. Stop can retire during spawn, but cannot finish before receiving that child handle. -fn restart_host_process_with( - shell: &Shell, - root: &Path, - name: &'static str, - generation: u64, - spawn: F, -) -> Result -where - F: FnOnce() -> std::io::Result, -{ - let mut children = shell.children.lock().unwrap(); - if shell.generation.load(std::sync::atomic::Ordering::SeqCst) != generation - || shell.root.lock().unwrap().as_deref() != Some(root) - { - return Ok(false); - } - let child = spawn().map_err(|error| { - Problem::with( - format!("OpenBot could not restart {name}."), - error.to_string(), - ) - })?; - #[cfg(unix)] - stack::replace_host_process(root, &mut children, name, child)?; - #[cfg(not(unix))] - stack::replace_windows_host_process_with( - root, - &mut children, - name, - child, - Path::new("powershell"), - )?; - Ok(shell.generation.load(std::sync::atomic::Ordering::SeqCst) == generation) -} - /// Point the window at OpenBot if it is up, and at the setup screen if it is not. /// /// Used by the tray and by a second launch, both of which happen at moments when the caller has no @@ -5985,7 +5661,7 @@ fn restore_window_on(app: &tauri::AppHandle, ports: &openb let destination = startup.as_ref().and_then(|_| { let root = cleanup_root(&shell, &stack::default_root()); if let Some(url) = (!recovery_required_or_pending_quit_notice(&shell, &root)) - .then(|| owned_app_url(&root, ports)) + .then(|| owned_app_url(&shell, &root, ports)) .flatten() { intelligence_bridge::allow(app, &url); @@ -6261,7 +5937,7 @@ fn install_menus( // The menu bar menu follows the system's appearance; its icons are drawn for each. let dark = system_is_dark(app); - let running = !app.state::().children.lock().unwrap().is_empty(); + let running = app.state::().root.lock().unwrap().is_some(); let dashboard = || { intelligence_menu_item( app.state::() @@ -12023,6 +11699,8 @@ fn main() { "docker-compose.yml", "--profile", "harness", + "--profile", + "openbot", "down", ], ); @@ -12485,33 +12163,33 @@ fn main() { let mut choice = serde_json::json!({"id":"byo-url", "agentUrl":remote}); let (expected_up, expected_image) = match case { "remote" | "remote-stale-image" => ( - "compose up -d --no-build --pull never postgres supervisor agent-computer agent-bot agent-langgraph", + "compose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot agent-langgraph", None, ), "anthropic-api" => { model = serde_json::json!({"provider":"anthropic", "login":"api-key", "apiKey":"synthetic-anthropic-key"}); choice = serde_json::json!({"id":"langgraph"}); - ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer agent-langgraph agent-harness", Some("agent-langgraph-agui")) + ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer server worker agent-langgraph agent-harness", Some("agent-langgraph-agui")) } "compatible" => { model = serde_json::json!({"provider":"openai-compatible", "login":"endpoint", "baseUrl":"http://127.0.0.1:11434/v1", "model":"synthetic-model", "apiKey":""}); - ("compose up -d --no-build --pull never postgres supervisor agent-computer agent-bot agent-langgraph", None) + ("compose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot agent-langgraph", None) } "installed" => { choice = serde_json::json!({"id":"langgraph"}); - ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer agent-bot agent-langgraph agent-harness", Some("agent-langgraph-agui")) + ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot agent-langgraph agent-harness", Some("agent-langgraph-agui")) } "none" => { choice = serde_json::Value::Null; - ("compose up -d --no-build --pull never postgres supervisor agent-computer agent-bot agent-langgraph", None) + ("compose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot agent-langgraph", None) } "chatgpt-plan" => { model = serde_json::json!({"provider":"openai", "login":"plan", "token":"{\"refresh_token\":\"synthetic-plan\"}"}); - ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer agent-harness", Some("agent-langgraph-agui")) + ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer server worker agent-harness", Some("agent-langgraph-agui")) } "claude-plan" => { model = serde_json::json!({"provider":"anthropic", "login":"plan", "token":"synthetic-claude-plan"}); - ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer agent-harness", Some("agent-claude-sdk")) + ("compose --profile harness up -d --no-build --pull never postgres supervisor agent-computer server worker agent-harness", Some("agent-claude-sdk")) } _ => panic!("unknown test case"), }; @@ -12592,6 +12270,24 @@ fn main() { ); assert!(commands.contains("\tcompose run --rm --pull never migrate\n")); assert!(!root.join(".logs").exists(), "no host runtime was launched"); + // The server container's own credential reaches Compose for `up`, and only Compose. + assert_eq!( + std::fs::read_to_string(root.join("host-token-at-up")).ok(), + Some("43".to_string()), + "case={case}: Compose must be handed the host token before it starts the server" + ); + assert!(!std::fs::read_to_string(root.join(".env")) + .unwrap() + .contains("OPENBOT_DESKTOP_HOST_TOKEN")); + // Compose's bind mount does not create the sign-in folder, so Start must have. + assert!(root.join(".openbot").is_dir()); + // Each Bot's computer goes on the network the server container is on. + assert_eq!( + openbot_env::read_already_set(&root.join(".env"), &["COMPUTER_NETWORK"]) + .unwrap() + .get("COMPUTER_NETWORK"), + Some(&format!("{}_default", stack::project_name(&root))) + ); let settings = openbot_env::read_already_set( &root.join(".env"), &[ @@ -12799,7 +12495,7 @@ fn main() { "{commands}" ); assert!(commands.contains("\tcompose version\n"), "{commands}"); - assert!(commands.contains("\tcompose up -d --no-build --pull never postgres supervisor agent-computer agent-bot agent-langgraph\n"), "{commands}"); + assert!(commands.contains("\tcompose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot agent-langgraph\n"), "{commands}"); assert!( commands.contains("\tcompose run --rm --pull never migrate\n"), "{commands}" @@ -12866,13 +12562,13 @@ fn main() { let commands = std::fs::read_to_string(&record).expect("command record"); assert!( commands.contains( - "\tcompose up -d --no-build --pull never postgres supervisor agent-computer agent-langgraph\n" + "\tcompose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-langgraph\n" ), "{commands}" ); assert!( !commands - .contains("compose up -d --no-build --pull never postgres supervisor agent-computer agent-bot"), + .contains("compose up -d --no-build --pull never postgres supervisor agent-computer server worker agent-bot"), "Anthropic Start must not target the OpenAI-only agent-bot: {commands}" ); assert_eq!(problem.said, "Part of OpenBot stopped during startup."); @@ -13569,665 +13265,118 @@ fn main() { std::fs::remove_dir_all(root).unwrap(); } - #[cfg(unix)] - struct InitialHostFixture { - base: PathBuf, - root: PathBuf, - bun: PathBuf, - pids: std::cell::RefCell>, - } - - #[cfg(unix)] - impl InitialHostFixture { - fn new(mode: &str) -> Self { - use std::os::unix::fs::PermissionsExt; - let base = temp_root("initial-host-launch"); - let root = base.join("deployment"); - std::fs::create_dir_all(&root).unwrap(); - for process in stack::host_processes(false) { - if mode == "first-fails" - || ((mode == "second-fails" || mode == "cleanup-refuses") - && process.name != "server") - { - break; - } - std::fs::create_dir(root.join(process.cwd)).unwrap(); - } - let bun = base.join("bun"); - // The production spawn boundary supplies cwd/argv/log files. Only the executable is - // synthetic: one direct child with no network, engine, or credential access. - std::fs::write( - &bun, - "#!/bin/sh\nprintf '%s' \"$$\" > child.pid\nexec /bin/sleep 60\n", - ) - .unwrap(); - std::fs::set_permissions(&bun, std::fs::Permissions::from_mode(0o700)).unwrap(); - Self { - base, - root, - bun, - pids: std::cell::RefCell::new(Vec::new()), - } - } - - fn observe(&self, name: &str) { - let path = self.root.join(name).join("child.pid"); - let until = std::time::Instant::now() + std::time::Duration::from_secs(5); - let pid = loop { - if let Ok(text) = std::fs::read_to_string(&path) { - if let Ok(pid) = text.parse::() { - break pid; - } - } - assert!(std::time::Instant::now() < until, "child did not start"); - std::thread::sleep(std::time::Duration::from_millis(5)); - }; - self.pids.borrow_mut().push(pid); - } - - fn alive(&self) -> Vec { - self.pids - .borrow() - .iter() - .copied() - .filter(|pid| unsafe { libc::kill(*pid as i32, 0) } == 0) - .collect() - } - } - - #[cfg(unix)] - impl Drop for InitialHostFixture { - fn drop(&mut self) { - // Even an old-code regression failure must not orphan the fixture. waitpid first - // proves this is still our direct, unreaped child; ECHILD never authorizes a signal. - for pid in self.pids.get_mut() { - if unsafe { libc::waitpid(*pid as i32, std::ptr::null_mut(), libc::WNOHANG) } == 0 { - unsafe { - libc::kill(*pid as i32, libc::SIGKILL); - libc::waitpid(*pid as i32, std::ptr::null_mut(), 0); - } - } - } - if self.base.is_file() { - std::fs::remove_file(&self.base).unwrap(); - } else { - std::fs::remove_dir_all(&self.base).unwrap(); - } - } - } - - #[cfg(unix)] - struct SupervisionFixture { - host: InitialHostFixture, - app: tauri::App, - watcher: Option>, - } - - #[cfg(unix)] - impl Drop for SupervisionFixture { - fn drop(&mut self) { - let shell = self.app.state::(); - shell - .generation - .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - stop_held_process_handles( - &mut shell - .children - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner), - ) - .unwrap(); - *shell - .root - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = None; - if let Some(watcher) = self.watcher.take() { - watcher.join().unwrap(); - } - eprintln!( - "{}", - serde_json::json!({"supervisionCleanup": self.host.root, - "heldChildren": shell.children.lock().unwrap_or_else(std::sync::PoisonError::into_inner).len(), "watcherJoined": true}) - ); - } + #[test] + fn recovery_is_resolved_only_by_a_completed_stop() { + let root = temp_root("recovery-completed-stop"); + let shell = Shell::default(); + mark_recovery_required(&shell, &root, 0); + *shell.last_failure.lock().unwrap() = Some(Problem::plain("server refused")); + assert!(recovery_required(&shell, &root)); + assert!(!recovery_required(&shell, &root.join("other"))); + assert!(stop_everything_with( + &shell, + &root, + |_| Ok(0), + |_| Err("synthetic container cleanup refusal".into()) + ) + .is_err()); + assert!( + recovery_required(&shell, &root), + "failed Stop cannot resolve recovery" + ); + stop_everything_with(&shell, &root, |_| Ok(0), |_| Ok(())).unwrap(); + assert!(!recovery_required(&shell, &root)); + assert!(shell.last_failure.lock().unwrap().is_none()); } - #[cfg(unix)] - fn failed_retry_keeps_survivor_supervised(failed_role: &str) { - let host = InitialHostFixture::new("success"); - std::fs::write( - &host.bun, - "#!/bin/sh\nprintf '%s' \"$$\" > child.pid\nif [ -f fail ]; then exit 71; fi\nexec /bin/sleep 300\n", - ).unwrap(); - let app = tauri::test::mock_builder() - .manage(Shell::default()) - .invoke_handler(tauri::generate_handler![start_stack]) - .build(tauri::test::mock_context(tauri::test::noop_assets())) - .unwrap(); - let window = tauri::WebviewWindowBuilder::new(&app, "main", Default::default()) - .build() - .unwrap(); - let mut fixture = SupervisionFixture { - host, - app, - watcher: None, + #[test] + fn recovery_notice_consumption_and_failed_retry_preserve_gate() { + let f = RestoreFixture::new(); + let setup = if cfg!(any(windows, target_os = "android")) { + "http://tauri.localhost/recovery" + } else { + "tauri://localhost/recovery" }; - let shell = fixture.app.state::(); - let setup = "tauri://localhost/failed-retry-setup"; - *shell.setup_url.lock().unwrap() = Some(setup.into()); - let attempt = StartAttempt::begin(&shell).unwrap(); - let generation = tauri::async_runtime::block_on(start_host_processes( - &attempt, - &fixture.host.root, - &fixture.host.root.join(".logs"), - &fixture.host.bun, - &stack::Secrets::new(), - None, - |name| fixture.host.observe(name), - |_| Ok(()), - )) - .unwrap(); - drop(attempt); - fixture.watcher = Some(supervise_host_processes( - fixture.app.handle().clone(), - fixture.host.root.clone(), - fixture.host.root.join(".logs"), - fixture.host.bun.clone(), - stack::Secrets::new(), - generation, - openbot_env::Ports::default(), - None, - )); - // The actual watcher exhausts its actual budget and backoffs after this role fails. - std::fs::write(fixture.host.root.join(failed_role).join("fail"), "").unwrap(); + let app = f.app(&f.owned, setup); + let shell = app.state::(); + *shell.root.lock().unwrap() = Some(f.owned.clone()); { - let mut children = shell.children.lock().unwrap(); - children - .iter_mut() - .find(|(name, _)| *name == failed_role) - .unwrap() - .1 - .kill() - .unwrap(); + let _startup = shell.startup.lock().unwrap(); + mark_recovery_required(&shell, &f.owned, 0); } - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(70); - loop { - if shell - .last_failure - .lock() - .unwrap() - .as_ref() - .is_some_and(|p| p.said.contains("could not be started again")) - && window.url().unwrap().as_str() == setup - { - break; - } - assert!( - std::time::Instant::now() < deadline, - "watcher did not produce recovery setup" - ); - std::thread::sleep(std::time::Duration::from_millis(20)); - } - let survivor_pid = { - let mut children = shell.children.lock().unwrap(); - assert!(!children.iter().any(|(name, _)| *name == failed_role)); - let (_, survivor) = children - .iter_mut() - .find(|(name, _)| *name == "worker") - .unwrap(); - assert!(survivor.try_wait().unwrap().is_none()); - survivor.id() - }; - // Deliberate invalid input rejects before saved-secret/deployment/engine access. This is - // the generated production Start handler, after recovery has exposed ordinary Start. + *shell.last_failure.lock().unwrap() = Some(Problem::plain("worker exhausted")); + assert!(last_failure_now(app.handle()).is_some()); + assert!(last_failure_now(app.handle()).is_none()); + assert!(recovery_required(&shell, &f.owned)); + write_quit_cleanup_notice( + &f.owned, + &["[exit] cleanup failed: Compose down failed: synthetic".into()], + ) + .unwrap(); + let notice = last_failure_now(app.handle()).expect("persisted Quit notice"); + assert_eq!(notice.said, "OpenBot had trouble shutting down last time."); + assert!(notice + .detail + .as_ref() + .unwrap() + .contains("containers stopped")); + assert!(!quit_cleanup_notice_path(&f.owned).exists()); + assert!( + recovery_required(&shell, &f.owned), + "consuming the persisted notice cannot make survivors adoptable" + ); + assert!(last_failure_now(app.handle()).is_none()); + let window = app.get_webview_window("main").unwrap(); let problem = tauri::test::get_ipc_response(&window, tauri::webview::InvokeRequest { cmd: "start_stack".into(), callback: tauri::ipc::CallbackFn(0), error: tauri::ipc::CallbackFn(1), - url: "tauri://localhost".parse().unwrap(), + url: setup.parse().unwrap(), body: tauri::ipc::InvokeBody::Json(serde_json::json!({ - "root": fixture.host.root, "apiUrl": "https://intelligence.example.test", + "root": f.owned, "apiUrl": "https://intelligence.example.test", "gatewayWsUrl": "wss://gateway.example.test", "apiKey": "synthetic-unused-key", "model": {"provider": "synthetic-invalid-provider", "login": "api-key"}, "harness": null, - })), - headers: Default::default(), invoke_key: tauri::test::INVOKE_KEY.into(), - }).expect_err("credential preflight must reject the synthetic provider"); + })), headers: Default::default(), invoke_key: tauri::test::INVOKE_KEY.into(), + }).expect_err("synthetic credential preflight must reject before store access"); assert!(problem["said"] .as_str() .unwrap() .contains("synthetic-invalid-provider")); assert_eq!( - shell.root.lock().unwrap().as_ref(), - Some(&fixture.host.root) - ); - assert!(!shell.starting.load(std::sync::atomic::Ordering::SeqCst)); - { - let mut children = shell.children.lock().unwrap(); - let (_, survivor) = children - .iter_mut() - .find(|(name, _)| *name == "worker") - .unwrap(); - assert_eq!(survivor.id(), survivor_pid); - assert!( - survivor.try_wait().unwrap().is_none(), - "preflight unexpectedly reclaimed survivor" - ); - survivor.kill().unwrap(); - } - // A subsequent real child death must still cause the existing watcher to restart it. - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(8); - let replacement = loop { - let replacement = shell - .children - .lock() - .unwrap() - .iter_mut() - .find(|(name, child)| *name == "worker" && child.id() != survivor_pid) - .and_then(|(_, child)| child.try_wait().unwrap().is_none().then_some(child.id())); - if replacement.is_some() - || fixture.watcher.as_ref().unwrap().is_finished() - || std::time::Instant::now() >= deadline - { - break replacement; - } - std::thread::sleep(std::time::Duration::from_millis(20)); - }; - eprintln!( - "{}", - serde_json::json!({ - "failedRole": failed_role, "recoveryUrl": window.url().unwrap().as_str(), - "ipcError": problem, "survivorPid": survivor_pid, "replacementPid": replacement, - "watcherRetired": fixture.watcher.as_ref().unwrap().is_finished(), - "generationBefore": generation, - "generationAfterPreflight": shell.generation.load(std::sync::atomic::Ordering::SeqCst), - }) + shell.generation.load(std::sync::atomic::Ordering::SeqCst), + 0 ); + assert!(recovery_required(&shell, &f.owned)); assert!( - replacement.is_some(), - "failed retry preflight abandoned the surviving host's watcher" + owned_app_url(&shell, &f.owned, &f.ports).is_some(), + "survivors must still answer" ); + assert!(show_openbot_on(app.handle().clone(), &f.ports).is_err()); + restore_window_on(app.handle(), &f.ports); + assert_eq!(window.url().unwrap().as_str(), setup); + // Reclaim may advance the active generation before a later Start failure. It still + // cannot clear the recovery marker; only accepted readiness can do that. + let attempt = StartAttempt::begin(&shell).unwrap(); + cleanup_before_start(app.handle(), &attempt, &f.owned, |_| Ok(0)).unwrap(); + assert!(recovery_required(&shell, &f.owned)); } - #[cfg(unix)] #[test] - fn worker_exhaustion_does_not_adopt_healthy_survivors() { - if crate::test_support::isolated_process( - "tests::worker_exhaustion_does_not_adopt_healthy_survivors", - ) { - return; - } - let host = InitialHostFixture::new("success"); - write_installed_deployment(&host.root); - let source = host.base.join("worker-recovery-host.rs"); - std::fs::write(&source, r#" -use std::{fs,io::{Read,Write},net::TcpListener,time::Duration}; -fn main() { - let cwd=std::env::current_dir().unwrap(); - let role=cwd.file_name().unwrap().to_str().unwrap(); - fs::write("child.pid",std::process::id().to_string()).unwrap(); - let mut starts=fs::OpenOptions::new().create(true).append(true).open("starts.log").unwrap(); - writeln!(starts,"{}",std::process::id()).unwrap(); - if role=="worker" { - loop { - if cwd.join("fail").exists() { std::process::exit(71); } - std::thread::sleep(Duration::from_millis(20)); - } - } - let port=match role { "server"=>3001,"app"=>3010,_=>panic!("unexpected role") }; - let listener=TcpListener::bind(("127.0.0.1",port)).unwrap(); - for stream in listener.incoming() { - let mut stream=stream.unwrap(); - stream.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); - let mut request=[0;2048]; - if stream.read(&mut request).unwrap_or(0)>0 { - let _=stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}"); - } - } -} -"#).unwrap(); - crate::test_support::compile_fixture(&source, &host.bun); - let app = tauri::test::mock_builder() - .manage(Shell::default()) - .invoke_handler(tauri::generate_handler![ - already_running, - show_openbot, - last_failure - ]) - .build(tauri::test::mock_context(tauri::test::noop_assets())) - .unwrap(); - let window = tauri::WebviewWindowBuilder::new(&app, "main", Default::default()) - .build() - .unwrap(); - let mut fixture = SupervisionFixture { - host, - app, - watcher: None, - }; - let shell = fixture.app.state::(); - let setup = "tauri://localhost/worker-recovery-setup"; - *shell.setup_url.lock().unwrap() = Some(setup.into()); - let attempt = StartAttempt::begin(&shell).unwrap(); - let logs = fixture.host.root.join(".logs"); - let wait_logs = logs.clone(); - let generation = tauri::async_runtime::block_on(start_host_processes( - &attempt, - &fixture.host.root, - &logs, - &fixture.host.bun, - &stack::Secrets::new(), - None, - |name| fixture.host.observe(name), - move |children| { - stack::wait_until_answering( - children, - &wait_logs, - &stack::Ready { - api: 3001, - app: 3010, - }, - std::time::Duration::from_secs(10), - ) - }, - )) - .unwrap(); - drop(attempt); - fixture.watcher = Some(supervise_host_processes( - fixture.app.handle().clone(), - fixture.host.root.clone(), - logs, - fixture.host.bun.clone(), - stack::Secrets::new(), - generation, - openbot_env::Ports::default(), - None, - )); - std::fs::write(fixture.host.root.join("worker/fail"), "").unwrap(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(70); - loop { - if shell - .last_failure - .lock() - .unwrap() - .as_ref() - .is_some_and(|p| { - p.said.contains("(worker)") && p.said.contains("could not be started again") - }) - && window.url().unwrap().as_str() == setup - { - break; - } - assert!( - std::time::Instant::now() < deadline, - "worker did not exhaust actual restart budget" - ); - std::thread::sleep(std::time::Duration::from_millis(20)); - } - assert!(server_capabilities_answer(3001)); - assert!(stack::recorded_server_owns_port(&fixture.host.root, 3001).unwrap()); - assert!(stack::recorded_process_owns_port(&fixture.host.root, "app", 3010).unwrap()); - let invoke = |command: &str, body: serde_json::Value| { - tauri::test::get_ipc_response( - &window, - tauri::webview::InvokeRequest { - cmd: command.into(), - callback: tauri::ipc::CallbackFn(0), - error: tauri::ipc::CallbackFn(1), - url: "tauri://localhost".parse().unwrap(), - body: tauri::ipc::InvokeBody::Json(body), - headers: Default::default(), - invoke_key: tauri::test::INVOKE_KEY.into(), - }, - ) - .map(|response| response.deserialize::().unwrap()) + fn pending_quit_notice_blocks_restore_before_react_consumes_it() { + let f = RestoreFixture::new(); + let setup = if cfg!(any(windows, target_os = "android")) { + "http://tauri.localhost/recovery" + } else { + "tauri://localhost/recovery" }; - let result = invoke( - "already_running", - serde_json::json!({"root":fixture.host.root}), + write_quit_cleanup_notice( + &f.owned, + &["[exit] cleanup failed: Compose down failed: synthetic".into()], ) - .unwrap() - .as_bool() .unwrap(); - let children = shell.children.lock().unwrap(); - let survivors: Vec<_> = children - .iter() - .map(|(name, child)| (*name, child.id())) - .collect(); - drop(children); - let starts = std::fs::read_to_string(fixture.host.root.join("worker/starts.log")).unwrap(); - let failure = shell.last_failure.lock().unwrap().clone(); - println!( - "WORKER_RECOVERY_PROOF={}", - serde_json::json!({ - "root":fixture.host.root,"base":fixture.host.base,"generation":generation, - "setupUrl":window.url().unwrap().as_str(),"alreadyRunning":result, - "survivors":survivors,"workerStarts":starts,"failure":failure, - }) - ); - assert_eq!(starts.lines().count(), supervise::MAX_RESTARTS as usize + 1); + let app = f.app(&f.owned, setup); + let shell = app.state::(); + let window = app.get_webview_window("main").unwrap(); assert!( - !result, - "exhausted worker was automatically adopted through generated already_running IPC" - ); - let notification = invoke("last_failure", serde_json::json!({})).unwrap(); - assert!(notification["said"].as_str().unwrap().contains("(worker)")); - assert!(invoke("last_failure", serde_json::json!({})) - .unwrap() - .is_null()); - assert!(recovery_required(&shell, &fixture.host.root)); - let show_error = invoke("show_openbot", serde_json::json!({})).unwrap_err(); - assert!(show_error["said"] - .as_str() - .unwrap() - .contains("needs recovery")); - restore_window_on(fixture.app.handle(), &openbot_env::Ports::default()); - assert_eq!(window.url().unwrap().as_str(), setup); - println!( - "WORKER_RECOVERY_COMMANDS={}", - serde_json::json!({ - "notification":notification,"generatedShowError":show_error, - "recoveryAfterNoticeRead":true,"restoreUrl":window.url().unwrap().as_str(), - }) - ); - } - - #[cfg(unix)] - #[test] - fn recovery_transient_worker_restart_does_not_require_recovery() { - let host = InitialHostFixture::new("success"); - let app = tauri::test::mock_builder() - .manage(Shell::default()) - .build(tauri::test::mock_context(tauri::test::noop_assets())) - .unwrap(); - let mut fixture = SupervisionFixture { - host, - app, - watcher: None, - }; - let shell = fixture.app.state::(); - let attempt = StartAttempt::begin(&shell).unwrap(); - let generation = tauri::async_runtime::block_on(start_host_processes( - &attempt, - &fixture.host.root, - &fixture.host.root.join(".logs"), - &fixture.host.bun, - &stack::Secrets::new(), - None, - |name| fixture.host.observe(name), - |_| Ok(()), - )) - .unwrap(); - drop(attempt); - fixture.watcher = Some(supervise_host_processes( - fixture.app.handle().clone(), - fixture.host.root.clone(), - fixture.host.root.join(".logs"), - fixture.host.bun.clone(), - stack::Secrets::new(), - generation, - openbot_env::Ports::default(), - None, - )); - let original = { - let mut children = shell.children.lock().unwrap(); - let (_, worker) = children - .iter_mut() - .find(|(name, _)| *name == "worker") - .unwrap(); - worker.kill().unwrap(); - worker.id() - }; - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(8); - let replacement = loop { - let replacement = shell - .children - .lock() - .unwrap() - .iter_mut() - .find(|(name, child)| *name == "worker" && child.id() != original) - .and_then(|(_, child)| child.try_wait().unwrap().is_none().then_some(child.id())); - if let Some(pid) = replacement { - break pid; - } - assert!( - std::time::Instant::now() < deadline, - "transient worker was not restarted" - ); - std::thread::sleep(std::time::Duration::from_millis(10)); - }; - assert!(!recovery_required(&shell, &fixture.host.root)); - assert!(shell.last_failure.lock().unwrap().is_none()); - println!( - "WORKER_TRANSIENT_PROOF={}", - serde_json::json!({"original":original,"replacement":replacement,"generation":generation,"recoveryRequired":false}) - ); - } - - #[cfg(unix)] - #[test] - fn recovery_ready_start_and_completed_stop_resolve_condition() { - let host = InitialHostFixture::new("success"); - let shell = Shell::default(); - mark_recovery_required(&shell, &host.root, 0); - *shell.last_failure.lock().unwrap() = Some(Problem::plain("worker exhausted")); - assert!(recovery_required(&shell, &host.root)); - assert!(!recovery_required(&shell, &host.base.join("other"))); - let attempt = StartAttempt::begin(&shell).unwrap(); - tauri::async_runtime::block_on(start_host_processes( - &attempt, - &host.root, - &host.root.join(".logs"), - &host.bun, - &stack::Secrets::new(), - None, - |name| host.observe(name), - |_| Ok(()), - )) - .unwrap(); - drop(attempt); - assert!(!recovery_required(&shell, &host.root)); - assert!(shell.last_failure.lock().unwrap().is_none()); - mark_recovery_required(&shell, &host.root, 0); - assert!( - stop_everything_with(&shell, &host.root, stack::stop_processes_under, |_| Err( - "synthetic container cleanup refusal".into() - )) - .is_err() - ); - assert!( - recovery_required(&shell, &host.root), - "failed Stop cannot resolve recovery" - ); - stop_everything_with(&shell, &host.root, stack::stop_processes_under, |_| Ok(())).unwrap(); - assert!(!recovery_required(&shell, &host.root)); - assert!(host.alive().is_empty()); - } - - #[test] - fn recovery_notice_consumption_and_failed_retry_preserve_gate() { - let f = RestoreFixture::new(); - let setup = if cfg!(any(windows, target_os = "android")) { - "http://tauri.localhost/recovery" - } else { - "tauri://localhost/recovery" - }; - let app = f.app(&f.owned, setup); - let shell = app.state::(); - *shell.root.lock().unwrap() = Some(f.owned.clone()); - { - let _startup = shell.startup.lock().unwrap(); - mark_recovery_required(&shell, &f.owned, 0); - } - *shell.last_failure.lock().unwrap() = Some(Problem::plain("worker exhausted")); - assert!(last_failure_now(app.handle()).is_some()); - assert!(last_failure_now(app.handle()).is_none()); - assert!(recovery_required(&shell, &f.owned)); - write_quit_cleanup_notice( - &f.owned, - &["[exit] cleanup failed: Compose down failed: synthetic".into()], - ) - .unwrap(); - let notice = last_failure_now(app.handle()).expect("persisted Quit notice"); - assert_eq!(notice.said, "OpenBot had trouble shutting down last time."); - assert!(notice - .detail - .as_ref() - .unwrap() - .contains("containers stopped")); - assert!(!quit_cleanup_notice_path(&f.owned).exists()); - assert!( - recovery_required(&shell, &f.owned), - "consuming the persisted notice cannot make survivors adoptable" - ); - assert!(last_failure_now(app.handle()).is_none()); - let window = app.get_webview_window("main").unwrap(); - let problem = tauri::test::get_ipc_response(&window, tauri::webview::InvokeRequest { - cmd: "start_stack".into(), callback: tauri::ipc::CallbackFn(0), error: tauri::ipc::CallbackFn(1), - url: setup.parse().unwrap(), - body: tauri::ipc::InvokeBody::Json(serde_json::json!({ - "root": f.owned, "apiUrl": "https://intelligence.example.test", - "gatewayWsUrl": "wss://gateway.example.test", "apiKey": "synthetic-unused-key", - "model": {"provider": "synthetic-invalid-provider", "login": "api-key"}, "harness": null, - })), headers: Default::default(), invoke_key: tauri::test::INVOKE_KEY.into(), - }).expect_err("synthetic credential preflight must reject before store access"); - assert!(problem["said"] - .as_str() - .unwrap() - .contains("synthetic-invalid-provider")); - assert_eq!( - shell.generation.load(std::sync::atomic::Ordering::SeqCst), - 0 - ); - assert!(recovery_required(&shell, &f.owned)); - assert!( - owned_app_url(&f.owned, &f.ports).is_some(), - "survivors must still answer" - ); - assert!(show_openbot_on(app.handle().clone(), &f.ports).is_err()); - restore_window_on(app.handle(), &f.ports); - assert_eq!(window.url().unwrap().as_str(), setup); - // Reclaim may advance the active generation before a later Start failure. It still - // cannot clear the recovery marker; only accepted readiness can do that. - let attempt = StartAttempt::begin(&shell).unwrap(); - cleanup_before_start(app.handle(), &attempt, &f.owned, |_| Ok(0)).unwrap(); - assert!(recovery_required(&shell, &f.owned)); - } - - #[test] - fn pending_quit_notice_blocks_restore_before_react_consumes_it() { - let f = RestoreFixture::new(); - let setup = if cfg!(any(windows, target_os = "android")) { - "http://tauri.localhost/recovery" - } else { - "tauri://localhost/recovery" - }; - write_quit_cleanup_notice( - &f.owned, - &["[exit] cleanup failed: Compose down failed: synthetic".into()], - ) - .unwrap(); - let app = f.app(&f.owned, setup); - let shell = app.state::(); - let window = app.get_webview_window("main").unwrap(); - assert!( - owned_app_url(&f.owned, &f.ports).is_some(), + owned_app_url(&shell, &f.owned, &f.ports).is_some(), "survivors must still answer before restore" ); assert!(!recovery_required(&shell, &f.owned)); @@ -14327,213 +13476,17 @@ fn main() { assert!(recovery_required(&app.state::(), &f.owned)); } - #[cfg(unix)] #[test] - fn failed_retry_after_server_exhaustion_keeps_survivor_supervised() { - failed_retry_keeps_survivor_supervised("server"); - } - - #[cfg(unix)] - #[test] - fn failed_retry_after_app_exhaustion_keeps_survivor_supervised() { - failed_retry_keeps_survivor_supervised("app"); - } - - #[cfg(unix)] - fn initial_host_case(mode: &'static str) { - let fixture = InitialHostFixture::new(mode); - let shell = Shell::default(); - let attempt = StartAttempt::begin(&shell).unwrap(); - let result = tauri::async_runtime::block_on(start_host_processes( - &attempt, - &fixture.root, - &fixture.root.join(".logs"), - &fixture.bun, - &stack::Secrets::new(), - None, - |name| { - fixture.observe(name); - if mode == "cleanup-refuses" { - // A real filesystem failure prevents both the second spawn and verification - // of the first child's deployment. No cleanup failure is mocked away. - std::fs::remove_dir_all(&fixture.base).unwrap(); - std::fs::write(&fixture.base, b"blocked fixture parent").unwrap(); - } - }, - move |_| match mode { - "wait-fails" => Err("synthetic readiness failure".into()), - "wait-panics" => panic!("synthetic readiness task panic"), - "success" => Ok(()), - _ => panic!("readiness must not run after a spawn failure"), - }, - )); - let alive = fixture.alive(); - eprintln!( - "{}", - serde_json::json!({ - "initialHostCase": mode, - "spawned": fixture.pids.borrow().len(), - "aliveAfterStart": alive, - "heldAfterStart": shell.children.lock().unwrap().len(), - "error": result.as_ref().err().map(|problem| &problem.said), - }) - ); - if mode == "success" { - assert!(result.is_ok(), "{result:?}"); - assert_eq!(alive.len(), 3); - assert_eq!(stack::recorded_host_pids(&fixture.root).unwrap().len(), 3); - } else { - let problem = result.unwrap_err(); - let expected = match mode { - "first-fails" => "could not start server:", - "second-fails" | "cleanup-refuses" => "could not start app:", - "wait-fails" => "synthetic readiness failure", - "wait-panics" => "the wait did not run:", - _ => unreachable!(), - }; - if matches!(mode, "wait-fails" | "wait-panics") { - assert!( - problem.said.contains("could not finish starting"), - "{problem:?}" - ); - assert!( - problem - .detail - .as_deref() - .unwrap_or_default() - .starts_with(expected), - "{problem:?}" - ); - } else { - assert!(problem.said.starts_with(expected), "{problem:?}"); - } - if mode == "cleanup-refuses" { - assert_eq!(alive.len(), 1); - assert_eq!(shell.children.lock().unwrap().len(), 1); - assert_eq!(shell.root.lock().unwrap().as_ref(), Some(&fixture.root)); - assert!( - problem.detail.is_some(), - "cleanup failure must remain visible" - ); - std::fs::remove_file(&fixture.base).unwrap(); - std::fs::create_dir_all(&fixture.root).unwrap(); - } else { - assert!( - alive.is_empty(), - "failed Start left owned children alive: {alive:?}" - ); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - } - } - if mode == "success" || mode == "cleanup-refuses" { - retire_host_processes(&shell, &fixture.root, stack::stop_processes_under).unwrap(); - assert!(fixture.alive().is_empty()); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - } - } - - #[cfg(unix)] - #[test] - fn initial_start_stopped_during_readiness_cannot_publish_and_cleans_real_children() { - use std::sync::{atomic::Ordering::SeqCst, Arc, Barrier}; - let fixture = InitialHostFixture::new("success"); - let shell = Arc::new(Shell::default()); - let entered = Arc::new(Barrier::new(2)); - let release = Arc::new(Barrier::new(2)); - let starter = { - let (shell, root, bun, entered, release) = ( - shell.clone(), - fixture.root.clone(), - fixture.bun.clone(), - entered.clone(), - release.clone(), - ); - std::thread::spawn(move || { - let attempt = StartAttempt::begin(&shell).unwrap(); - tauri::async_runtime::block_on(start_host_processes( - &attempt, - &root, - &root.join(".logs"), - &bun, - &stack::Secrets::new(), - None, - |_| {}, - move |_| { - entered.wait(); - release.wait(); - Ok(()) - }, - )) - }) - }; - entered.wait(); - for process in stack::host_processes(false) { - fixture.observe(process.cwd); - } - assert_eq!( - stack::recorded_host_pids(&fixture.root).unwrap().len(), - stack::host_processes(false).count() - ); - stop_everything_with(&shell, &fixture.root, stack::stop_processes_under, |_| { - Ok(()) - }) - .unwrap(); - let stopped_generation = shell.generation.load(SeqCst); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - release.wait(); - let result = starter.join().unwrap(); - let published = shell.children.lock().unwrap().len(); - let active_root = shell.root.lock().unwrap().clone(); - let completed_generation = shell.generation.load(SeqCst); - let alive = fixture.alive(); - // The old-code run must also leave no fixture processes behind. - retire_host_processes(&shell, &fixture.root, stack::stop_processes_under).unwrap(); - eprintln!( - "{}", - serde_json::json!({ - "stoppedGeneration": stopped_generation, - "completedGeneration": completed_generation, - "publishedAfterStop": published, - "aliveAfterStartFinished": alive, - "result": result.as_ref().err().map(|problem| &problem.said), - }) - ); - assert!( - result.is_err(), - "Start succeeded after Stop completed: {result:?}" - ); - assert_eq!(completed_generation, stopped_generation); - assert_eq!(published, 0); - assert!(active_root.is_none()); - assert!( - alive.is_empty(), - "cancelled Start leaked its children: {alive:?}" - ); - } - - #[test] - fn initial_start_cancelled_during_preparation_never_launches_hosts() { + fn initial_start_cancelled_during_preparation_cannot_take_the_startup_lock() { let shell = Shell::default(); let attempt = StartAttempt::begin(&shell).unwrap(); // Deployment/download or dependency preparation returns after Stop has completed. stop_everything_with(&shell, Path::new("unused-root"), |_| Ok(0), |_| Ok(())).unwrap(); - let problem = tauri::async_runtime::block_on(start_host_processes( - &attempt, - Path::new("unused-root"), - Path::new("unused-logs"), - Path::new("must-not-be-launched"), - &stack::Secrets::new(), - None, - |_| panic!("cancelled preparation launched a host"), - |_| panic!("cancelled preparation reached readiness"), - )) - .unwrap_err(); + let problem = attempt + .lock_current() + .expect_err("a cancelled Start resumed a side effect"); assert_eq!(problem.said, StartAttempt::cancelled().said); assert!(shell.root.lock().unwrap().is_none()); - assert!(shell.children.lock().unwrap().is_empty()); assert!( StartAttempt::begin(&shell).is_err(), "cancelled attempt is still unwinding" @@ -14601,283 +13554,6 @@ fn main() { } } - #[cfg(unix)] - #[test] - fn initial_host_second_spawn_failure_cleans_the_real_first_child() { - initial_host_case("second-fails"); - } - - #[cfg(unix)] - #[test] - fn initial_host_first_spawn_failure_never_waits_or_publishes() { - initial_host_case("first-fails"); - } - - #[cfg(unix)] - #[test] - fn initial_host_success_records_then_stops_all_children() { - initial_host_case("success"); - } - - #[cfg(unix)] - #[test] - fn initial_host_readiness_failure_preserves_error_and_cleans_children() { - initial_host_case("wait-fails"); - } - - #[cfg(unix)] - #[test] - fn initial_host_wait_panic_keeps_children_available_for_cleanup() { - initial_host_case("wait-panics"); - } - - #[cfg(unix)] - #[test] - fn initial_host_cleanup_refusal_keeps_original_error_and_stop_ownership() { - initial_host_case("cleanup-refuses"); - } - - #[cfg(unix)] - #[test] - fn failed_host_recording_retires_children_and_preserves_recording_failure() { - let root = temp_root("failed-recording-lifecycle"); - std::fs::create_dir_all(&root).unwrap(); - let worker = std::process::Command::new("/bin/sleep") - .arg("60") - .current_dir(&root) - .spawn() - .unwrap(); - let children = vec![("bogus", worker)]; - let shell = Shell::default(); - let attempt = StartAttempt::begin(&shell).unwrap(); - let result = finish_host_start(&attempt, &root, children, Ok(())).unwrap_err(); - assert_eq!( - result.said, - "OpenBot could not verify its host process ownership." - ); - assert!( - result - .detail - .as_deref() - .is_some_and(|detail| detail.contains("invalid host launch bogus")), - "{result:?}" - ); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - assert_eq!( - shell.generation.load(std::sync::atomic::Ordering::SeqCst), - 1 - ); - assert!( - !restart_host_process_with(&shell, &root, "worker", 0, || panic!( - "failed recording attempt restarted" - )) - .unwrap() - ); - std::fs::remove_dir_all(root).unwrap(); - } - - #[cfg(unix)] - #[test] - fn failed_host_recording_keeps_root_and_handles_when_forced_cleanup_fails() { - let root = temp_root("failed-recording-stubborn-child"); - std::fs::create_dir_all(&root).unwrap(); - let child = std::process::Command::new("/bin/sleep") - .arg("60") - .current_dir(&root) - .spawn() - .unwrap(); - let pid = child.id(); - let mut children = vec![("server", child)]; - let shell = Shell::default(); - *shell.root.lock().unwrap() = Some(root.clone()); - let result = cleanup_after_host_recording_failure( - &shell, - &root, - &mut children, - Problem::with("recording failed", "recording detail"), - |_, _| { - Err(Problem::with( - "normal cleanup failed", - "normal cleanup detail", - )) - }, - |_| { - Err(Problem::with( - "forced cleanup failed", - "forced cleanup detail", - )) - }, - ) - .unwrap_err(); - - assert_eq!(result.said, "recording failed"); - let detail = result.detail.as_deref().unwrap_or_default(); - assert!(detail.contains("recording detail"), "{detail}"); - assert!(detail.contains("normal cleanup detail"), "{detail}"); - assert!(detail.contains("forced cleanup detail"), "{detail}"); - assert_eq!(shell.root.lock().unwrap().as_deref(), Some(root.as_path())); - assert_eq!(children.len(), 1); - assert_eq!(children[0].1.id(), pid); - for (_, child) in children.iter_mut() { - let _ = child.kill(); - let _ = child.wait(); - } - std::fs::remove_dir_all(root).unwrap(); - } - - #[cfg(unix)] - #[test] - fn failed_host_recording_clears_root_and_handles_when_forced_cleanup_succeeds() { - let root = temp_root("failed-recording-forced-cleanup"); - std::fs::create_dir_all(&root).unwrap(); - let child = std::process::Command::new("/bin/sleep") - .arg("60") - .current_dir(&root) - .spawn() - .unwrap(); - let mut children = vec![("server", child)]; - let shell = Shell::default(); - *shell.root.lock().unwrap() = Some(root.clone()); - let result = cleanup_after_host_recording_failure( - &shell, - &root, - &mut children, - Problem::with("recording failed", "recording detail"), - |_, _| { - Err(Problem::with( - "normal cleanup failed", - "normal cleanup detail", - )) - }, - |children| { - for (_, child) in children.iter_mut() { - let _ = child.kill(); - let _ = child.wait(); - } - children.clear(); - Ok(()) - }, - ) - .unwrap_err(); - - assert_eq!(result.said, "recording failed"); - let detail = result.detail.as_deref().unwrap_or_default(); - assert!(detail.contains("recording detail"), "{detail}"); - assert!(detail.contains("normal cleanup detail"), "{detail}"); - assert!(!detail.contains("forced cleanup"), "{detail}"); - assert!(shell.root.lock().unwrap().is_none()); - assert!(children.is_empty()); - std::fs::remove_dir_all(root).unwrap(); - } - - #[cfg(unix)] - #[test] - fn failed_readiness_retires_children_and_preserves_original_failure() { - let root = temp_root("failed-readiness-lifecycle"); - std::fs::create_dir_all(&root).unwrap(); - let mut failed = std::process::Command::new("/bin/sh") - .args(["-c", "exit 71"]) - .spawn() - .unwrap(); - failed.wait().unwrap(); - let worker = std::process::Command::new("/bin/sleep") - .arg("60") - .current_dir(&root) - .spawn() - .unwrap(); - let mut children = vec![("server", failed), ("worker", worker)]; - let original = stack::wait_until_answering( - &mut children, - &root, - &stack::Ready { api: 0, app: 0 }, - std::time::Duration::from_secs(1), - ) - .unwrap_err(); - let shell = Shell::default(); - let attempt = StartAttempt::begin(&shell).unwrap(); - let result = - finish_host_start(&attempt, &root, children, Err(original.clone())).unwrap_err(); - assert_eq!(result.said, original); - assert!(result.detail.is_none()); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - assert_eq!( - shell.generation.load(std::sync::atomic::Ordering::SeqCst), - 1 - ); - assert!( - !restart_host_process_with(&shell, &root, "server", 0, || panic!( - "failed attempt restarted" - )) - .unwrap() - ); - std::fs::remove_dir_all(root).unwrap(); - } - - #[cfg(unix)] - #[test] - fn stop_serializes_with_a_restart_already_inside_spawn() { - use std::sync::{atomic::Ordering::SeqCst, Arc, Barrier}; - let root = temp_root("restart-stop-barrier"); - std::fs::create_dir_all(&root).unwrap(); - let shell = Arc::new(Shell::default()); - *shell.root.lock().unwrap() = Some(root.clone()); - shell.generation.store(1, SeqCst); - let entered = Arc::new(Barrier::new(2)); - let release = Arc::new(Barrier::new(2)); - let restart = { - let (shell, root, entered, release) = ( - shell.clone(), - root.clone(), - entered.clone(), - release.clone(), - ); - std::thread::spawn(move || { - restart_host_process_with(&shell, &root, "server", 1, || { - entered.wait(); - release.wait(); - std::process::Command::new("/bin/sleep") - .arg("60") - .current_dir(&root) - .spawn() - }) - }) - }; - entered.wait(); - let (sent, completed) = std::sync::mpsc::channel(); - let stop = { - let (shell, root) = (shell.clone(), root.clone()); - std::thread::spawn(move || { - let result = - stop_everything_with(&shell, &root, stack::stop_processes_under, |_| Ok(())); - sent.send(()).unwrap(); - result - }) - }; - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - while shell.generation.load(SeqCst) == 1 { - assert!(std::time::Instant::now() < deadline); - std::thread::yield_now(); - } - assert!(matches!( - completed.try_recv(), - Err(std::sync::mpsc::TryRecvError::Empty) - )); - release.wait(); - assert!(!restart.join().unwrap().unwrap()); - stop.join().unwrap().unwrap(); - assert!(shell.children.lock().unwrap().is_empty()); - assert!(shell.root.lock().unwrap().is_none()); - assert!( - !restart_host_process_with(&shell, &root, "server", 1, || panic!( - "retired restart spawned" - )) - .unwrap() - ); - std::fs::remove_dir_all(root).unwrap(); - } - #[test] fn failed_retry_cleanup_retires_generation_and_retains_selected_root() { let app = tauri::test::mock_builder() @@ -14920,12 +13596,31 @@ fn main() { 5 ); assert_eq!(shell.root.lock().unwrap().as_ref(), Some(&root)); - assert!( - !restart_host_process_with(&shell, &root, "server", 4, || panic!( - "old generation resumed" - )) - .unwrap() - ); + } + + /// Which roots' `server` containers publish which host ports, as far as the code under test + /// can tell. Keyed by root, and every test root is its own temporary folder, so tests running + /// side by side cannot see each other's. + static PUBLISHED_BY_SERVER: Mutex)>> = Mutex::new(Vec::new()); + + pub(super) fn published_by_server(root: &Path, ports: &[u16]) -> bool { + PUBLISHED_BY_SERVER + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .any(|(published, by)| { + published == root && !ports.is_empty() && ports.iter().all(|port| by.contains(port)) + }) + } + + fn publish_from_server(root: &Path, ports: &[u16]) { + let mut published = PUBLISHED_BY_SERVER + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + published.retain(|(published, _)| published != root); + if !ports.is_empty() { + published.push((root.to_path_buf(), ports.to_vec())); + } } fn temp_root(name: &str) -> PathBuf { @@ -15051,30 +13746,26 @@ fn main() { assert!( lines.lines().any(|line| line == format!( - "{}\tcompose -f docker-compose.yml --profile harness down", + "{}\tcompose -f docker-compose.yml --profile harness --profile openbot down", root.display() )), "{lines}" ); } - /// Real loopback responder in a separate process, so root/PID ownership checks use the same - /// OS inventory as production. The Tauri mock replaces only the window, never the HTTP/PID path. + /// Real loopback responders in separate processes, so the HTTP probes are the production ones. + /// The Tauri mock replaces only the window, and `publish_from_server` only the engine's word on + /// which root's `server` container publishes the two ports. struct RestoreFixture { base: PathBuf, selected: PathBuf, owned: PathBuf, child: std::process::Child, app_child: std::process::Child, - app_descendant_pid: Option, ports: openbot_env::Ports, } impl RestoreFixture { fn new() -> Self { - Self::with_app_descendant(false) - } - - fn with_app_descendant(descendant: bool) -> Self { let base = temp_root("restore-owned-loopback"); let selected = base.join("selected"); let owned = base.join("owned"); @@ -15101,19 +13792,6 @@ fn serve(listener: TcpListener) { } } fn main() { - let args: Vec = std::env::args().collect(); - if args.get(1).map(String::as_str) == Some("--parent") { - let mut child = std::process::Command::new(std::env::current_exe().unwrap()) - .arg("--record-pid").arg(&args[2]).stdin(std::process::Stdio::null()).spawn().unwrap(); - let mut stop = [0; 1]; - let _ = std::io::stdin().read(&mut stop); - let _ = child.kill(); - let _ = child.wait(); - return; - } - if args.get(1).map(String::as_str) == Some("--record-pid") { - std::fs::write(&args[2], std::process::id().to_string()).unwrap(); - } let api = TcpListener::bind("127.0.0.1:0").unwrap(); let app = TcpListener::bind("127.0.0.1:0").unwrap(); println!("{} {}", api.local_addr().unwrap().port(), app.local_addr().unwrap().port()); @@ -15143,13 +13821,7 @@ fn main() { .split_whitespace() .map(|n| n.parse().unwrap()) .collect(); - let mut app_command = std::process::Command::new(&binary); - let descendant_file = base.join("app-listener.pid"); - if descendant { - app_command.arg("--parent").arg(&descendant_file); - } - let mut app_child = app_command - .stdin(std::process::Stdio::piped()) + let mut app_child = std::process::Command::new(&binary) .current_dir(&owned) .stdout(std::process::Stdio::piped()) .spawn() @@ -15164,33 +13836,19 @@ fn main() { .split_whitespace() .map(|n| n.parse().unwrap()) .collect(); - let app_descendant_pid = descendant.then(|| { - std::fs::read_to_string(descendant_file) - .unwrap() - .parse() - .unwrap() - }); let fixture = Self { base, selected, owned, child, app_child, - app_descendant_pid, ports: openbot_env::Ports { server: numbers[0], app: app_numbers[1], ..Default::default() }, }; - stack::record_host_processes( - &fixture.owned, - &[ - ("server", fixture.child.id()), - ("app", fixture.app_child.id()), - ], - ) - .unwrap(); + publish_from_server(&fixture.owned, &[fixture.ports.server, fixture.ports.app]); fixture } @@ -15221,13 +13879,9 @@ fn main() { fn drop(&mut self) { let _ = self.child.kill(); let _ = self.child.wait(); - if self.app_descendant_pid.is_some() { - // Dropping the pipe asks the fixture launcher to kill and reap its own child. - drop(self.app_child.stdin.take()); - } else { - let _ = self.app_child.kill(); - } + let _ = self.app_child.kill(); let _ = self.app_child.wait(); + publish_from_server(&self.owned, &[]); let _ = std::fs::remove_dir_all(&self.base); } } @@ -15553,8 +14207,8 @@ fn main() { let f = RestoreFixture::new(); assert!(server_capabilities_answer(f.ports.server)); assert!(stack::app_url(f.ports.app).is_some()); - assert!(!stack::recorded_server_owns_port(&f.selected, f.ports.server).unwrap()); - assert!(stack::recorded_server_owns_port(&f.owned, f.ports.server).unwrap()); + assert!(!published_by_server(&f.selected, &[f.ports.server])); + assert!(published_by_server(&f.owned, &[f.ports.server])); for setup in ["tauri://localhost/", "http://tauri.localhost/"] { let app = f.app(&f.selected, setup); restore_window_on(app.handle(), &f.ports); @@ -15595,7 +14249,7 @@ fn main() { app.state::().selected_root.lock().unwrap().as_ref(), Some(&f.owned) ); - assert!(stack::recorded_server_owns_port(&f.owned, f.ports.server).unwrap()); + assert!(published_by_server(&f.owned, &[f.ports.server])); } #[test] @@ -15674,7 +14328,8 @@ fn main() { #[test] fn restore_window_unproven_identity_shows_setup_without_losing_selected_root() { let f = RestoreFixture::new(); - std::fs::write(f.owned.join(".logs/host-pids.json"), "not-json").unwrap(); + // Answering on both ports, but the engine does not say this root's server publishes them. + publish_from_server(&f.owned, &[]); let app = f.app(&f.owned, "tauri://localhost/"); restore_window_on(app.handle(), &f.ports); assert_eq!( @@ -15697,16 +14352,16 @@ fn main() { #[test] fn app_adoption_and_restore_refuse_foreign_app_with_owned_api_still_running() { let f = RestoreFixture::new(); - stack::record_host_processes(&f.owned, &[("server", f.child.id())]).unwrap(); - stack::record_host_processes(&f.selected, &[("app", f.app_child.id())]).unwrap(); + // This root's server publishes the API port; something else has the app port. + publish_from_server(&f.owned, &[f.ports.server]); assert!(already_running_on( &f.owned, f.ports.server, - stack::recorded_server_owns_port + |root, port| { server_container_publishes(None, root, &[port]) } )); assert!(stack::app_url(f.ports.app).is_some()); - let initial_adoption = already_running_at(&f.owned, &f.ports); let app = f.app(&f.owned, "tauri://localhost/"); + let initial_adoption = already_running_at(&app.state::(), &f.owned, &f.ports); let shown = show_openbot_on(app.handle().clone(), &f.ports); restore_window_on(app.handle(), &f.ports); let destination = app.get_webview_window("main").unwrap().url().unwrap(); @@ -15715,25 +14370,23 @@ fn main() { } #[test] - fn app_adoption_and_restore_allow_owned_app_direct_and_launcher_descendant() { - for descendant in [false, true] { - let f = RestoreFixture::with_app_descendant(descendant); - assert_ne!(f.child.id(), f.app_child.id()); - if descendant { - assert_ne!(f.app_descendant_pid.unwrap(), f.app_child.id()); - } - assert!(already_running_at(&f.owned, &f.ports)); - let app = f.app(&f.owned, "tauri://localhost/"); - show_openbot_on(app.handle().clone(), &f.ports).unwrap(); - restore_window_on(app.handle(), &f.ports); - assert_eq!( - app.get_webview_window("main") - .unwrap() - .url() - .unwrap() - .as_str(), - format!("http://127.0.0.1:{}/", f.ports.app) - ); - } + fn app_adoption_and_restore_allow_the_app_its_own_server_container_publishes() { + let f = RestoreFixture::new(); + let app = f.app(&f.owned, "tauri://localhost/"); + assert!(already_running_at( + &app.state::(), + &f.owned, + &f.ports + )); + show_openbot_on(app.handle().clone(), &f.ports).unwrap(); + restore_window_on(app.handle(), &f.ports); + assert_eq!( + app.get_webview_window("main") + .unwrap() + .url() + .unwrap() + .as_str(), + format!("http://127.0.0.1:{}/", f.ports.app) + ); } } diff --git a/desktop/src-tauri/src/preparation.rs b/desktop/src-tauri/src/preparation.rs index df7fa6290c..6e8f84e3bf 100644 --- a/desktop/src-tauri/src/preparation.rs +++ b/desktop/src-tauri/src/preparation.rs @@ -475,11 +475,16 @@ fn complete_with( } /// `selection=None` is the auth check, which has no authority to change the Bot selection. +/// +/// No runtime and no packages: OpenBot's server and worker are containers now, so what Start needs +/// from the installation is the deployment it laid out and the images it downloaded. Installation +/// still records the host packages, for the self-hosted Intelligence sign-in that runs on the host, +/// and that sign-in checks them itself. pub fn require( root: &Path, selection: Option<&Option>, address: &Address, -) -> Result { +) -> Result<(), Problem> { require_with(root, selection, |images| require_images(address, images)) } @@ -487,7 +492,7 @@ fn require_with( root: &Path, selection: Option<&Option>, verify: impl FnOnce(&[String]) -> Result<(), Problem>, -) -> Result { +) -> Result<(), Problem> { let record: Prepared = serde_json::from_slice( &std::fs::read(root.join(FILE)).map_err(|e| required(e.to_string()))?, ) @@ -511,9 +516,7 @@ fn require_with( "The deployment was laid out by an earlier OpenBot and is missing files this version runs.", )); } - let bun = dependencies_ready(root)?; - verify(&record.images)?; - Ok(bun) + verify(&record.images) } /// Public Compose overrides suffice to identify and download software before authentication. @@ -733,7 +736,7 @@ fn main() { } #[test] - fn changed_bot_lockfile_or_deleted_package_requires_installation() { + fn changed_bot_requires_installation_but_host_packages_do_not() { let f = Fixture::new(); let chosen = Some(HarnessChoice { id: "langgraph".into(), @@ -745,11 +748,14 @@ fn main() { "wrong selection must stop first" )) .is_err()); + // Start runs containers, so the host's packages and runtime are not its business. std::fs::remove_file(f.root.join("node_modules/synthetic-package/package.json")).unwrap(); - assert!(require_with(&f.root, Some(&chosen), |_| Ok(())).is_err()); - f.dependencies(); - f.prepared(chosen.as_ref()); - std::fs::write(f.root.join("bun.lock"), "changed-lock").unwrap(); + std::fs::write(f.root.join("runtime-version"), "1.2.15").unwrap(); + require_with(&f.root, Some(&chosen), |_| Ok(())).unwrap(); + // The images still are. + assert!(require_with(&f.root, Some(&chosen), |_| Err(required("missing image"))).is_err()); + // And the deployment Compose runs. + std::fs::write(f.root.join("docker-compose.yml"), "services: {x: {}}\n").unwrap(); assert!(require_with(&f.root, Some(&chosen), |_| Ok(())).is_err()); } diff --git a/desktop/src-tauri/src/stack.rs b/desktop/src-tauri/src/stack.rs index 6758b4b997..43b66a361c 100644 --- a/desktop/src-tauri/src/stack.rs +++ b/desktop/src-tauri/src/stack.rs @@ -1715,6 +1715,46 @@ fn configure_local_browser( Ok(()) } +/// The last build error can precede Bun's wrapper stack, version, and exit message. +/// Bound both the file read and displayed lines while preserving that useful context. +fn tail_of(logs: &Path, name: &str) -> String { + use std::io::{Read, Seek, SeekFrom}; + const MAX_BYTES: u64 = 8 * 1024; + let read = || -> std::io::Result> { + let mut file = std::fs::File::open(logs.join(format!("{name}.log")))?; + let offset = file.metadata()?.len().saturating_sub(MAX_BYTES); + file.seek(SeekFrom::Start(offset))?; + let mut bytes = Vec::new(); + file.take(MAX_BYTES).read_to_end(&mut bytes)?; + // Do not show a truncated first line, which may include part of a credential. + if offset > 0 { + let first_line = bytes.iter().position(|byte| *byte == b'\n'); + bytes.drain(..first_line.map_or(bytes.len(), |index| index + 1)); + } + Ok(bytes) + }; + let bytes = match read() { + Ok(bytes) => bytes, + Err(error) => return format!("Could not read {name}.log: {error}"), + }; + let text = String::from_utf8_lossy(&bytes); + let tail: Vec<&str> = text + .lines() + .filter(|line| !line.trim().is_empty()) + .rev() + .take(40) + .collect(); + if tail.is_empty() { + return format!("No complete lines were available in the tail of {name}.log."); + } + let mut lines = tail; + lines.reverse(); + format!( + "Last from {name}.log (up to 40 lines, 8 KiB):\n{}", + lines.join("\n") + ) +} + /// Keep a replacement handle even if refreshing durable ownership fails. The caller must /// report success only after this Result succeeds; Stop still has the handle on failure. #[cfg(unix)] @@ -3320,6 +3360,93 @@ pub fn ports_we_already_publish(engine: &Address, root: &Path) -> std::collectio published_in(&listing) } +/** +Whether this deployment's own `server` container publishes every one of `ports` on the host. + +THIS IS WHAT OWNERSHIP MEANS NOW. The server and the app used to be host processes, and a port +belonged to this deployment when a process recorded in its `host-pids.json` was listening on it. +They are one container now, and the listener on the host is the engine's port forwarder, which no +pid file names. What does name it is Compose: this project's `server` row lists the host ports it +was given. Something else answering on the same port, another installation's server or a stranger, +is not in that row. + +Every service rather than `server` alone, read row by row: naming a service on `ps` makes Compose +resolve it, and `server` sits behind a profile this command would otherwise have to repeat. +*/ +pub fn server_publishes(engine: &Address, root: &Path, ports: &[u16]) -> Result { + let operation = format!("{} compose ps", engine.engine.binary()); + let output = compose_command(engine, root, &Secrets::new()) + .args(["ps", "--format", "{{.Service}}\t{{.Ports}}"]) + .output() + .map_err(|error| { + Problem::with( + "OpenBot could not inspect its Compose services.", + format!("could not run {operation}: {error}"), + ) + })?; + if !output.status.success() { + return Err(Problem::with( + "OpenBot could not inspect its Compose services.", + format!("{operation}: {}", command_said(&output.stderr)), + )); + } + Ok(server_publishes_in( + &String::from_utf8_lossy(&output.stdout), + ports, + )) +} + +/// `server_publishes` over a `{{.Service}}\t{{.Ports}}` listing. Pure, for the reason +/// `published_in` is. +pub fn server_publishes_in(listing: &str, ports: &[u16]) -> bool { + let published: std::collections::HashSet = listing + .lines() + .filter_map(|row| row.split_once('\t')) + .filter(|(service, _)| service.trim() == "server") + .flat_map(|(_, mappings)| published_in(mappings)) + .collect(); + !ports.is_empty() && ports.iter().all(|port| published.contains(port)) +} + +/** +Restart OpenBot itself, the server and the worker, and leave everything else running. + +What the Status window's Restart did to the host processes, now done by the engine that runs them: +the database and the Bots stay up, so nothing is lost and it takes seconds. The profile is named +because both services sit behind it, and Compose refuses to restart a service it cannot see. +*/ +pub fn restart_openbot(engine: &Address, root: &Path) -> Result<(), String> { + let output = compose_command(engine, root, &Secrets::new()) + .args(["--profile", OPENBOT_PROFILE, "restart", "server", "worker"]) + .output() + .map_err(|error| format!("could not restart OpenBot: {error}"))?; + if output.status.success() { + return Ok(()); + } + Err(command_said(&output.stderr)) +} + +/** +The name Compose gives this deployment's project, and so the prefix of its networks. + +The desktop runs Compose from the deployment folder with no `-p` and the file names no `name:`, so +the project is the folder's own name as Compose normalises it: lower case, with everything but +`a-z`, `0-9`, `_` and `-` dropped, then any leading `_` or `-` trimmed. Worked out here rather than +asked of Compose, because the answer is wanted before `up`, to tell the supervisor which network to +put each Bot's computer on, and it has to match what `up` is about to create. +*/ +pub fn project_name(root: &Path) -> String { + let base = root + .file_name() + .map(|name| name.to_string_lossy().to_lowercase()) + .unwrap_or_default(); + base.chars() + .filter(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == '_' || *c == '-') + .collect::() + .trim_start_matches(['_', '-']) + .to_string() +} + /** The published ports in a `compose ps` listing. @@ -3406,21 +3533,11 @@ pub fn port_already_taken_except( None } -/// Wait until the API answers, or say why it never did. -/// -/// Spawning is not starting. Each of these three can exit in the first second for a reason that has -/// nothing to do with the others, and a shell that reports "running" because it called `spawn` -/// three times is telling somebody the stack is up while nothing is listening. That is worse than -/// an error, because the next thing they do is open a page that will not load and go looking for -/// the fault in the wrong place. -/// -/// So: watch the child, and watch the port. Whichever fails first is what gets reported, with the -/// tail of the log that explains it. -/// The two things that have to answer before anybody is told the stack is up. +/// The two addresses that have to answer before anybody is told the stack is up. /// /// The API alone is not enough. The window navigates to the app, so a person told "running" who -/// then gets a blank window has been told something that is not true, and the API was answering the -/// whole time. +/// then gets a blank window has been told something that is not true. Both are the one `server` +/// container now, published on two host ports, and either mapping can be the one that is missing. pub struct Ready { pub api: u16, pub app: u16, @@ -3463,33 +3580,28 @@ pub fn app_url(port: u16) -> Option { answering_at(port, "/") } -/// Wait until the stack is genuinely usable, or say which part is not. +/// Wait until the server container answers its health check on both host ports, or say which +/// one never did. /// -/// Watches the children as well as the ports, because three processes that died leave a port -/// unanswered for the same length of time as three that are still starting, and only one of those -/// is worth waiting out. -pub fn wait_until_answering( - children: &mut [(&'static str, std::process::Child)], - logs: &Path, +/// `compose up` returns once it has asked for the container, not once the server inside it is +/// listening, and a server still running its first start leaves the port unanswered for the same +/// length of time as one that is crash-looping under its restart policy. Only an answer settles it. +/// +/// `keep_going` is asked between polls, so a Stop pressed during the wait ends it there rather than +/// after the full patience. What a stopped wait returns does not matter: the caller asks its own +/// ticket first, and the cancellation is what it reports. +pub fn wait_until_healthy( ready: &Ready, patience: std::time::Duration, + keep_going: impl Fn() -> bool, ) -> Result<(), String> { let deadline = std::time::Instant::now() + patience; let mut api_up = false; - while std::time::Instant::now() < deadline { - for (name, child) in children.iter_mut() { - if let Ok(Some(status)) = child.try_wait() { - return Err(format!( - "{name} stopped straight away ({status}). {}", - tail_of(logs, name) - )); - } - } - + while std::time::Instant::now() < deadline && keep_going() { // An earlier API success is no longer sufficient when the app becomes ready later. - api_up = answering_at(ready.api, "/api/capabilities").is_some(); - if api_up && app_url(ready.app).is_some() { + api_up = answering_at(ready.api, "/health").is_some(); + if api_up && answering_at(ready.app, "/health").is_some() { return Ok(()); } @@ -3498,56 +3610,11 @@ pub fn wait_until_answering( if api_up { return Err(format!( - "the API is answering, but the app is not answering on port {}. {}", - ready.app, - tail_of(logs, "app") + "the API is answering, but the app is not answering on port {}.", + ready.app )); } - Err(format!( - "the API is not answering on port {}. {}", - ready.api, - tail_of(logs, "server") - )) -} - -/// The last build error can precede Bun's wrapper stack, version, and exit message. -/// Bound both the file read and displayed lines while preserving that useful context. -fn tail_of(logs: &Path, name: &str) -> String { - use std::io::{Read, Seek, SeekFrom}; - const MAX_BYTES: u64 = 8 * 1024; - let read = || -> std::io::Result> { - let mut file = std::fs::File::open(logs.join(format!("{name}.log")))?; - let offset = file.metadata()?.len().saturating_sub(MAX_BYTES); - file.seek(SeekFrom::Start(offset))?; - let mut bytes = Vec::new(); - file.take(MAX_BYTES).read_to_end(&mut bytes)?; - // Do not show a truncated first line, which may include part of a credential. - if offset > 0 { - let first_line = bytes.iter().position(|byte| *byte == b'\n'); - bytes.drain(..first_line.map_or(bytes.len(), |index| index + 1)); - } - Ok(bytes) - }; - let bytes = match read() { - Ok(bytes) => bytes, - Err(error) => return format!("Could not read {name}.log: {error}"), - }; - let text = String::from_utf8_lossy(&bytes); - let tail: Vec<&str> = text - .lines() - .filter(|line| !line.trim().is_empty()) - .rev() - .take(40) - .collect(); - if tail.is_empty() { - return format!("No complete lines were available in the tail of {name}.log."); - } - let mut lines = tail; - lines.reverse(); - format!( - "Last from {name}.log (up to 40 lines, 8 KiB):\n{}", - lines.join("\n") - ) + Err(format!("the API is not answering on port {}.", ready.api)) } /// Keep a short startup headline and useful local diagnostics, with credentials removed. @@ -3690,29 +3757,6 @@ mod tests { use super::*; use crate::test_support::temp_root; - #[test] - fn startup_log_tail_retains_build_error_before_wrapper_without_loading_whole_log() { - let root = temp_root("startup-build-diagnostic"); - std::fs::create_dir_all(&root).unwrap(); - let mut log = "old output must be omitted\n".repeat(1_000); - log.push_str("error during build: Could not resolve imported module\n"); - log.push_str(&" at synthetic build frame\n".repeat(20)); - log.push_str(" at run (app/scripts/serve-or-build.ts:36:11)\n"); - log.push_str("Bun v1.2.15 (macOS arm64)\nerror: script serve exited with code 1\n"); - std::fs::write(root.join("app.log"), log).unwrap(); - - let detail = tail_of(&root, "app"); - assert!( - detail.contains("Could not resolve imported module"), - "{detail}" - ); - assert!(detail.contains("serve-or-build.ts:36:11"), "{detail}"); - assert!(!detail.contains(&"old output must be omitted\n".repeat(20))); - assert!(detail.len() < 8_500, "unbounded startup diagnostic"); - assert!(detail.lines().count() <= 42, "too many startup log lines"); - std::fs::remove_dir_all(root).unwrap(); - } - #[test] fn startup_problem_keeps_build_and_cleanup_details_but_removes_credentials() { let secrets = Secrets::from([ @@ -8484,7 +8528,14 @@ fn main() { let services = up(&podman, &root, true, BundledBots::none(), &Secrets::new()).unwrap(); assert_eq!( services, - ["postgres", "supervisor", "agent-computer", "agent-harness"] + [ + "postgres", + "supervisor", + "agent-computer", + "server", + "worker", + "agent-harness" + ] ); migrate(&podman, &root, &Secrets::new()).unwrap(); let log = std::fs::read_to_string(&record).unwrap(); @@ -8638,7 +8689,9 @@ fn main() { .unwrap(); let snapshot = log.find("ps --quiet").unwrap(); let computers = log.find("stop current late restarted").unwrap(); - let teardown = log.find("--profile harness down").unwrap(); + let teardown = log + .find("--profile harness --profile openbot down") + .unwrap(); assert!( config < supervisor && supervisor < snapshot @@ -8717,7 +8770,7 @@ fn main() { ); assert!( log.contains(&format!( - "{prefix}compose -f docker-compose.yml --profile harness down" + "{prefix}compose -f docker-compose.yml --profile harness --profile openbot down" )), "{log}" ); @@ -8935,7 +8988,7 @@ fn main() { } fn wait(&mut self, patience: std::time::Duration) -> Result<(), String> { - wait_until_answering(&mut self.children, &self.root, &self.ready, patience) + wait_until_healthy(&self.ready, patience, || true) } fn requests(&self, name: &str) -> String { @@ -8973,31 +9026,30 @@ fn main() { .children .iter_mut() .all(|(_, child)| child.try_wait().unwrap().is_none()); - let api_now = answering_at(fixture.ready.api, "/api/capabilities"); - let app_now = app_url(fixture.ready.app); + let api_now = answering_at(fixture.ready.api, "/health"); + let app_now = answering_at(fixture.ready.app, "/health"); eprintln!("current API readiness: result={result:?}; children_alive={live}; api_requests={api_requests:?}; api_now={api_now:?}; app_now={app_now:?}"); drop(fixture); assert!(live && api_now.is_none() && app_now.is_some()); let error = result.expect_err("a historical API success cannot satisfy readiness"); assert!(api_requests.lines().count() >= 2, "{api_requests}"); assert!(error.contains("the API is not answering"), "{error}"); - assert!(error.contains("synthetic server diagnostic"), "{error}"); } #[test] - fn readiness_accepts_currently_healthy_api_and_app() { + fn readiness_asks_the_server_health_check_on_both_ports() { let mut fixture = ReadinessFixture::new("200", "200"); let result = fixture.wait(std::time::Duration::from_secs(2)); let api_requests = fixture.requests("server"); let app_requests = fixture.requests("app"); drop(fixture); assert!(result.is_ok(), "{result:?}"); - assert!(api_requests.contains("200 GET /api/capabilities HTTP/1.1")); - assert!(app_requests.contains("200 GET / HTTP/1.1")); + assert!(api_requests.contains("200 GET /health HTTP/1.1")); + assert!(app_requests.contains("200 GET /health HTTP/1.1")); } #[test] - fn readiness_timeout_names_the_currently_unavailable_service() { + fn readiness_timeout_names_the_currently_unavailable_port() { for (api, app, missing) in [("503", "200", "server"), ("200", "503", "app")] { let mut fixture = ReadinessFixture::new(api, app); let result = fixture.wait(std::time::Duration::from_millis(100)); @@ -9013,26 +9065,52 @@ fn main() { error.contains(&format!("the {service} is not answering on port {port}")), "{error}" ); - assert!( - error.contains(&format!("synthetic {missing} diagnostic")), - "{error}" - ); } } #[test] - fn readiness_reports_child_exit_without_waiting_for_timeout() { - let mut fixture = ReadinessFixture::new("exit", "200"); + fn readiness_ends_at_once_when_the_start_is_stopped() { + // Port 1 needs privilege to bind, so nothing here will ever answer it. + let ready = Ready { api: 1, app: 1 }; let started = std::time::Instant::now(); - let result = fixture.wait(std::time::Duration::from_secs(30)); - let elapsed = started.elapsed(); - let status = fixture.children[0].1.try_wait().unwrap().unwrap(); - drop(fixture); - let error = result.unwrap_err(); - assert!(elapsed < std::time::Duration::from_secs(5), "{elapsed:?}"); - assert_eq!(status.code(), Some(17)); - assert!(error.contains("server stopped straight away"), "{error}"); - assert!(error.contains("synthetic server diagnostic"), "{error}"); + let result = wait_until_healthy(&ready, std::time::Duration::from_secs(30), || false); + assert!(result.is_err()); + assert!( + started.elapsed() < std::time::Duration::from_secs(5), + "a stopped Start sat out the readiness wait" + ); + } + + #[test] + fn project_name_is_the_folder_name_as_compose_normalises_it() { + assert_eq!(project_name(Path::new("/Users/someone/OpenBot")), "openbot"); + assert_eq!( + project_name(Path::new("/home/someone/My OpenBot!")), + "myopenbot" + ); + assert_eq!(project_name(Path::new("/srv/_x")), "x"); + assert_eq!(project_name(Path::new("/srv/-_open.bot_2-")), "openbot_2-"); + assert_eq!( + project_name(Path::new("C:/Users/someone/ÖpenBot")), + "penbot" + ); + } + + #[test] + fn only_the_server_row_can_own_the_app_and_api_ports() { + let listing = "postgres\t127.0.0.1:5432->5432/tcp\n\ + server\t127.0.0.1:3010->3001/tcp, [::1]:3010->3001/tcp, 127.0.0.1:3001->3001/tcp, [::1]:3001->3001/tcp\n\ + worker\t\n"; + assert!(server_publishes_in(listing, &[3001, 3010])); + assert!(server_publishes_in(listing, &[3001])); + // Published by the project, but not by the server: still not the app. + assert!(!server_publishes_in(listing, &[3001, 5432])); + assert!(!server_publishes_in(listing, &[3002])); + assert!(!server_publishes_in("", &[3001])); + assert!( + !server_publishes_in(listing, &[]), + "nothing asked is nothing proven" + ); } #[test] diff --git a/desktop/src-tauri/src/stop_ipc_tests.rs b/desktop/src-tauri/src/stop_ipc_tests.rs index e66a747645..017d6a9793 100644 --- a/desktop/src-tauri/src/stop_ipc_tests.rs +++ b/desktop/src-tauri/src/stop_ipc_tests.rs @@ -185,7 +185,8 @@ mod stop_ipc { let fixture = Fixture::new(); let compose = fixture.compose_barrier(); let shell = fixture.app.state::(); - let cleanup = shell.children.lock().unwrap(); + // The first lock host cleanup takes; held here, it keeps that cleanup waiting. + let cleanup = shell.host_access.lock().unwrap(); let dispatch = fixture.dispatch(); wait_until(|| shell.generation.load(SeqCst) == 1); // Always release both barriers before asserting against the old synchronous command. @@ -278,19 +279,10 @@ mod stop_ipc { pending, "Stop finished while Start still held the side-effect lock" ); - let error = tauri::async_runtime::block_on(start_host_processes( - &attempt, - &fixture.root, - &fixture.root.join(".logs"), - Path::new("must-not-launch"), - &stack::Secrets::new(), - None, - |_| panic!("cancelled Start published a child"), - |_| panic!("cancelled Start reached readiness"), - )) - .unwrap_err(); + let error = attempt + .lock_current() + .expect_err("a cancelled Start resumed a side effect"); assert_eq!(error.said, StartAttempt::cancelled().said); - assert!(shell.children.lock().unwrap().is_empty()); assert!(shell.root.lock().unwrap().is_none()); } @@ -394,7 +386,7 @@ mod stop_ipc { let shell = fixture.app.state::(); // A poisoned cleanup lock is a real panic boundary; it must reject IPC, not abandon it. std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let _cleanup = shell.children.lock().unwrap(); + let _cleanup = shell.host_access.lock().unwrap(); panic!("synthetic cleanup lock poisoning"); })) .expect_err("poison the cleanup lock"); diff --git a/desktop/src-tauri/tests/fixtures/engine.rs b/desktop/src-tauri/tests/fixtures/engine.rs index 4f8bf8c90e..ab78c91875 100644 --- a/desktop/src-tauri/tests/fixtures/engine.rs +++ b/desktop/src-tauri/tests/fixtures/engine.rs @@ -94,7 +94,7 @@ fn main() { if SCENARIO == "stop-ipc" && (joined.starts_with("ps ") || joined.contains("stop supervisor")) { return; } - if SCENARIO == "stop-ipc" && joined == "compose -f docker-compose.yml --profile harness down" { + if SCENARIO == "stop-ipc" && joined == "compose -f docker-compose.yml --profile harness --profile openbot down" { let mut barrier = std::net::TcpStream::connect( std::fs::read_to_string("stop-barrier-address") .unwrap() @@ -137,7 +137,14 @@ fn main() { value if value.starts_with("compose up -d --no-build --pull never ") || value - .starts_with("compose --profile harness up -d --no-build --pull never ") => {} + .starts_with("compose --profile harness up -d --no-build --pull never ") => + { + // Whether Compose was handed the desktop's host token, by its length only: the + // token itself is never written anywhere. + if let Some(token) = std::env::var_os("OPENBOT_DESKTOP_HOST_TOKEN") { + std::fs::write("host-token-at-up", token.len().to_string()).unwrap(); + } + } "compose run --rm --pull never migrate" => { if SCENARIO == "harness" { eprintln!("synthetic migration barrier"); diff --git a/desktop/src/StatusWindow.tsx b/desktop/src/StatusWindow.tsx index f568d15a93..ed04bb3318 100644 --- a/desktop/src/StatusWindow.tsx +++ b/desktop/src/StatusWindow.tsx @@ -97,8 +97,8 @@ const BUSY_SAID: Record = { /** * What OpenBot is running, what each part costs, and Restart and Stop for each. * - * Memory is read every few seconds while the window shows; it is Docker's figure per container and - * the resident size of OpenBot's own processes. Intelligence's readiness is the CLI's, asked only on + * Memory is read every few seconds while the window shows; it is Docker's figure per container, + * OpenBot's own server and worker among them. Intelligence's readiness is the CLI's, asked only on * opening, on Refresh and after an action here: the main window's own Intelligence calls take the * same lock, and a status window polling it would make them fail as busy. */ @@ -211,7 +211,7 @@ export function StatusWindow({ readEveryMs = 5000 }: { readEveryMs?: number }) { {status.openbotRunning && (
- {/* Restart is the supervisor bringing back processes this session holds. */} + {/* Restart is Compose restarting the server and worker containers this session started. */} {status.openbotHeld && (