diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index afd119c84d..4313acb2e4 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1099,6 +1099,7 @@ dependencies = [ "mesh-llm-sdk", "mesh-llm-system", "neteq", + "nix 0.31.3", "nostr", "notify-rust", "objc2", diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index 1ba814da47..a5ea2f2217 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -36,6 +36,7 @@ tauri-build = { version = "2", features = [] } [target.'cfg(unix)'.dependencies] libc = "0.2" +nix = { version = "0.31", default-features = false, features = ["signal", "process"] } ctrlc = { version = "3", features = ["termination"] } [target.'cfg(target_os = "linux")'.dependencies] @@ -62,7 +63,7 @@ user-idle = { version = "0.6", default-features = false } plist = "1" [target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = ["Win32_Security", "Win32_Storage_FileSystem", "Win32_System_JobObjects", "Win32_System_Registry", "Win32_System_Threading", "Win32_Foundation"] } +windows-sys = { version = "0.61", features = ["Win32_Security", "Win32_Storage_FileSystem", "Win32_System_Diagnostics_ToolHelp", "Win32_System_JobObjects", "Win32_System_Registry", "Win32_System_Threading", "Win32_Foundation"] } keyring = { version = "3.6.3", default-features = false, features = ["windows-native", "vendored"], optional = true } user-idle = { version = "0.6", default-features = false } diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index fc90e6ab14..97736f85d7 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, io::Write, sync::{ atomic::{AtomicBool, AtomicU16, AtomicU8}, @@ -52,6 +52,10 @@ pub struct AppState { pub managed_agents_store_lock: Mutex<()>, pub channel_templates_store_lock: Mutex<()>, pub managed_agent_processes: Mutex>, + /// Per-pair intent tokens spanning the unlocked process-preparation/spawn + /// phase. The mutex is held only to insert/remove a key; no filesystem, + /// readiness, or child-process work may run while it is held. + pub managed_agent_start_reservations: Mutex>, pub huddle_state: Mutex, pub huddle_audio: crate::huddle::tts_settings::HuddleAudioSettingsState, /// Tauri app handle — stored after setup so huddle commands can emit @@ -215,6 +219,7 @@ pub fn build_app_state() -> AppState { managed_agents_store_lock: Mutex::new(()), channel_templates_store_lock: Mutex::new(()), managed_agent_processes: Mutex::new(HashMap::new()), + managed_agent_start_reservations: Mutex::new(HashSet::new()), session_config_cache: Mutex::new(HashMap::new()), huddle_state: Mutex::new(HuddleState::default()), huddle_audio: Default::default(), diff --git a/desktop/src-tauri/src/commands/agent_discovery/install_exec.rs b/desktop/src-tauri/src/commands/agent_discovery/install_exec.rs index 3b94f2ef8a..18216809e0 100644 --- a/desktop/src-tauri/src/commands/agent_discovery/install_exec.rs +++ b/desktop/src-tauri/src/commands/agent_discovery/install_exec.rs @@ -235,6 +235,8 @@ fn await_install_child( // Save the PID before moving `child` into the wait thread so we can // kill the process on timeout. let child_pid = child.id(); + #[cfg(windows)] + let child_identity = crate::managed_agents::child_process_identity(&child).ok(); std::thread::spawn(move || { let _ = events_tx.send(Settled::Exited(child.wait())); @@ -257,6 +259,11 @@ fn await_install_child( // was still running and this is a timeout — the status the kill produces // moments later describes the kill, not the install, so it is discarded. let install_finished = settle.status.is_some(); + #[cfg(unix)] + terminate_install_group(child_pid); + #[cfg(windows)] + terminate_install_group(child_pid, child_identity); + #[cfg(not(any(unix, windows)))] terminate_install_group(child_pid); // Reaping the child and finishing the drains share one bound. Both // normally complete within microseconds of the kill, which closes the @@ -427,14 +434,21 @@ fn signal_reaches(target: i32) -> bool { std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH) } -/// Windows has no process groups on this path: `terminate_process` runs -/// `taskkill /T /F`, which is already tree-wide and unconditional, so there is -/// no escalation to get wrong. -#[cfg(not(unix))] -fn terminate_install_group(pid: u32) { - let _ = crate::managed_agents::terminate_process(pid); +/// Windows binds cleanup to the creation identity captured from the spawned +/// child's stable handle. If metadata could not be read, teardown fails closed +/// rather than acting on a potentially recycled PID. +#[cfg(windows)] +fn terminate_install_group(pid: u32, process_identity: Option) { + if let Err(error) = + crate::managed_agents::terminate_process_with_identity(pid, process_identity) + { + eprintln!("buzz-desktop: bounded Windows installer cleanup failed: {error}"); + } } +#[cfg(not(any(unix, windows)))] +fn terminate_install_group(_pid: u32) {} + /// A failure carrying whatever the drains captured, with `reason` leading /// stderr so the surfaced message names the failure before the install's own /// output. diff --git a/desktop/src-tauri/src/commands/agent_discovery/managed_node.rs b/desktop/src-tauri/src/commands/agent_discovery/managed_node.rs index fbfb068c0e..2291e86e60 100644 --- a/desktop/src-tauri/src/commands/agent_discovery/managed_node.rs +++ b/desktop/src-tauri/src/commands/agent_discovery/managed_node.rs @@ -121,8 +121,9 @@ pub(super) fn managed_node_runtime_ready() -> bool { /// /// Cleanup: the child runs in its own process group on Unix (`process_group(0)`) /// so an unconditional group SIGKILL on every exit path terminates all -/// descendants. On Windows, `terminate_process` issues `taskkill /T /F` for -/// tree-wide cleanup. SIGKILL to an already-dead group returns ESRCH (no-op). +/// descendants. On Windows, it is spawned suspended, assigned to a dedicated +/// kill-on-close Job Object, and only then resumed. SIGKILL to an already-dead +/// Unix group returns ESRCH (no-op). pub(super) fn probe_node( executable: &std::path::Path, expected_version: &str, @@ -148,8 +149,15 @@ pub(super) fn probe_node( use std::os::unix::process::CommandExt; cmd.process_group(0); } - let Ok(mut child) = cmd.spawn() else { - return false; + #[cfg(windows)] + let (mut child, mut probe_job) = match crate::managed_agents::spawn_probe_in_job(&mut cmd) { + Ok(contained) => contained, + Err(_) => return false, + }; + #[cfg(not(windows))] + let mut child = match cmd.spawn() { + Ok(child) => child, + Err(_) => return false, }; let deadline = std::time::Instant::now() + timeout; @@ -158,14 +166,22 @@ pub(super) fn probe_node( Ok(Some(status)) => break status, Ok(None) => { if std::time::Instant::now() >= deadline { - kill_probe_group(child.id()); + let _ = terminate_probe_group( + &mut child, + #[cfg(windows)] + &mut probe_job, + ); let _ = child.wait(); return false; } std::thread::sleep(Duration::from_millis(50)); } Err(_) => { - kill_probe_group(child.id()); + let _ = terminate_probe_group( + &mut child, + #[cfg(windows)] + &mut probe_job, + ); let _ = child.wait(); return false; } @@ -173,7 +189,13 @@ pub(super) fn probe_node( }; // Group-kill unconditionally: SIGKILL to a dead group is ESRCH (no-op). - kill_probe_group(child.id()); + if !terminate_probe_group( + &mut child, + #[cfg(windows)] + &mut probe_job, + ) { + return false; + } if !exit_status.success() { return false; @@ -188,18 +210,31 @@ pub(super) fn probe_node( /// Kill the probe's process group/tree unconditionally (no TERM grace — this /// is a probe, not an agent session). ESRCH on a dead group is fine. -fn kill_probe_group(pid: u32) { +fn terminate_probe_group( + child: &mut std::process::Child, + #[cfg(windows)] probe_job: &mut crate::managed_agents::JobHandle, +) -> bool { #[cfg(unix)] unsafe { - libc::kill(-(pid as i32), libc::SIGKILL); + libc::kill(-(child.id() as i32), libc::SIGKILL); + true } #[cfg(windows)] { - let _ = crate::managed_agents::terminate_process(pid); + let cleaned = match probe_job.terminate_and_wait(Duration::from_secs(1)) { + Ok(()) => true, + Err(error) => { + eprintln!("buzz-desktop: managed Node readiness cleanup failed: {error}"); + false + } + }; + let _ = child.kill(); + cleaned } #[cfg(not(any(unix, windows)))] { - let _ = pid; + let _ = child; + true } } diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index 3b114b0474..99de2ccc69 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -7,8 +7,8 @@ use crate::{ build_managed_agent_summary, current_instance_id, discover_provider_candidates, ensure_persona_is_active, find_managed_agent_mut, load_managed_agents, load_personas, load_teams, managed_agent_avatar_url, normalize_agent_args, provider_deploy, - resolve_provider_binary, save_managed_agents, start_managed_agent_process, - stop_managed_agent_process, stop_managed_agent_workspace_pair, + reserve_managed_agent_start, resolve_provider_binary, save_managed_agents, + start_managed_agent_process, stop_managed_agent_process, stop_managed_agent_workspace_pair, sync_managed_agent_processes, try_regenerate_nest, validate_provider_config, BackendKind, CreateManagedAgentRequest, CreateManagedAgentResponse, ManagedAgentRecord, ManagedAgentSummary, RelayMeshConfig, DEFAULT_ACP_COMMAND, DEFAULT_AGENT_PARALLELISM, @@ -395,56 +395,169 @@ pub(super) async fn start_local_agent_with_preflight( ); ensure_relay_mesh_for_record(app, mesh_model_id.as_deref(), allow_fresh_create_start).await?; - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let mut records = load_managed_agents(app)?; - let mut runtimes = state - .managed_agent_processes - .lock() - .map_err(|e| e.to_string())?; - let record = find_managed_agent_mut(&mut records, pubkey)?; - if record.backend != BackendKind::Local { - return Err(format!("agent {pubkey} is no longer a local agent")); - } - // Re-snapshot the persona onto the record at every spawn so the agent always - // starts with the current persona config (system_prompt, model, provider, - // runtime). This clears the "out of date" drift badge without requiring a - // delete+recreate. See `apply_persona_snapshot` for the precedence and - // env-override self-heal rules. - // Load personas once: used for snapshot application below and summary build - // at the end — avoids a second disk read for the same file in the same call. + // Phase A: snapshot the record and reserve its workspace pair. No lock is + // carried into command discovery, readiness, or process spawning. let personas = load_personas(app).unwrap_or_default(); - if let Some(persona_id) = record.persona_id.clone() { - match personas.iter().find(|p| p.id == persona_id) { - Some(persona) => { - crate::managed_agents::persona_events::apply_persona_snapshot(record, persona); - record.updated_at = crate::util::now_iso(); - } - None => { - return Err( - crate::managed_agents::effective_config::ORPHANED_INSTANCE_ERROR.to_string(), - ); + let workspace_relay = crate::relay::relay_ws_url_with_override(state); + let (mut staged_record, base_updated_at, key, _reservation) = { + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + if state + .shutdown_started + .load(std::sync::atomic::Ordering::Acquire) + { + return Err("desktop shutdown has started".into()); + } + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(app)?; + let record = records + .iter() + .find(|record| record.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + if record.backend != BackendKind::Local { + return Err(format!("agent {pubkey} is no longer a local agent")); + } + let mut staged = record.clone(); + let base_updated_at = record.updated_at.clone(); + if let Some(persona_id) = staged.persona_id.clone() { + match personas.iter().find(|p| p.id == persona_id) { + Some(persona) => { + crate::managed_agents::persona_events::apply_persona_snapshot( + &mut staged, + persona, + ); + staged.updated_at = crate::util::now_iso(); + } + None => { + return Err( + crate::managed_agents::effective_config::ORPHANED_INSTANCE_ERROR + .to_string(), + ); + } } } + let relay_url = + crate::relay::effective_agent_relay_url(&staged.relay_url, &workspace_relay); + let key = + crate::managed_agents::ManagedAgentRuntimeKey::new(staged.pubkey.clone(), &relay_url)?; + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + return build_managed_agent_summary(app, record, &runtimes, &personas, &global); + } + runtimes.remove(&key); + let reservation = reserve_managed_agent_start(state, &key)?; + (staged, base_updated_at, key, reservation) + }; + + // Phase B: reuse the established spawn helper against an isolated + // staging map. It may run a login probe and spawn buzz-acp, so this phase + // deliberately holds none of the transition/store/process mutexes. + let mut staged_runtimes = std::collections::HashMap::new(); + start_managed_agent_process( + app, + &mut staged_record, + &mut staged_runtimes, + Some(owner_hex), + )?; + let mut staged_process = staged_runtimes.remove(&key).map(|runtime| runtime.process); + if staged_process.is_none() { + return Err("managed runtime spawn did not produce the reserved pair".into()); } - start_managed_agent_process(app, record, &mut runtimes, Some(owner_hex))?; - save_managed_agents(app, &records)?; - if let Some(saved_record) = records.iter().find(|r| r.pubkey == pubkey) { - retain_managed_agent_pending(app, state, saved_record); + + // Phase C: register only if the record generation and shutdown state still + // match Phase A. A concurrent edit/stop wins, and cleanup happens below + // after all runtime locks have been released. + let mut wrote_receipt = false; + let registration = (|| -> Result { + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + if state + .shutdown_started + .load(std::sync::atomic::Ordering::Acquire) + { + return Err("desktop shutdown started while managed runtime was spawning".into()); + } + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let mut records = load_managed_agents(app)?; + let record = find_managed_agent_mut(&mut records, pubkey)?; + if record.updated_at != base_updated_at { + return Err("managed agent changed while runtime was spawning".into()); + } + if record.backend != BackendKind::Local { + return Err(format!("agent {pubkey} is no longer a local agent")); + } + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + return build_managed_agent_summary(app, record, &runtimes, &personas, &global); + } + runtimes.remove(&key); + *record = staged_record.clone(); + let process = staged_process + .as_ref() + .ok_or_else(|| "managed runtime spawn result was already consumed".to_string())?; + let receipt = crate::managed_agents::ManagedAgentRuntimeReceipt { + key: key.clone(), + pid: process.child.id(), + process_identity: crate::managed_agents::managed_process_identity(process), + desktop_instance_id: current_instance_id(app), + started_at: record + .last_started_at + .clone() + .unwrap_or_else(crate::util::now_iso), + }; + crate::managed_agents::write_agent_runtime_receipt(app, &receipt)?; + wrote_receipt = true; + let Some(process) = staged_process.take() else { + return Err("managed runtime spawn result was already consumed".into()); + }; + runtimes.insert( + key.clone(), + crate::managed_agents::ManagedAgentPairRuntime::starting(process), + ); + if let Err(error) = save_managed_agents(app, &records) { + staged_process = runtimes.remove(&key).map(|runtime| runtime.process); + return Err(error); + } + if let Some(saved_record) = records.iter().find(|r| r.pubkey == pubkey) { + retain_managed_agent_pending(app, state, saved_record); + } + let record = records + .iter() + .find(|record| record.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + build_managed_agent_summary(app, record, &runtimes, &personas, &global) + })(); + + if let Some(mut process) = staged_process { + let _ = crate::managed_agents::terminate_managed_process(&mut process); + let _ = process.child.wait(); + if wrote_receipt { + crate::managed_agents::remove_agent_runtime_receipt(app, &key); + } } - let record = records - .iter() - .find(|record| record.pubkey == pubkey) - .ok_or_else(|| format!("agent {pubkey} not found"))?; - build_managed_agent_summary( - app, - record, - &runtimes, - &personas, - &crate::managed_agents::load_global_agent_config(app).unwrap_or_default(), - ) + registration } /// Deploy an agent to a provider backend. Resolves the binary, calls deploy via diff --git a/desktop/src-tauri/src/managed_agents/discovery.rs b/desktop/src-tauri/src/managed_agents/discovery.rs index fafcb2589d..dfd5e1ebe6 100644 --- a/desktop/src-tauri/src/managed_agents/discovery.rs +++ b/desktop/src-tauri/src/managed_agents/discovery.rs @@ -1,8 +1,6 @@ -use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Command; use std::sync::OnceLock; -use std::time::{Duration, Instant}; use crate::managed_agents::{ buzz_managed_command_path, buzz_managed_node_bin_dir, buzz_managed_npm_bin_dir, @@ -574,9 +572,8 @@ pub fn resolve_command(command: &str) -> Option { pub fn clear_resolve_cache() { let mut guard = resolve_cache().lock().unwrap_or_else(|e| e.into_inner()); guard.clear(); - // Also invalidate the adapter-availability cache so a freshly-installed - // adapter is reflected the next time the summary builder checks the badge. clear_adapter_availability_cache(); + super::readiness::cli_probe::clear_login_probe_cache(); } // ── Adapter availability cache (Phase-2 badge fallback) ───────────────────── @@ -999,102 +996,19 @@ pub(crate) fn is_npm_global_install(cmd: &str) -> bool { || t.starts_with("npm uninstall -g ") } -/// Run a CLI auth probe with a 10-second process-level timeout. -/// -/// Spawns the probe CLI as a child process. Stdout and stderr are drained on -/// background threads to prevent pipe-buffer deadlock. On timeout the child is -/// killed and `Unknown` is returned; no orphaned threads or processes are left -/// behind. Returns `Unknown` on timeout. +/// Run a cached, single-flight CLI auth probe with the shared five-second +/// process deadline and bounded-output collector. fn probe_auth_status(binary_path: &Path, probe_args: &[&str]) -> AuthStatus { use crate::managed_agents::readiness::cli_probe; let augmented_path = cli_probe::augmented_path(); - - let mut command = std::process::Command::new(binary_path); - command.args(&probe_args[1..]); - if let Some(ref path) = augmented_path { - command.env("PATH", path); - } - command - .stdin(std::process::Stdio::null()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); - crate::util::configure_no_window(&mut command); - - let mut child = match command.spawn() { - Ok(c) => c, - Err(_) => return AuthStatus::Unknown, - }; - - // Drain stdout/stderr on background threads to prevent pipe-buffer deadlock. - let stdout_pipe = child.stdout.take(); - let stderr_pipe = child.stderr.take(); - - let stdout_thread = std::thread::spawn(move || { - let mut buf = Vec::new(); - if let Some(mut pipe) = stdout_pipe { - let _ = pipe.read_to_end(&mut buf); - } - }); - let stderr_thread = std::thread::spawn(move || { - let mut buf = Vec::new(); - if let Some(mut pipe) = stderr_pipe { - let _ = pipe.read_to_end(&mut buf); - } - buf - }); - - // Save PID for kill-on-timeout before moving child into the wait thread. - let child_pid = child.id(); - let (tx, rx) = std::sync::mpsc::channel(); - let wait_thread = std::thread::spawn(move || { - let _ = tx.send(child.wait()); - }); - - // 10-second timeout for auth probes. - let deadline = Instant::now() + Duration::from_secs(10); - let exit_status = loop { - let remaining = deadline.saturating_duration_since(Instant::now()); - if remaining.is_zero() { - #[cfg(unix)] - unsafe { - libc::kill(child_pid as i32, libc::SIGTERM); - } - #[cfg(not(unix))] - let _ = child_pid; - drop(rx); - let _ = wait_thread.join(); - let _ = stdout_thread.join(); - let _ = stderr_thread.join(); - return AuthStatus::Unknown; - } - match rx.recv_timeout(Duration::from_millis(100).min(remaining)) { - Ok(Ok(status)) => break status, - Ok(Err(_)) => { - let _ = wait_thread.join(); - let _ = stdout_thread.join(); - let _ = stderr_thread.join(); - return AuthStatus::Unknown; - } - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue, - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - let _ = stdout_thread.join(); - let _ = stderr_thread.join(); - return AuthStatus::Unknown; - } - } - }; - - let _ = wait_thread.join(); - let _ = stdout_thread.join(); - let stderr_bytes = stderr_thread.join().unwrap_or_default(); - - match cli_probe::classify_probe_output(&stderr_bytes, exit_status.success()) { + match cli_probe::login_probe(binary_path, probe_args, augmented_path.as_deref()) { cli_probe::ProbeOutcome::LoggedIn => AuthStatus::LoggedIn, cli_probe::ProbeOutcome::LoggedOut => AuthStatus::LoggedOut, cli_probe::ProbeOutcome::ConfigInvalid { stderr_excerpt } => AuthStatus::ConfigInvalid { diagnostic: stderr_excerpt, }, + cli_probe::ProbeOutcome::TimedOut => AuthStatus::Unknown, } } diff --git a/desktop/src-tauri/src/managed_agents/global_config/mod.rs b/desktop/src-tauri/src/managed_agents/global_config/mod.rs index 162f447981..af203ba1c3 100644 --- a/desktop/src-tauri/src/managed_agents/global_config/mod.rs +++ b/desktop/src-tauri/src/managed_agents/global_config/mod.rs @@ -204,7 +204,9 @@ pub fn save_global_agent_config(app: &AppHandle, config: &GlobalAgentConfig) -> let path = global_config_path(app)?; let payload = serde_json::to_vec_pretty(&config) .map_err(|e| format!("failed to serialize global agent config: {e}"))?; - atomic_write_json_restricted(&path, &payload) + atomic_write_json_restricted(&path, &payload)?; + crate::managed_agents::readiness::cli_probe::clear_login_probe_cache(); + Ok(()) } /// Resolve the effective model and provider for an agent. diff --git a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs index 479d6ec913..2e65ff89dc 100644 --- a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs +++ b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs @@ -6,14 +6,23 @@ //! - [`JobHandle`] / [`create_job_for_child`] — the in-process stop path. A //! Job Object owns the tree and `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` kills //! it when the handle drops. -//! - [`taskkill_tree`] — the after-restart path, where only the PID survives -//! in the record and no job handle is available. +//! - [`terminate_process_tree`] — the after-restart path, where only the PID +//! survives in the record and no job handle is available. It uses Win32 +//! process enumeration/termination directly so teardown never launches an +//! external helper while a runtime-management lock is held. //! //! This module is `#[cfg(windows)]`-only; nothing here compiles on other //! platforms. +use std::{collections::HashMap, os::windows::io::AsRawHandle, time::Duration}; + use windows_sys::Win32::Foundation::HANDLE; +use super::runtime::{ + next_verified_windows_descendants, terminate_if_windows_identity_matches, + WindowsIdentityObservation, WindowsProcessIdentity, WindowsProcessSnapshotEntry, +}; + /// Win32 Job Object that owns the harness process and (via Windows' default /// child-inheritance) every process it spawns. Dropping the handle with /// `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` set kills the whole tree — the Windows @@ -40,25 +49,68 @@ impl Drop for JobHandle { } } -/// Create a Job Object, assign `pid` to it, and configure it to kill the whole -/// tree when the returned handle is dropped. Returns `None` on any failure so -/// the caller can fall back to `Child::kill()` — a degraded teardown beats a -/// failed spawn. +impl JobHandle { + /// Terminate every process assigned to this job and wait for the job's + /// active-process count to reach zero. Keeping the job handle open during + /// the wait prevents the containment boundary from disappearing early. + pub(crate) fn terminate_and_wait(&self, timeout: Duration) -> Result<(), String> { + use windows_sys::Win32::Foundation::FALSE; + use windows_sys::Win32::System::JobObjects::{ + JobObjectBasicAccountingInformation, QueryInformationJobObject, TerminateJobObject, + JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, + }; + + unsafe { + if TerminateJobObject(self.0, 1) == FALSE { + return Err(format!( + "failed to terminate Windows Job Object: error {}", + windows_sys::Win32::Foundation::GetLastError() + )); + } + } + let started = std::time::Instant::now(); + loop { + let mut info = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default(); + let queried = unsafe { + QueryInformationJobObject( + self.0, + JobObjectBasicAccountingInformation, + &mut info as *mut _ as *mut _, + std::mem::size_of::() as u32, + std::ptr::null_mut(), + ) + }; + if queried == FALSE { + return Err(format!( + "failed to query Windows Job Object cleanup: error {}", + unsafe { windows_sys::Win32::Foundation::GetLastError() } + )); + } + if info.ActiveProcesses == 0 { + return Ok(()); + } + if started.elapsed() >= timeout { + return Err(format!( + "Windows Job Object still owns {} process(es) after {} ms", + info.ActiveProcesses, + timeout.as_millis() + )); + } + std::thread::sleep(Duration::from_millis(10)); + } + } +} + +/// Create a Job Object, assign `child` through its stable process handle, and +/// configure it to kill the whole tree when the returned handle is dropped. +/// Using the child handle avoids reopening a possibly recycled numeric PID. /// /// Assignment happens immediately after spawn, on the same parent thread. The -/// child (buzz-acp) does spawn its 24 workers before it connects to the relay, -/// so the window between our spawn and our assignment is NOT structurally empty. -/// What closes it is assign-latency: `OpenProcess` + `AssignProcessToJobObject` -/// are a few synchronous Win32 calls (microseconds), while buzz-acp must init -/// tokio, parse its config, and spawn 24 children (tens-to-hundreds of ms), so -/// the assign reliably wins before any worker exists. Once assigned, Windows -/// places every subsequently-spawned descendant in the job automatically. -/// -/// `CREATE_SUSPENDED` -> assign -> `ResumeThread` would make the window airtight -/// regardless of child timing, but it requires raw `CreateProcessW`/`ResumeThread` -/// (materially more unsafe Win32) to close a microsecond race, so it is -/// deliberately not used here. -fn create_job_for_child(pid: u32) -> Option { +/// long-running ACP harness is not suspended here, so a child could still exit +/// or create a descendant before assignment; assignment failure therefore +/// degrades only to identity-bound teardown. Readiness probes use the stricter +/// suspended-spawn path below and never execute before job assignment. +fn create_job_for_child(child: &std::process::Child) -> Result { use std::ptr::null; use windows_sys::Win32::Foundation::{CloseHandle, FALSE}; use windows_sys::Win32::System::JobObjects::{ @@ -66,14 +118,14 @@ fn create_job_for_child(pid: u32) -> Option { SetInformationJobObject, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, }; - use windows_sys::Win32::System::Threading::{ - OpenProcess, PROCESS_SET_QUOTA, PROCESS_TERMINATE, - }; unsafe { let job = CreateJobObjectW(null(), null()); if job.is_null() { - return None; + return Err(format!( + "failed to create Windows Job Object: error {}", + windows_sys::Win32::Foundation::GetLastError() + )); } let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed(); @@ -85,74 +137,493 @@ fn create_job_for_child(pid: u32) -> Option { std::mem::size_of::() as u32, ); if ok == FALSE { + let error = windows_sys::Win32::Foundation::GetLastError(); CloseHandle(job); - return None; + return Err(format!( + "failed to configure Windows Job Object: error {error}" + )); } - let process = OpenProcess(PROCESS_SET_QUOTA | PROCESS_TERMINATE, FALSE, pid); - if process.is_null() { - CloseHandle(job); - return None; - } + let process = child.as_raw_handle() as HANDLE; let assigned = AssignProcessToJobObject(job, process); - CloseHandle(process); if assigned == FALSE { + let error = windows_sys::Win32::Foundation::GetLastError(); CloseHandle(job); - return None; + return Err(format!( + "failed to assign child {} to Windows Job Object: error {error}", + child.id() + )); } - Some(JobHandle(job)) + Ok(JobHandle(job)) + } +} + +struct OwnedProcessHandle(HANDLE); + +impl Drop for OwnedProcessHandle { + fn drop(&mut self) { + unsafe { windows_sys::Win32::Foundation::CloseHandle(self.0) }; + } +} + +fn creation_time_from_handle(handle: HANDLE) -> Result { + use windows_sys::Win32::Foundation::{FALSE, FILETIME}; + use windows_sys::Win32::System::Threading::GetProcessTimes; + + let mut creation = FILETIME::default(); + let mut exit = FILETIME::default(); + let mut kernel = FILETIME::default(); + let mut user = FILETIME::default(); + let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) }; + if ok == FALSE { + return Err(format!( + "failed to query Windows process creation time: error {}", + unsafe { windows_sys::Win32::Foundation::GetLastError() } + )); + } + Ok(((creation.dwHighDateTime as u64) << 32) | creation.dwLowDateTime as u64) +} + +pub(crate) fn child_process_identity(child: &std::process::Child) -> Result { + creation_time_from_handle(child.as_raw_handle() as HANDLE) +} + +fn open_process_handle(pid: u32, access: u32) -> Result, String> { + use windows_sys::Win32::Foundation::{ERROR_INVALID_PARAMETER, FALSE}; + use windows_sys::Win32::System::Threading::OpenProcess; + + let handle = unsafe { OpenProcess(access, FALSE, pid) }; + if handle.is_null() { + let error = unsafe { windows_sys::Win32::Foundation::GetLastError() }; + if error == ERROR_INVALID_PARAMETER { + return Ok(None); + } + return Err(format!( + "failed to open Windows process {pid}: error {error}" + )); + } + Ok(Some(OwnedProcessHandle(handle))) +} + +fn identity_from_pid(pid: u32) -> WindowsIdentityObservation { + use windows_sys::Win32::System::Threading::PROCESS_QUERY_LIMITED_INFORMATION; + + let Ok(handle) = open_process_handle(pid, PROCESS_QUERY_LIMITED_INFORMATION) else { + return WindowsIdentityObservation::Unverified; + }; + let Some(handle) = handle else { + return WindowsIdentityObservation::Exited; + }; + match creation_time_from_handle(handle.0) { + Ok(creation_time) => { + WindowsIdentityObservation::Verified(WindowsProcessIdentity { pid, creation_time }) + } + Err(_) => WindowsIdentityObservation::Unverified, + } +} + +pub(crate) fn process_identity_matches(pid: u32, expected_creation_time: u64) -> bool { + use windows_sys::Win32::Foundation::{FALSE, STILL_ACTIVE}; + use windows_sys::Win32::System::Threading::{ + GetExitCodeProcess, PROCESS_QUERY_LIMITED_INFORMATION, + }; + + let Ok(Some(handle)) = open_process_handle(pid, PROCESS_QUERY_LIMITED_INFORMATION) else { + return false; + }; + let Ok(creation_time) = creation_time_from_handle(handle.0) else { + return false; + }; + let mut exit_code = 0; + let queried = unsafe { GetExitCodeProcess(handle.0, &mut exit_code) }; + queried != FALSE && exit_code == STILL_ACTIVE as u32 && creation_time == expected_creation_time +} + +fn resume_suspended_process(pid: u32) -> Result<(), String> { + use windows_sys::Win32::Foundation::{CloseHandle, FALSE, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, Thread32First, Thread32Next, TH32CS_SNAPTHREAD, THREADENTRY32, + }; + use windows_sys::Win32::System::Threading::{OpenThread, ResumeThread, THREAD_SUSPEND_RESUME}; + + unsafe { + let snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0); + if snapshot == INVALID_HANDLE_VALUE { + return Err(format!( + "failed to snapshot Windows threads for suspended probe {pid}: error {}", + windows_sys::Win32::Foundation::GetLastError() + )); + } + let mut entry = THREADENTRY32::default(); + entry.dwSize = std::mem::size_of::() as u32; + let mut found = None; + if Thread32First(snapshot, &mut entry) != FALSE { + loop { + if entry.th32OwnerProcessID == pid { + found = Some(entry.th32ThreadID); + break; + } + if Thread32Next(snapshot, &mut entry) == FALSE { + break; + } + } + } + CloseHandle(snapshot); + let Some(thread_id) = found else { + return Err(format!( + "failed to find primary thread for suspended Windows probe {pid}" + )); + }; + let thread = OpenThread(THREAD_SUSPEND_RESUME, FALSE, thread_id); + if thread.is_null() { + return Err(format!( + "failed to open primary thread for suspended Windows probe {pid}: error {}", + windows_sys::Win32::Foundation::GetLastError() + )); + } + let previous = ResumeThread(thread); + let error = windows_sys::Win32::Foundation::GetLastError(); + CloseHandle(thread); + if previous == u32::MAX { + return Err(format!( + "failed to resume suspended Windows probe {pid}: error {error}" + )); + } + Ok(()) } } -/// Kill the entire process tree rooted at `pid` via `taskkill /T`, the closest -/// equivalent to the Unix process-group kill. Used on the after-restart path -/// where no job handle survived. `CREATE_NO_WINDOW` keeps taskkill's own -/// console from flashing. -pub fn taskkill_tree(pid: u32) -> Result<(), String> { +/// Spawn a probe suspended, bind it to a fresh kill-on-close Job Object, then +/// resume its primary thread. Assignment failures occur before user code can +/// execute, so no probe is allowed to run outside its dedicated containment +/// boundary. +pub(crate) fn spawn_probe_in_job( + command: &mut std::process::Command, +) -> Result<(std::process::Child, JobHandle), String> { use std::os::windows::process::CommandExt; + use windows_sys::Win32::System::Threading::CREATE_SUSPENDED; + const CREATE_NO_WINDOW: u32 = 0x0800_0000; - let status = std::process::Command::new("taskkill") - .args(["/T", "/F", "/PID", &pid.to_string()]) - .creation_flags(CREATE_NO_WINDOW) - .status() - .map_err(|error| format!("failed to run taskkill for pid {pid}: {error}"))?; - if status.success() { + command.creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED); + let mut child = command + .spawn() + .map_err(|error| format!("failed to spawn suspended Windows readiness probe: {error}"))?; + let job = match create_job_for_child(&child) { + Ok(job) => job, + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(error); + } + }; + if let Err(error) = resume_suspended_process(child.id()) { + drop(job); + let _ = child.kill(); + let _ = child.wait(); + return Err(error); + } + Ok((child, job)) +} + +/// Snapshot `(pid, parent_pid)` for every visible process without launching a +/// helper executable. +fn process_snapshot() -> Result, String> { + use windows_sys::Win32::Foundation::{CloseHandle, FALSE, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W, + TH32CS_SNAPPROCESS, + }; + + unsafe { + let snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0); + if snapshot == INVALID_HANDLE_VALUE { + return Err(format!( + "failed to snapshot Windows processes: error {}", + windows_sys::Win32::Foundation::GetLastError() + )); + } + let mut entry: PROCESSENTRY32W = std::mem::zeroed(); + entry.dwSize = std::mem::size_of::() as u32; + let mut entries = Vec::new(); + if Process32FirstW(snapshot, &mut entry) != FALSE { + loop { + let identity = match identity_from_pid(entry.th32ProcessID) { + WindowsIdentityObservation::Verified(identity) => Some(identity), + WindowsIdentityObservation::Exited | WindowsIdentityObservation::Unverified => { + None + } + }; + entries.push(WindowsProcessSnapshotEntry { + pid: entry.th32ProcessID, + parent_pid: entry.th32ParentProcessID, + identity, + }); + if Process32NextW(snapshot, &mut entry) == FALSE { + break; + } + } + } + CloseHandle(snapshot); + Ok(entries) + } +} + +struct KnownProcess { + identity: WindowsProcessIdentity, + handle: OwnedProcessHandle, + depth: usize, + terminated: bool, +} + +fn open_verified_process( + expected: WindowsProcessIdentity, +) -> Result, String> { + use windows_sys::Win32::System::Threading::{ + PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE, + }; + + let Some(handle) = open_process_handle( + expected.pid, + PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, + )? + else { + return Ok(None); + }; + let observed = creation_time_from_handle(handle.0) + .map(|creation_time| { + WindowsIdentityObservation::Verified(WindowsProcessIdentity { + pid: expected.pid, + creation_time, + }) + }) + .unwrap_or(WindowsIdentityObservation::Unverified); + terminate_if_windows_identity_matches(expected, observed, || Ok(()))?; + Ok(Some(handle)) +} + +fn terminate_open_process(process: &mut KnownProcess) -> Result<(), String> { + use windows_sys::Win32::Foundation::{FALSE, WAIT_OBJECT_0}; + use windows_sys::Win32::System::Threading::{TerminateProcess, WaitForSingleObject}; + + let observed = creation_time_from_handle(process.handle.0) + .map(|creation_time| { + WindowsIdentityObservation::Verified(WindowsProcessIdentity { + pid: process.identity.pid, + creation_time, + }) + }) + .unwrap_or(WindowsIdentityObservation::Unverified); + terminate_if_windows_identity_matches(process.identity, observed, || { + let terminated = unsafe { TerminateProcess(process.handle.0, 1) }; + if terminated == FALSE + && unsafe { WaitForSingleObject(process.handle.0, 0) } != WAIT_OBJECT_0 + { + return Err(format!( + "failed to terminate verified Windows process {}: error {}", + process.identity.pid, + unsafe { windows_sys::Win32::Foundation::GetLastError() } + )); + } + Ok(()) + })?; + if unsafe { WaitForSingleObject(process.handle.0, 1_000) } != WAIT_OBJECT_0 { + return Err(format!( + "verified Windows process {} did not exit within one second", + process.identity.pid + )); + } + process.terminated = true; + Ok(()) +} + +fn capture_verified_descendants( + entries: &[WindowsProcessSnapshotEntry], + known: &mut HashMap, + errors: &mut Vec, +) -> usize { + let mut added_total = 0; + loop { + let identities = known + .iter() + .map(|(pid, process)| (*pid, process.identity)) + .collect::>(); + let (candidates, unverified) = next_verified_windows_descendants(entries, &identities); + for pid in unverified { + let message = format!( + "refusing to terminate descendant PID {pid}: stable identity was unavailable" + ); + if !errors.contains(&message) { + errors.push(message); + } + } + if candidates.is_empty() { + break; + } + let mut added = 0; + for identity in candidates { + let parent_pid = entries + .iter() + .find(|entry| entry.identity == Some(identity)) + .map(|entry| entry.parent_pid); + let Some(parent_pid) = parent_pid else { + continue; + }; + let depth = known + .get(&parent_pid) + .map(|parent| parent.depth + 1) + .unwrap_or(1); + match open_verified_process(identity) { + Ok(Some(handle)) => { + known.insert( + identity.pid, + KnownProcess { + identity, + handle, + depth, + terminated: false, + }, + ); + added += 1; + } + Ok(None) => { + // Natural exit between enumeration and opening is safe. + } + Err(error) => errors.push(error), + } + } + added_total += added; + if added == 0 { + break; + } + } + added_total +} + +/// Terminate the Windows process tree only when the root still has the +/// creation identity persisted at spawn time. Every descendant is likewise +/// opened and identity-checked before termination; handles remain open until +/// teardown ends so none of their PIDs can be recycled under the algorithm. +pub fn terminate_process_tree(pid: u32, expected_creation_time: Option) -> Result<(), String> { + let expected_creation_time = expected_creation_time.ok_or_else(|| { + format!( + "refusing to terminate Windows process tree {pid}: no stable root identity was recorded" + ) + })?; + let root_identity = WindowsProcessIdentity { + pid, + creation_time: expected_creation_time, + }; + let Some(root_handle) = open_verified_process(root_identity)? else { + return Ok(()); + }; + let mut known = HashMap::from([( + pid, + KnownProcess { + identity: root_identity, + handle: root_handle, + depth: 0, + terminated: false, + }, + )]); + let mut errors = Vec::new(); + + // Capture all descendants visible before stopping the root. Their open + // handles bind the enumerated PIDs to those exact process instances. + let before = process_snapshot()?; + capture_verified_descendants(&before, &mut known, &mut errors); + if let Some(root) = known.get_mut(&pid) { + terminate_open_process(root)?; + } + + // A descendant can race the first snapshot. Re-snapshot after root exit, + // capture descendants of every handle-bound process, terminate deepest + // first, and repeat until a full pass finds no new process. The bound is a + // cleanup-failure boundary, never permission to kill an ambiguous PID. + let mut clean_passes = 0; + for _ in 0..8 { + let snapshot = process_snapshot()?; + let added = capture_verified_descendants(&snapshot, &mut known, &mut errors); + let mut order = known + .iter() + .filter_map(|(candidate, process)| { + (*candidate != pid && !process.terminated).then_some((process.depth, *candidate)) + }) + .collect::>(); + order.sort_unstable_by(|left, right| right.cmp(left)); + for (_, candidate) in order { + if let Some(process) = known.get_mut(&candidate) { + if let Err(error) = terminate_open_process(process) { + errors.push(error); + } + } + } + if added == 0 { + clean_passes += 1; + if clean_passes == 2 { + break; + } + } else { + clean_passes = 0; + } + std::thread::sleep(Duration::from_millis(10)); + } + if clean_passes < 2 { + errors.push(format!( + "managed Windows process tree {pid} did not quiesce within eight identity-checked passes" + )); + } + if errors.is_empty() { Ok(()) } else { Err(format!( - "taskkill exited with status {status} for pid {pid}" + "failed to terminate part of managed process tree {pid}: {}", + errors.join("; ") )) } } -/// Assign a freshly-spawned harness `child` to a Job Object and package it into -/// a [`ManagedAgentProcess`]. On job-assignment failure the process is still -/// returned with `job: None` — teardown then falls back to `Child::kill()`, -/// which kills only the harness (a degraded teardown beats a failed spawn). +/// Record the stable process identity, assign a freshly-spawned harness to a +/// Job Object, and package it into a [`ManagedAgentProcess`]. Identity failure +/// aborts and reaps the child so an unidentifiable process is never registered. +/// Job assignment may still degrade to identity-bound native teardown. pub fn finish_spawn( - child: std::process::Child, + mut child: std::process::Child, log_path: std::path::PathBuf, spawn_config: super::spawn_snapshot::SpawnConfigSnapshot, setup_mode: bool, adapter_availability: Option, start_nonce: String, agent_name: &str, -) -> super::ManagedAgentProcess { - let job = create_job_for_child(child.id()); - if job.is_none() { - eprintln!( - "buzz-desktop: failed to assign agent {agent_name} to a Job Object; \ - teardown will fall back to killing only the harness process" - ); - } - super::ManagedAgentProcess { +) -> Result { + let process_identity = match child_process_identity(&child) { + Ok(identity) => identity, + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(format!( + "failed to record Windows process identity for agent {agent_name}: {error}" + )); + } + }; + let job = match create_job_for_child(&child) { + Ok(job) => Some(job), + Err(error) => { + eprintln!( + "buzz-desktop: failed to assign agent {agent_name} to a Job Object; \ + teardown remains identity-bound: {error}" + ); + None + } + }; + Ok(super::ManagedAgentProcess { child, log_path, spawn_config, setup_mode, adapter_availability, start_nonce, + process_identity, job, - } + }) } diff --git a/desktop/src-tauri/src/managed_agents/readiness/cli_login.rs b/desktop/src-tauri/src/managed_agents/readiness/cli_login.rs index 4036d9f239..98c3ce7c40 100644 --- a/desktop/src-tauri/src/managed_agents/readiness/cli_login.rs +++ b/desktop/src-tauri/src/managed_agents/readiness/cli_login.rs @@ -61,6 +61,11 @@ pub(super) fn requirements( diagnostic: stderr_excerpt, }] } + cli_probe::ProbeOutcome::TimedOut => vec![missing_requirement( + probe_args, + setup_copy, + AcpAvailabilityStatus::Available, + )], } } other => vec![missing_requirement(probe_args, setup_copy, other)], diff --git a/desktop/src-tauri/src/managed_agents/readiness/cli_probe.rs b/desktop/src-tauri/src/managed_agents/readiness/cli_probe.rs index 513da4e2a8..2daca613b1 100644 --- a/desktop/src-tauri/src/managed_agents/readiness/cli_probe.rs +++ b/desktop/src-tauri/src/managed_agents/readiness/cli_probe.rs @@ -1,4 +1,14 @@ -use std::path::Path; +use std::{ + collections::HashMap, + ffi::OsString, + io::{Read, Seek, SeekFrom}, + path::{Path, PathBuf}, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, Condvar, Mutex, OnceLock, + }, + time::{Duration, Instant}, +}; use crate::managed_agents::runtime::build_augmented_path; @@ -22,7 +32,7 @@ pub(crate) fn augmented_path() -> Option { } /// Outcome of a CLI login-status probe. -#[derive(Debug, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum ProbeOutcome { /// The CLI reported a successful login (exit 0). LoggedIn, @@ -36,6 +46,287 @@ pub(crate) enum ProbeOutcome { /// A trimmed excerpt of the stderr message to surface in the nudge. stderr_excerpt: String, }, + /// The CLI did not finish within the hard process deadline. The child was + /// killed and reaped before this result was returned. + TimedOut, +} + +const LOGIN_PROBE_TIMEOUT: Duration = Duration::from_secs(5); +const LOGIN_PROBE_CACHE_TTL: Duration = Duration::from_secs(45); +const CONFIG_ERROR_CACHE_TTL: Duration = Duration::from_secs(5); +const MAX_PROBE_OUTPUT_BYTES: usize = 64 * 1024; + +trait ProbeTreeGuard { + fn terminate_and_wait(&mut self) -> Result<(), String>; +} + +#[cfg(unix)] +struct NativeProbeTreeGuard { + process_group: i32, +} + +#[cfg(unix)] +impl ProbeTreeGuard for NativeProbeTreeGuard { + fn terminate_and_wait(&mut self) -> Result<(), String> { + use nix::{ + errno::Errno, + sys::signal::{killpg, Signal}, + unistd::Pid, + }; + + match killpg(Pid::from_raw(self.process_group), Signal::SIGKILL) { + Ok(()) | Err(Errno::ESRCH) => Ok(()), + Err(error) => Err(format!( + "failed to terminate readiness probe process group {}: {error}", + self.process_group + )), + } + } +} + +#[cfg(windows)] +struct NativeProbeTreeGuard { + job: crate::managed_agents::JobHandle, +} + +#[cfg(windows)] +impl ProbeTreeGuard for NativeProbeTreeGuard { + fn terminate_and_wait(&mut self) -> Result<(), String> { + self.job.terminate_and_wait(Duration::from_secs(1)) + } +} + +#[cfg(not(any(unix, windows)))] +struct NativeProbeTreeGuard; + +#[cfg(not(any(unix, windows)))] +impl ProbeTreeGuard for NativeProbeTreeGuard { + fn terminate_and_wait(&mut self) -> Result<(), String> { + Ok(()) + } +} + +fn spawn_contained_probe( + command: &mut std::process::Command, +) -> Result<(std::process::Child, NativeProbeTreeGuard), String> { + #[cfg(windows)] + { + let (child, job) = crate::managed_agents::spawn_probe_in_job(command)?; + return Ok((child, NativeProbeTreeGuard { job })); + } + #[cfg(not(windows))] + { + let child = command + .spawn() + .map_err(|error| format!("failed to spawn readiness probe: {error}"))?; + #[cfg(unix)] + let guard = NativeProbeTreeGuard { + process_group: child.id() as i32, + }; + #[cfg(not(any(unix, windows)))] + let guard = NativeProbeTreeGuard; + Ok((child, guard)) + } +} + +fn terminate_contained_probe( + child: &mut std::process::Child, + guard: &mut impl ProbeTreeGuard, +) -> Result<(), String> { + let tree_result = guard.terminate_and_wait(); + let _ = child.kill(); + tree_result +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct ProbeCacheKey { + generation: u64, + runtime: String, + binary_path: PathBuf, + args: Vec, + effective_path: Option, + effective_environment: Vec<(&'static str, Option)>, +} + +impl ProbeCacheKey { + fn same_probe_identity(&self, other: &Self) -> bool { + self.runtime == other.runtime + && self.binary_path == other.binary_path + && self.args == other.args + && self.effective_path == other.effective_path + && self.effective_environment == other.effective_environment + } +} + +#[derive(Debug)] +struct ProbeFlight { + result: Mutex>, + ready: Condvar, +} + +impl ProbeFlight { + fn wait(&self) -> ProbeOutcome { + let mut result = self + .result + .lock() + .unwrap_or_else(|error| error.into_inner()); + while result.is_none() { + result = self + .ready + .wait(result) + .unwrap_or_else(|error| error.into_inner()); + } + result.clone().unwrap_or(ProbeOutcome::LoggedOut) + } + + fn publish(&self, outcome: ProbeOutcome) { + let mut result = self + .result + .lock() + .unwrap_or_else(|error| error.into_inner()); + *result = Some(outcome); + self.ready.notify_all(); + } +} + +#[derive(Debug)] +enum ProbeCacheEntry { + InFlight(Arc), + Complete { + completed_at: Instant, + outcome: ProbeOutcome, + }, +} + +#[derive(Default)] +struct LoginProbeCache { + generation: AtomicU64, + entries: Mutex>, +} + +enum ProbeDecision { + Cached(ProbeOutcome), + Wait(Arc), + Run(Arc), +} + +impl LoginProbeCache { + fn probe(&self, mut key: ProbeCacheKey, now: N, run: F) -> ProbeOutcome + where + F: FnOnce() -> ProbeOutcome, + N: Fn() -> Instant, + { + key.generation = self.generation.load(Ordering::Acquire); + let decision = { + let observed_at = now(); + let mut entries = self + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + // Invalidation expires completed values immediately, but an + // already-running identical command remains the authoritative + // flight until it settles. Clearing it here would allow the same + // probe key to run twice during a configuration refresh. + entries.retain(|candidate, entry| { + candidate.generation == key.generation + || matches!(entry, ProbeCacheEntry::InFlight(_)) + }); + let cached = entries.get(&key).and_then(|entry| match entry { + ProbeCacheEntry::Complete { + completed_at, + outcome, + } => { + let ttl = if matches!(outcome, ProbeOutcome::ConfigInvalid { .. }) { + CONFIG_ERROR_CACHE_TTL + } else { + LOGIN_PROBE_CACHE_TTL + }; + (observed_at.saturating_duration_since(*completed_at) < ttl) + .then(|| outcome.clone()) + } + ProbeCacheEntry::InFlight(_) => None, + }); + if let Some(outcome) = cached { + ProbeDecision::Cached(outcome) + } else if let Some(flight) = entries.iter().find_map(|(candidate, entry)| { + if candidate.same_probe_identity(&key) { + if let ProbeCacheEntry::InFlight(flight) = entry { + return Some(Arc::clone(flight)); + } + } + None + }) { + ProbeDecision::Wait(flight) + } else { + entries.remove(&key); + let flight = Arc::new(ProbeFlight { + result: Mutex::new(None), + ready: Condvar::new(), + }); + entries.insert(key.clone(), ProbeCacheEntry::InFlight(Arc::clone(&flight))); + ProbeDecision::Run(flight) + } + }; + + match decision { + ProbeDecision::Cached(outcome) => outcome, + ProbeDecision::Wait(flight) => flight.wait(), + ProbeDecision::Run(flight) => { + // The external command runs without the cache mutex. Identical + // callers wait on this key's flight; different keys remain + // independent and may probe concurrently. + let outcome = run(); + flight.publish(outcome.clone()); + let completed_at = now(); + let mut entries = self + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + if self.generation.load(Ordering::Acquire) == key.generation + && matches!( + entries.get(&key), + Some(ProbeCacheEntry::InFlight(current)) if Arc::ptr_eq(current, &flight) + ) + { + entries.insert( + key, + ProbeCacheEntry::Complete { + completed_at, + outcome: outcome.clone(), + }, + ); + } else if matches!( + entries.get(&key), + Some(ProbeCacheEntry::InFlight(current)) if Arc::ptr_eq(current, &flight) + ) { + // The result belongs to the pre-invalidation generation. + // Wake its waiters but do not leave a stale flight/value + // behind; the next call recomputes against current config. + entries.remove(&key); + } + outcome + } + } + } + + fn invalidate(&self) { + self.generation.fetch_add(1, Ordering::AcqRel); + self.entries + .lock() + .unwrap_or_else(|error| error.into_inner()) + .retain(|_, entry| matches!(entry, ProbeCacheEntry::InFlight(_))); + } +} + +fn login_probe_cache() -> &'static LoginProbeCache { + static CACHE: OnceLock = OnceLock::new(); + CACHE.get_or_init(LoginProbeCache::default) +} + +pub(crate) fn clear_login_probe_cache() { + // The cache mutex is never held while an external process runs, so + // configuration writes and manual discovery refreshes invalidate promptly. + login_probe_cache().invalidate(); } /// Signals emitted to stderr by codex (and related CLI tools) when they @@ -58,25 +349,139 @@ pub(crate) fn login_probe( probe_args: &[&str], augmented_path: Option<&str>, ) -> ProbeOutcome { + login_probe_with_timeout(binary_path, probe_args, augmented_path, LOGIN_PROBE_TIMEOUT) +} + +fn login_probe_with_timeout( + binary_path: &Path, + probe_args: &[&str], + augmented_path: Option<&str>, + timeout: Duration, +) -> ProbeOutcome { + let key = ProbeCacheKey { + generation: 0, + runtime: probe_args.first().copied().unwrap_or_default().to_string(), + binary_path: binary_path + .canonicalize() + .unwrap_or_else(|_| binary_path.to_path_buf()), + args: probe_args + .iter() + .map(|value| (*value).to_string()) + .collect(), + effective_path: augmented_path.map(str::to_owned), + effective_environment: ["HOME", "XDG_CONFIG_HOME", "CODEX_HOME", "CLAUDE_CONFIG_DIR"] + .into_iter() + .map(|name| (name, std::env::var_os(name))) + .collect(), + }; + login_probe_cache().probe(key, Instant::now, || { + run_login_probe(binary_path, probe_args, augmented_path, timeout) + }) +} + +fn run_login_probe( + binary_path: &Path, + probe_args: &[&str], + augmented_path: Option<&str>, + timeout: Duration, +) -> ProbeOutcome { + // Regular files cannot be held at a blocking EOF by forked descendants. + // This keeps output collection inside the same deadline as the child even + // when a CLI exits after spawning a process that inherited stdout/stderr. + let Ok(mut stdout) = tempfile::tempfile() else { + return ProbeOutcome::LoggedOut; + }; + let Ok(mut stderr) = tempfile::tempfile() else { + return ProbeOutcome::LoggedOut; + }; + let (Ok(stdout_writer), Ok(stderr_writer)) = (stdout.try_clone(), stderr.try_clone()) else { + return ProbeOutcome::LoggedOut; + }; let mut command = std::process::Command::new(binary_path); command.args(&probe_args[1..]); if let Some(path) = augmented_path { command.env("PATH", path); } crate::util::configure_no_window(&mut command); + command.stdout(stdout_writer).stderr(stderr_writer); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.process_group(0); + } - match command.output() { - Ok(o) if o.status.success() => ProbeOutcome::LoggedIn, - Ok(o) => classify_probe_output(&o.stderr, false), - Err(_) => ProbeOutcome::LoggedOut, + let (mut child, mut tree_guard) = match spawn_contained_probe(&mut command) { + Ok(contained) => contained, + Err(error) => { + eprintln!("buzz-desktop: readiness probe containment failed: {error}"); + return ProbeOutcome::LoggedOut; + } + }; + let started_at = Instant::now(); + let (status, timed_out, mut containment_ok) = loop { + match child.try_wait() { + Ok(Some(status)) => break (Some(status), false, true), + Ok(None) if started_at.elapsed() < timeout => { + std::thread::sleep(Duration::from_millis(20)); + } + Ok(None) => { + let containment_ok = + cleanup_contained_probe(&mut child, &mut tree_guard, "after timeout"); + break (child.wait().ok(), true, containment_ok); + } + Err(_) => { + let containment_ok = + cleanup_contained_probe(&mut child, &mut tree_guard, "after wait failure"); + let _ = child.wait(); + break (None, false, containment_ok); + } + } + }; + // The direct child may have exited after forking a helper that inherited + // its output descriptors. Terminate that probe-only process group before + // returning so no authentication helper survives a completed probe. + containment_ok &= cleanup_contained_probe(&mut child, &mut tree_guard, "after completion"); + let _stdout = read_bounded_output(&mut stdout); + let stderr = read_bounded_output(&mut stderr); + if !containment_ok { + ProbeOutcome::LoggedOut + } else if timed_out { + ProbeOutcome::TimedOut + } else if let Some(status) = status { + classify_probe_output(&stderr, status.success()) + } else { + ProbeOutcome::LoggedOut } } +fn cleanup_contained_probe( + child: &mut std::process::Child, + guard: &mut impl ProbeTreeGuard, + context: &str, +) -> bool { + match terminate_contained_probe(child, guard) { + Ok(()) => true, + Err(error) => { + eprintln!("buzz-desktop: readiness probe cleanup failed {context}: {error}"); + false + } + } +} + +fn read_bounded_output(file: &mut std::fs::File) -> Vec { + if file.seek(SeekFrom::Start(0)).is_err() { + return Vec::new(); + } + let mut retained = Vec::new(); + let _ = file + .take(MAX_PROBE_OUTPUT_BYTES as u64) + .read_to_end(&mut retained); + retained +} + /// Classify collected probe output into a `ProbeOutcome`. /// -/// Shared between `login_probe` (which has the full `Output`) and the -/// process-level timeout path in `probe_auth_status` (which drains stderr -/// on a background thread and collects it separately). +/// Shared by the bounded process runner and classification-focused tests. pub(crate) fn classify_probe_output(stderr_bytes: &[u8], exit_success: bool) -> ProbeOutcome { if exit_success { return ProbeOutcome::LoggedIn; @@ -98,7 +503,341 @@ pub(crate) fn classify_probe_output(stderr_bytes: &[u8], exit_success: bool) -> #[cfg(test)] mod tests { - use super::{ProbeOutcome, CONFIG_PARSE_SIGNALS}; + use super::{ + LoginProbeCache, ProbeCacheKey, ProbeOutcome, ProbeTreeGuard, CONFIG_PARSE_SIGNALS, + LOGIN_PROBE_CACHE_TTL, + }; + + struct FakeProbeTreeGuard<'a> { + descendants_alive: &'a std::cell::Cell, + terminate_calls: &'a std::cell::Cell, + } + + impl ProbeTreeGuard for FakeProbeTreeGuard<'_> { + fn terminate_and_wait(&mut self) -> Result<(), String> { + self.terminate_calls.set(self.terminate_calls.get() + 1); + self.descendants_alive.set(0); + Ok(()) + } + } + + #[test] + fn readiness_timeout_tree_guard_confirms_no_probe_descendants_remain() { + let descendants_alive = std::cell::Cell::new(3); + let terminate_calls = std::cell::Cell::new(0); + let mut guard = FakeProbeTreeGuard { + descendants_alive: &descendants_alive, + terminate_calls: &terminate_calls, + }; + guard + .terminate_and_wait() + .expect("terminate fake probe job"); + assert_eq!(descendants_alive.get(), 0); + assert_eq!(terminate_calls.get(), 1); + } + + #[test] + fn windows_readiness_probe_is_suspended_until_job_assignment() { + let lifecycle_source = include_str!("../process_lifecycle.rs"); + let probe_source = include_str!("cli_probe.rs"); + let managed_node_source = include_str!("../../commands/agent_discovery/managed_node.rs"); + assert!(lifecycle_source.contains("CREATE_SUSPENDED")); + assert!(lifecycle_source.contains("AssignProcessToJobObject")); + assert!(lifecycle_source.contains("ResumeThread")); + assert!(lifecycle_source.contains("QueryInformationJobObject")); + assert!(probe_source.contains("spawn_contained_probe")); + assert!(probe_source.contains("terminate_contained_probe")); + assert!(managed_node_source.contains("spawn_probe_in_job")); + assert!(managed_node_source.contains("terminate_and_wait")); + } + + fn cache_key(runtime: &str) -> ProbeCacheKey { + ProbeCacheKey { + generation: 0, + runtime: runtime.to_string(), + binary_path: std::path::PathBuf::from(format!("/test/{runtime}")), + args: vec![runtime.to_string(), "login".into(), "status".into()], + effective_path: Some("/test/bin".into()), + effective_environment: Vec::new(), + } + } + + #[cfg(unix)] + fn executable_script(contents: &str) -> (tempfile::TempDir, std::path::PathBuf) { + use std::fs; + use std::os::unix::fs::PermissionsExt; + + let temp = tempfile::tempdir().expect("temp dir"); + let script_path = temp.path().join("probe-script"); + fs::write(&script_path, contents).expect("write script"); + fs::set_permissions(&script_path, fs::Permissions::from_mode(0o755)).expect("chmod script"); + (temp, script_path) + } + + #[cfg(unix)] + #[test] + fn login_probe_timeout_kills_and_reaps_child() { + use std::fs; + use std::time::{Duration, Instant}; + + super::clear_login_probe_cache(); + let (temp, script_path) = + executable_script("#!/bin/sh\nprintf '%s' \"$$\" > \"$1\"\nexec sleep 30\n"); + let marker = temp.path().join("pid"); + let marker_arg = marker.to_string_lossy().into_owned(); + let started_at = Instant::now(); + let outcome = super::login_probe_with_timeout( + &script_path, + &["probe-script", &marker_arg], + None, + Duration::from_millis(100), + ); + assert_eq!(outcome, ProbeOutcome::TimedOut); + assert!(started_at.elapsed() < Duration::from_secs(2)); + let pid: i32 = fs::read_to_string(marker) + .expect("pid marker") + .parse() + .expect("numeric pid"); + assert!( + nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_err(), + "probe child survived" + ); + } + + #[cfg(unix)] + #[test] + fn login_probe_drains_but_bounds_large_output() { + use std::time::{Duration, Instant}; + + let (_temp, script_path) = + executable_script("#!/bin/sh\nyes x | head -c 1048576 >&2\nexit 1\n"); + let started_at = Instant::now(); + let outcome = super::run_login_probe( + &script_path, + &["probe-script"], + None, + Duration::from_secs(2), + ); + assert_eq!(outcome, ProbeOutcome::LoggedOut); + assert!(started_at.elapsed() < Duration::from_secs(2)); + } + + #[cfg(unix)] + #[test] + fn login_probe_does_not_wait_for_descendant_held_output() { + use std::fs; + use std::time::{Duration, Instant}; + + let (temp, script_path) = + executable_script("#!/bin/sh\n(sleep 30) >&2 &\nprintf '%s' \"$!\" > \"$1\"\nexit 0\n"); + let marker = temp.path().join("descendant-pid"); + let marker_arg = marker.to_string_lossy().into_owned(); + let started_at = Instant::now(); + let outcome = super::run_login_probe( + &script_path, + &["probe-script", &marker_arg], + None, + Duration::from_secs(1), + ); + assert_eq!(outcome, ProbeOutcome::LoggedIn); + assert!(started_at.elapsed() < Duration::from_secs(2)); + + let pid: i32 = fs::read_to_string(marker) + .expect("descendant pid marker") + .parse() + .expect("numeric descendant pid"); + let deadline = Instant::now() + Duration::from_secs(1); + while nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok() + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_err(), + "probe descendant survived" + ); + } + + #[cfg(unix)] + #[test] + fn ten_equivalent_agents_share_one_authentication_probe() { + use std::sync::{atomic::AtomicUsize, Arc, Barrier}; + + let cache = Arc::new(LoginProbeCache::default()); + let start = Arc::new(Barrier::new(11)); + let calls = Arc::new(AtomicUsize::new(0)); + let mut threads = Vec::new(); + for _ in 0..10 { + let cache = Arc::clone(&cache); + let start = Arc::clone(&start); + let calls = Arc::clone(&calls); + threads.push(std::thread::spawn(move || { + start.wait(); + cache.probe(cache_key("codex"), std::time::Instant::now, || { + calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(75)); + ProbeOutcome::LoggedIn + }) + })); + } + start.wait(); + for thread in threads { + assert_eq!(thread.join().expect("probe caller"), ProbeOutcome::LoggedIn); + } + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + } + + #[test] + fn different_probe_keys_run_independently() { + use std::sync::{atomic::AtomicUsize, Arc, Barrier}; + + let cache = Arc::new(LoginProbeCache::default()); + let start = Arc::new(Barrier::new(3)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let mut threads = Vec::new(); + for runtime in ["codex", "claude"] { + let cache = Arc::clone(&cache); + let start = Arc::clone(&start); + let active = Arc::clone(&active); + let max_active = Arc::clone(&max_active); + threads.push(std::thread::spawn(move || { + start.wait(); + cache.probe(cache_key(runtime), std::time::Instant::now, || { + let count = active.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1; + max_active.fetch_max(count, std::sync::atomic::Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(100)); + active.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + ProbeOutcome::LoggedIn + }) + })); + } + start.wait(); + for thread in threads { + assert_eq!(thread.join().expect("probe caller"), ProbeOutcome::LoggedIn); + } + assert_eq!(max_active.load(std::sync::atomic::Ordering::SeqCst), 2); + } + + #[test] + fn probe_cache_expiration_and_explicit_invalidation_use_fake_time() { + use std::cell::Cell; + + let cache = LoginProbeCache::default(); + let started_at = std::time::Instant::now(); + let now = Cell::new(started_at); + let calls = Cell::new(0_u32); + let key = cache_key("codex"); + let mut run = || { + calls.set(calls.get() + 1); + ProbeOutcome::LoggedIn + }; + + assert_eq!( + cache.probe(key.clone(), || now.get(), &mut run), + ProbeOutcome::LoggedIn + ); + assert_eq!( + cache.probe(key.clone(), || now.get(), &mut run), + ProbeOutcome::LoggedIn + ); + assert_eq!(calls.get(), 1); + + now.set(started_at + LOGIN_PROBE_CACHE_TTL + std::time::Duration::from_millis(1)); + assert_eq!( + cache.probe(key.clone(), || now.get(), &mut run), + ProbeOutcome::LoggedIn + ); + assert_eq!(calls.get(), 2); + + cache.invalidate(); + assert_eq!( + cache.probe(key, || now.get(), &mut run), + ProbeOutcome::LoggedIn + ); + assert_eq!(calls.get(), 3); + } + + #[test] + fn invalidation_does_not_duplicate_an_identical_in_flight_probe() { + use std::sync::{atomic::AtomicUsize, mpsc, Arc}; + + let cache = Arc::new(LoginProbeCache::default()); + let calls = Arc::new(AtomicUsize::new(0)); + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + + let first = { + let cache = Arc::clone(&cache); + let calls = Arc::clone(&calls); + std::thread::spawn(move || { + cache.probe(cache_key("codex"), std::time::Instant::now, || { + calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + started_tx.send(()).expect("announce probe start"); + release_rx.recv().expect("release probe"); + ProbeOutcome::LoggedIn + }) + }) + }; + started_rx.recv().expect("first probe started"); + cache.invalidate(); + + let second = { + let cache = Arc::clone(&cache); + let calls = Arc::clone(&calls); + std::thread::spawn(move || { + cache.probe(cache_key("codex"), std::time::Instant::now, || { + calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + ProbeOutcome::LoggedOut + }) + }) + }; + std::thread::sleep(std::time::Duration::from_millis(25)); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + release_tx.send(()).expect("release first probe"); + assert_eq!(first.join().expect("first caller"), ProbeOutcome::LoggedIn); + assert_eq!( + second.join().expect("second caller"), + ProbeOutcome::LoggedIn + ); + + assert_eq!( + cache.probe(cache_key("codex"), std::time::Instant::now, || { + calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + ProbeOutcome::LoggedOut + }), + ProbeOutcome::LoggedOut + ); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + } + + #[test] + fn configuration_errors_expire_quickly_instead_of_being_cached_indefinitely() { + use std::cell::Cell; + + let cache = LoginProbeCache::default(); + let started_at = std::time::Instant::now(); + let now = Cell::new(started_at); + let calls = Cell::new(0_u32); + let key = cache_key("codex-invalid-config"); + let mut run = || { + calls.set(calls.get() + 1); + ProbeOutcome::ConfigInvalid { + stderr_excerpt: "invalid config".into(), + } + }; + + assert!(matches!( + cache.probe(key.clone(), || now.get(), &mut run), + ProbeOutcome::ConfigInvalid { .. } + )); + now.set(started_at + super::CONFIG_ERROR_CACHE_TTL + std::time::Duration::from_millis(1)); + assert!(matches!( + cache.probe(key, || now.get(), &mut run), + ProbeOutcome::ConfigInvalid { .. } + )); + assert_eq!(calls.get(), 2); + } #[cfg(unix)] #[test] diff --git a/desktop/src-tauri/src/managed_agents/restore.rs b/desktop/src-tauri/src/managed_agents/restore.rs index 25dadbeec6..55843c6d19 100644 --- a/desktop/src-tauri/src/managed_agents/restore.rs +++ b/desktop/src-tauri/src/managed_agents/restore.rs @@ -21,10 +21,9 @@ enum SpawnOutcome { /// Boxed: the spawned process carries its full spawn-config snapshot, so an /// inline variant would make every `Skipped`/`Failed` outcome pay for it. Spawned(super::ManagedAgentRuntimeKey, Box), - Skipped, Failed(String), } -type AgentSpawnResult = (String, SpawnOutcome); +type AgentSpawnResult = (String, String, SpawnOutcome); /// Backfill the pinned persona snapshot for pre-existing agents created before /// the record became the spawn source of truth. Runs once at launch, before @@ -276,93 +275,96 @@ pub async fn restore_managed_agents_on_launch( return Ok(()); } - // Serialize spawning and runtime registration with shutdown cleanup. The - // shutdown flag is rechecked after taking the lock so shutdown either - // prevents this transition or waits until every child is tracked and can - // be terminated. - let restore_transition = state - .managed_agent_runtime_transition - .lock() - .map_err(|error| error.to_string())?; - if shutdown_started.load(Ordering::SeqCst) { + // Phase A: reserve each absent pair under the transition/process locks, + // then release every runtime-management mutex before command discovery, + // readiness probes, or child spawning. The reservation keys serialize + // concurrent restore/reconcile/manual starts without carrying a guard. + let reserved_agents = { + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|error| error.to_string())?; + if shutdown_started.load(Ordering::SeqCst) { + return Ok(()); + } + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|error| error.to_string())?; + let workspace_relay = crate::relay::relay_ws_url_with_override(&state); + let mut reserved = Vec::new(); + for record in agents_to_start { + let relay_url = + crate::relay::effective_agent_relay_url(&record.relay_url, &workspace_relay); + let key = super::ManagedAgentRuntimeKey::new(record.pubkey.clone(), &relay_url)?; + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + continue; + } + runtimes.remove(&key); + let reservation = super::reserve_managed_agent_start(&state, &key)?; + reserved.push((record, key, reservation)); + } + reserved + }; + if reserved_agents.is_empty() { return Ok(()); } - // ── Phase B (transition lock held): resolve commands and spawn in parallel ── + // Phase B: resolve commands and spawn in parallel with all lifecycle, + // store, process-map, and reservation mutex guards released. let spawn_results: Vec = std::thread::scope(|scope| { let owner_hex_ref = owner_hex.as_deref(); - let handles: Vec<_> = agents_to_start + let handles: Vec<_> = reserved_agents .iter() .filter(|_| !shutdown_started.load(Ordering::SeqCst)) - .map(|record| { + .map(|(record, key, _reservation)| { + let pubkey = record.pubkey.clone(); + let expected_updated_at = record.updated_at.clone(); let handle = scope.spawn(move || { - let workspace_relay = - crate::relay::relay_ws_url_with_override(&app.state::()); - let relay_url = crate::relay::effective_agent_relay_url( - &record.relay_url, - &workspace_relay, - ); let outcome = - match super::ManagedAgentRuntimeKey::new(record.pubkey.clone(), &relay_url) - { - Ok(key) => { - // F2: if a concurrent startup reconcile already - // tracked a live child for this exact pair during - // the Phase A window, leave it alone. Mirrors the - // live-child guard in `start_pair`. - let already_live = app - .state::() - .managed_agent_processes - .lock() - .ok() - .and_then(|mut runtimes| { - runtimes.get_mut(&key).map(|runtime| { - runtime.child.try_wait().ok().flatten().is_none() - }) - }) - .unwrap_or(false); - if already_live { - SpawnOutcome::Skipped - } else { - match super::terminate_untracked_pair_runtime(app, &key) - .and_then(|()| { - // F1: restore spawns lazy, matching - // reconcile and manual start. Eager on - // restore buys nothing — a crashed - // mid-turn session is not resumed by an - // eager child — and silently reintroduces - // N idle brains on every launch. - spawn_agent_child( - app, - record, - &key.relay_url, - true, - owner_hex_ref, - ) - }) { - Ok(process) => { - SpawnOutcome::Spawned(key, Box::new(process)) - } - Err(error) => SpawnOutcome::Failed(error), - } - } - } - Err(error) => SpawnOutcome::Failed(error), - }; - (record.pubkey.clone(), outcome) + super::terminate_untracked_pair_runtime(app, key).and_then(|()| { + // Restore is lazy, matching reconcile/manual pair start. + spawn_agent_child(app, record, &key.relay_url, true, owner_hex_ref) + }); + let outcome = match outcome { + Ok(process) => SpawnOutcome::Spawned(key.clone(), Box::new(process)), + Err(error) => SpawnOutcome::Failed(error), + }; + (record.pubkey.clone(), record.updated_at.clone(), outcome) }); - handle + (pubkey, expected_updated_at, handle) }) .collect(); - handles.into_iter().map(|h| h.join().unwrap()).collect() + handles + .into_iter() + .map( + |(pubkey, expected_updated_at, handle)| match handle.join() { + Ok(result) => result, + Err(_) => ( + pubkey, + expected_updated_at, + SpawnOutcome::Failed("managed runtime spawn worker panicked".into()), + ), + }, + ) + .collect() }); if spawn_results.is_empty() { return Ok(()); } - // ── Phase C (re-acquire lock): write back PIDs and status to records ── + // Phase C: re-acquire briefly, generation-check every result, and register + // it. Shutdown or a concurrent record edit wins; discarded children are + // terminated after the guards are dropped. + let restore_transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|error| error.to_string())?; let _store_guard = state .managed_agents_store_lock .lock() @@ -374,28 +376,38 @@ pub async fn restore_managed_agents_on_launch( .map_err(|error| error.to_string())?; let mut successfully_spawned: Vec = Vec::new(); + let mut discarded_processes = Vec::new(); + let mut registered_keys = Vec::new(); - for (pubkey, outcome) in spawn_results { + for (pubkey, expected_updated_at, outcome) in spawn_results { match outcome { - // Skipped means a concurrent reconcile already owns a live child for - // this pair; leave its runtime and record state untouched. - SpawnOutcome::Skipped => continue, - SpawnOutcome::Spawned(key, mut process) => { + SpawnOutcome::Spawned(key, process) => { let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else { + discarded_processes.push((key, process, false)); continue; }; + if shutdown_started.load(Ordering::SeqCst) + || record.updated_at != expected_updated_at + || runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + discarded_processes.push((key, process, false)); + continue; + } + runtimes.remove(&key); let now = util::now_iso(); let receipt = super::ManagedAgentRuntimeReceipt { key: key.clone(), pid: process.child.id(), + process_identity: super::managed_process_identity(&process), desktop_instance_id: super::current_instance_id(app), started_at: now.clone(), }; if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) { - let _ = super::terminate_process(process.child.id()); - let _ = process.child.wait(); record.updated_at = now; record.last_error = Some(error); + discarded_processes.push((key, process, false)); continue; } record.updated_at = now.clone(); @@ -404,13 +416,20 @@ pub async fn restore_managed_agents_on_launch( record.last_stopped_at = None; record.last_exit_code = None; record.last_error = None; - runtimes.insert(key, super::ManagedAgentPairRuntime::starting(*process)); + runtimes.insert( + key.clone(), + super::ManagedAgentPairRuntime::starting(*process), + ); + registered_keys.push(key); successfully_spawned.push(pubkey); } SpawnOutcome::Failed(error) => { let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else { continue; }; + if record.updated_at != expected_updated_at { + continue; + } record.updated_at = util::now_iso(); record.last_error = Some(error); } @@ -448,11 +467,27 @@ pub async fn restore_managed_agents_on_launch( }) .collect(); - save_managed_agents(app, &records)?; + let save_result = save_managed_agents(app, &records); + if save_result.is_err() { + for key in registered_keys { + if let Some(runtime) = runtimes.remove(&key) { + discarded_processes.push((key, Box::new(runtime.process), true)); + } + } + } drop(runtimes); drop(_store_guard); drop(restore_transition); + for (key, mut process, wrote_receipt) in discarded_processes { + let _ = super::terminate_managed_process(&mut process); + let _ = process.child.wait(); + if wrote_receipt { + super::remove_agent_runtime_receipt(app, &key); + } + } + save_result?; + // ── Profile reconciliation (fire-and-forget) ──────────────────────────── // Spawn background tasks to ensure each restored agent's kind:0 profile is // published on the relay. Same pattern as the UI start path. diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 9fa9e0cce6..1ecbd5e813 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -26,6 +26,42 @@ pub(crate) use metadata::{ DISPLAY_NAME_ENV_VAR, SESSION_TITLE_ENV_VAR, }; +/// Exclusive intent for one managed-agent pair while its child is prepared +/// and spawned outside every runtime-management lock. +/// +/// The reservation mutex is never held beyond insertion/removal. Keeping the +/// key in the set prevents concurrent start/restore paths from creating a +/// second child during the deliberately unlocked spawn phase. +pub struct ManagedAgentStartReservation<'a> { + state: &'a crate::app_state::AppState, + key: ManagedAgentRuntimeKey, +} + +impl Drop for ManagedAgentStartReservation<'_> { + fn drop(&mut self) { + if let Ok(mut reservations) = self.state.managed_agent_start_reservations.lock() { + reservations.remove(&self.key); + } + } +} + +pub fn reserve_managed_agent_start<'a>( + state: &'a crate::app_state::AppState, + key: &ManagedAgentRuntimeKey, +) -> Result, String> { + let mut reservations = state + .managed_agent_start_reservations + .lock() + .map_err(|error| error.to_string())?; + if !reservations.insert(key.clone()) { + return Err("managed-agent runtime start is already in progress".into()); + } + Ok(ManagedAgentStartReservation { + state, + key: key.clone(), + }) +} + mod stop; pub(crate) use stop::managed_agent_runtime_keys; pub use stop::{stop_managed_agent_process, stop_managed_agent_workspace_pair}; @@ -36,14 +72,23 @@ pub(crate) use sweep::sweep_untracked_bundle_harnesses; type RespondToEnv = (Vec<(&'static str, String)>, Vec<&'static str>); mod process; +#[cfg(any(windows, test))] +pub(crate) use process::descendant_process_waves; #[cfg(test)] use process::{ buzz_marker_entry, name_matches_interpreter, name_matches_known_binary, terminate_runtime_receipt_with, valid_agent_runtime_receipt_with, }; pub(crate) use process::{ - current_instance_id, process_belongs_to_us, process_has_buzz_marker, process_is_running, - terminate_process, terminate_untracked_pair_runtime, valid_agent_runtime_receipt, + current_instance_id, managed_process_identity, process_belongs_to_us, process_has_buzz_marker, + process_is_running, terminate_managed_process, terminate_process, + terminate_untracked_pair_runtime, valid_agent_runtime_receipt, +}; +#[cfg(windows)] +pub(crate) use process::{ + next_verified_windows_descendants, terminate_if_windows_identity_matches, + terminate_process_with_identity, WindowsIdentityObservation, WindowsProcessIdentity, + WindowsProcessSnapshotEntry, }; mod orphan_sweep; @@ -932,7 +977,7 @@ pub fn spawn_agent_child( // Windows: assign the harness to a Job Object so its whole tree dies with // the handle. The Unix process-group equivalent is set above. #[cfg(windows)] - return Ok(super::process_lifecycle::finish_spawn( + return super::process_lifecycle::finish_spawn( child, log_path, spawn_config, @@ -940,7 +985,7 @@ pub fn spawn_agent_child( spawned_adapter_availability, start_nonce, &record.name, - )); + ); #[cfg(not(windows))] Ok(crate::managed_agents::ManagedAgentProcess { child, @@ -992,20 +1037,8 @@ pub fn start_managed_agent_process( // Scalar PIDs are migration-only and never establish pair liveness. record.runtime_pid = None; - let mut process = spawn_agent_child(app, record, &key.relay_url, false, owner_hex)?; + let process = spawn_agent_child(app, record, &key.relay_url, false, owner_hex)?; let now = now_iso(); - let receipt = super::ManagedAgentRuntimeReceipt { - key: key.clone(), - pid: process.child.id(), - desktop_instance_id: current_instance_id(app), - started_at: now.clone(), - }; - if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) { - let _ = terminate_process(process.child.id()); - let _ = process.child.wait(); - return Err(error); - } - record.updated_at = now.clone(); record.last_started_at = Some(now); record.last_stopped_at = None; diff --git a/desktop/src-tauri/src/managed_agents/runtime/process.rs b/desktop/src-tauri/src/managed_agents/runtime/process.rs index 37eb5659a4..d2c81e9f86 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/process.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/process.rs @@ -30,6 +30,120 @@ pub(crate) const KNOWN_AGENT_BINARIES: &[&str] = &[ /// `process_has_buzz_marker()`). This avoids sweeping unrelated node processes. pub(crate) const KNOWN_SCRIPT_INTERPRETERS: &[&str] = &["node"]; +/// Expand a process snapshot into descendant generations rooted at `root`. +/// Kept platform-independent so the Windows recovery teardown topology can be +/// unit-tested on every CI host. +#[cfg(any(windows, test))] +pub(crate) fn descendant_process_waves(entries: &[(u32, u32)], root: u32) -> Vec> { + let mut known = std::collections::HashSet::from([root]); + let mut waves = Vec::new(); + loop { + let mut wave_seen = std::collections::HashSet::new(); + let wave = entries + .iter() + .filter_map(|(pid, parent)| { + (*pid != root + && !known.contains(pid) + && known.contains(parent) + && wave_seen.insert(*pid)) + .then_some(*pid) + }) + .collect::>(); + if wave.is_empty() { + break; + } + known.extend(wave.iter().copied()); + waves.push(wave); + } + waves +} + +/// Stable identity for one Windows process instance. Windows may recycle a +/// numeric PID as soon as the prior process object is released, while the +/// creation time remains unique for the lifetime of that process instance. +#[cfg(any(windows, test))] +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub(crate) struct WindowsProcessIdentity { + pub(crate) pid: u32, + pub(crate) creation_time: u64, +} + +/// Identity observation made immediately before a destructive Windows action. +/// `Unverified` is deliberately distinct from `Exited`: callers must fail +/// closed when access is denied or metadata cannot be read. +#[cfg(any(windows, test))] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum WindowsIdentityObservation { + Verified(WindowsProcessIdentity), + Exited, + Unverified, +} + +/// Run `terminate` only when the currently opened Windows process object still +/// has the expected creation identity. This small platform-independent seam is +/// used by the native implementation and makes the no-kill-on-reuse rule +/// directly testable on non-Windows CI. +#[cfg(any(windows, test))] +pub(crate) fn terminate_if_windows_identity_matches( + expected: WindowsProcessIdentity, + observed: WindowsIdentityObservation, + terminate: impl FnOnce() -> Result<(), String>, +) -> Result { + match observed { + WindowsIdentityObservation::Verified(actual) if actual == expected => { + terminate()?; + Ok(true) + } + WindowsIdentityObservation::Exited => Ok(false), + WindowsIdentityObservation::Verified(actual) => Err(format!( + "refusing to terminate recycled Windows PID {}: expected creation {}, observed {}", + expected.pid, expected.creation_time, actual.creation_time + )), + WindowsIdentityObservation::Unverified => Err(format!( + "refusing to terminate Windows PID {} because process identity could not be verified", + expected.pid + )), + } +} + +/// One identity-bearing ToolHelp row. `identity: None` means the PID was +/// visible but its stable identity could not be queried; such a row can be +/// reported as a bounded cleanup failure but must never become a kill target. +#[cfg(any(windows, test))] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct WindowsProcessSnapshotEntry { + pub(crate) pid: u32, + pub(crate) parent_pid: u32, + pub(crate) identity: Option, +} + +/// Find the next direct descendant generation whose parent is already bound +/// to a verified, open process object. Existing identities are never replaced +/// by a same-PID/different-creation-time row. +#[cfg(any(windows, test))] +pub(crate) fn next_verified_windows_descendants( + entries: &[WindowsProcessSnapshotEntry], + known: &std::collections::HashMap, +) -> (Vec, Vec) { + let mut verified = Vec::new(); + let mut unverified = Vec::new(); + let mut seen = std::collections::HashSet::new(); + for entry in entries { + if entry.pid == 0 + || known.contains_key(&entry.pid) + || !known.contains_key(&entry.parent_pid) + || !seen.insert(entry.pid) + { + continue; + } + match entry.identity { + Some(identity) if identity.pid == entry.pid => verified.push(identity), + _ => unverified.push(entry.pid), + } + } + (verified, unverified) +} + /// Check if a process name matches any of our known agent binaries. /// Uses exact match or prefix-with-separator to avoid false positives /// (e.g. `"goose"` must not match `"mongoose"`). @@ -262,10 +376,17 @@ pub(crate) fn terminate_process(pid: u32) -> Result<(), String> { #[cfg(windows)] pub(crate) fn terminate_process(pid: u32) -> Result<(), String> { - // No job handle is available on this path (e.g. after an app restart, when - // we only recovered the PID from the record), so fall back to taskkill on - // the whole tree. - super::super::process_lifecycle::taskkill_tree(pid) + Err(format!( + "refusing to terminate Windows process tree {pid} without a stable process identity" + )) +} + +#[cfg(windows)] +pub(crate) fn terminate_process_with_identity( + pid: u32, + process_identity: Option, +) -> Result<(), String> { + super::super::process_lifecycle::terminate_process_tree(pid, process_identity) } #[cfg(not(any(unix, windows)))] @@ -273,6 +394,45 @@ pub(crate) fn terminate_process(_pid: u32) -> Result<(), String> { Err("managed agent shutdown after app restart is not supported on this platform".to_string()) } +pub(crate) fn managed_process_identity(process: &super::super::ManagedAgentProcess) -> Option { + #[cfg(windows)] + { + return Some(process.process_identity); + } + #[cfg(not(windows))] + { + let _ = process; + None + } +} + +/// Terminate a process that is still represented by its stable `Child` +/// generation. Windows prefers the existing Job Object; if assignment failed, +/// native fallback remains bound to the creation identity captured at spawn. +pub(crate) fn terminate_managed_process( + process: &mut super::super::ManagedAgentProcess, +) -> Result<(), String> { + #[cfg(unix)] + { + terminate_process(process.child.id()) + } + #[cfg(windows)] + { + if let Some(job) = process.job.take() { + job.terminate_and_wait(std::time::Duration::from_secs(1)) + } else { + terminate_process_with_identity(process.child.id(), Some(process.process_identity)) + } + } + #[cfg(not(any(unix, windows)))] + { + process + .child + .kill() + .map_err(|error| format!("failed to kill managed process: {error}")) + } +} + /// Send SIGTERM to all given PIDs (as process groups), wait, then SIGKILL /// any survivors. Uses `-pid` to kill the entire process group — if an /// orphaned agent called `setsid()`, it IS the group leader, so this @@ -384,6 +544,22 @@ pub(crate) fn valid_agent_runtime_receipt( receipt: &super::super::ManagedAgentRuntimeReceipt, instance_id: &str, ) -> bool { + #[cfg(windows)] + { + let Ok(canonical) = + ManagedAgentRuntimeKey::new(receipt.key.pubkey.clone(), &receipt.key.relay_url) + else { + return false; + }; + return canonical == receipt.key + && path.file_name().and_then(|name| name.to_str()) + == Some(&format!("{}.json", receipt.key.runtime_id())) + && receipt.desktop_instance_id == instance_id + && receipt.process_identity.is_some_and(|identity| { + super::super::process_lifecycle::process_identity_matches(receipt.pid, identity) + }); + } + #[cfg(not(windows))] valid_agent_runtime_receipt_with( path, receipt, @@ -462,8 +638,162 @@ pub(crate) fn terminate_untracked_pair_runtime( terminate_runtime_receipt_with( &path, &receipt, - terminate_process, - process_is_running, + |pid| { + #[cfg(windows)] + { + terminate_process_with_identity(pid, receipt.process_identity) + } + #[cfg(not(windows))] + { + terminate_process(pid) + } + }, + |pid| { + #[cfg(windows)] + { + receipt.process_identity.is_some_and(|identity| { + super::super::process_lifecycle::process_identity_matches(pid, identity) + }) + } + #[cfg(not(windows))] + { + process_is_running(pid) + } + }, super::super::remove_agent_runtime_receipt_path, ) } + +#[cfg(test)] +mod windows_identity_tests { + use super::{ + next_verified_windows_descendants, terminate_if_windows_identity_matches, + WindowsIdentityObservation, WindowsProcessIdentity, WindowsProcessSnapshotEntry, + }; + use std::{cell::Cell, collections::HashMap}; + + fn identity(pid: u32, creation_time: u64) -> WindowsProcessIdentity { + WindowsProcessIdentity { pid, creation_time } + } + + #[test] + fn root_pid_reuse_never_invokes_termination() { + let calls = Cell::new(0); + let result = terminate_if_windows_identity_matches( + identity(10, 100), + WindowsIdentityObservation::Verified(identity(10, 200)), + || { + calls.set(calls.get() + 1); + Ok(()) + }, + ); + assert!(result.is_err()); + assert_eq!(calls.get(), 0); + } + + #[test] + fn descendant_pid_reuse_is_not_rediscovered_or_terminated() { + let root = identity(10, 100); + let child = identity(11, 110); + let mut known = HashMap::from([(root.pid, root)]); + let first = [WindowsProcessSnapshotEntry { + pid: child.pid, + parent_pid: root.pid, + identity: Some(child), + }]; + let (discovered, unverified) = next_verified_windows_descendants(&first, &known); + assert_eq!(discovered, vec![child]); + assert!(unverified.is_empty()); + known.insert(child.pid, child); + + let replacement = identity(child.pid, 999); + let second = [WindowsProcessSnapshotEntry { + pid: replacement.pid, + parent_pid: root.pid, + identity: Some(replacement), + }]; + let (discovered, unverified) = next_verified_windows_descendants(&second, &known); + assert!(discovered.is_empty()); + assert!(unverified.is_empty()); + + let calls = Cell::new(0); + assert!(terminate_if_windows_identity_matches( + child, + WindowsIdentityObservation::Verified(replacement), + || { + calls.set(calls.get() + 1); + Ok(()) + }, + ) + .is_err()); + assert_eq!(calls.get(), 0); + } + + #[test] + fn descendants_created_during_teardown_are_discovered_by_generation() { + let root = identity(10, 100); + let child = identity(11, 110); + let grandchild = identity(12, 120); + let mut known = HashMap::from([(root.pid, root)]); + let first = [WindowsProcessSnapshotEntry { + pid: child.pid, + parent_pid: root.pid, + identity: Some(child), + }]; + let (first_wave, _) = next_verified_windows_descendants(&first, &known); + assert_eq!(first_wave, vec![child]); + known.insert(child.pid, child); + + let later = [WindowsProcessSnapshotEntry { + pid: grandchild.pid, + parent_pid: child.pid, + identity: Some(grandchild), + }]; + let (second_wave, _) = next_verified_windows_descendants(&later, &known); + assert_eq!(second_wave, vec![grandchild]); + } + + #[test] + fn natural_exit_race_is_harmless_and_idempotent() { + let calls = Cell::new(0); + for _ in 0..2 { + assert_eq!( + terminate_if_windows_identity_matches( + identity(10, 100), + WindowsIdentityObservation::Exited, + || { + calls.set(calls.get() + 1); + Ok(()) + }, + ), + Ok(false) + ); + } + assert_eq!(calls.get(), 0); + } + + #[test] + fn unverifiable_identity_never_invokes_termination() { + let calls = Cell::new(0); + let result = terminate_if_windows_identity_matches( + identity(10, 100), + WindowsIdentityObservation::Unverified, + || { + calls.set(calls.get() + 1); + Ok(()) + }, + ); + assert!(result.is_err()); + assert_eq!(calls.get(), 0); + + let known = HashMap::from([(10, identity(10, 100))]); + let entries = [WindowsProcessSnapshotEntry { + pid: 11, + parent_pid: 10, + identity: None, + }]; + let (verified, unverified) = next_verified_windows_descendants(&entries, &known); + assert!(verified.is_empty()); + assert_eq!(unverified, vec![11]); + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/stop.rs b/desktop/src-tauri/src/managed_agents/runtime/stop.rs index 08bca15feb..67fabf461c 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/stop.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/stop.rs @@ -4,8 +4,8 @@ use tauri::AppHandle; use super::{ append_log_marker, current_instance_id, now_iso, process_belongs_to_us, - process_has_buzz_marker, process_is_running, terminate_process, ManagedAgentPairRuntime, - ManagedAgentRecord, ManagedAgentRuntimeKey, + process_has_buzz_marker, process_is_running, terminate_managed_process, terminate_process, + ManagedAgentPairRuntime, ManagedAgentRecord, ManagedAgentRuntimeKey, }; pub(crate) fn managed_agent_runtime_keys( @@ -47,21 +47,7 @@ fn stop_managed_agent_pair( return Ok(()); }; let result = (|| -> Result<(), String> { - #[cfg(unix)] - terminate_process(runtime.child.id())?; - #[cfg(windows)] - match runtime.job.take() { - Some(job) => drop(job), - None => runtime - .child - .kill() - .map_err(|error| format!("failed to kill agent process: {error}"))?, - } - #[cfg(not(any(unix, windows)))] - runtime - .child - .kill() - .map_err(|error| format!("failed to kill agent process: {error}"))?; + terminate_managed_process(&mut runtime.process)?; let status = runtime .child .wait() diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index bea4b1c3e3..a438663a47 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -904,6 +904,7 @@ fn receipt_fixture( crate::managed_agents::ManagedAgentRuntimeReceipt { key, pid: std::process::id(), + process_identity: None, desktop_instance_id: "test-instance".into(), started_at: "now".into(), } @@ -1282,6 +1283,8 @@ fn make_pair_runtime_placeholder() -> crate::managed_agents::ManagedAgentPairRun adapter_availability: None, start_nonce: "test-nonce".to_string(), #[cfg(windows)] + process_identity: 0, + #[cfg(windows)] job: None, }; crate::managed_agents::ManagedAgentPairRuntime::starting(process) diff --git a/desktop/src-tauri/src/managed_agents/runtime_commands.rs b/desktop/src-tauri/src/managed_agents/runtime_commands.rs index c0e55184b1..b317a5180c 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_commands.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_commands.rs @@ -1,40 +1,82 @@ -use std::sync::atomic::Ordering; +use std::sync::{ + atomic::{AtomicU64, Ordering}, + Arc, OnceLock, +}; use tauri::{AppHandle, Emitter, Manager}; use super::{ agent_readiness, append_log_marker, current_instance_id, find_managed_agent_mut, load_global_agent_config, load_managed_agents, load_personas, managed_agent_runtime_log_path, - process_is_running, record_agent_command, resolve_effective_agent_env, save_managed_agents, - spawn_agent_child, terminate_process, terminate_untracked_pair_runtime, - write_agent_runtime_receipt, AgentReadiness, BackendKind, ManagedAgentPairRuntime, - ManagedAgentRuntimeKey, ManagedAgentRuntimeLifecycle, ManagedAgentRuntimeReceipt, - ManagedAgentRuntimeStatus, + managed_process_identity, record_agent_command, reserve_managed_agent_start, + resolve_effective_agent_env, save_managed_agents, spawn_agent_child, terminate_managed_process, + terminate_untracked_pair_runtime, write_agent_runtime_receipt, AgentReadiness, BackendKind, + ManagedAgentPairRuntime, ManagedAgentRuntimeKey, ManagedAgentRuntimeLifecycle, + ManagedAgentRuntimeReceipt, ManagedAgentRuntimeStatus, }; use crate::app_state::AppState; const STATUS_EVENT: &str = "managed-agent-runtime-status"; -fn status_for( - app: &AppHandle, - record: &super::ManagedAgentRecord, - key: &ManagedAgentRuntimeKey, - runtime: Option<&ManagedAgentPairRuntime>, - requested_relay_url: Option, -) -> ManagedAgentRuntimeStatus { - let personas = load_personas(app).unwrap_or_default(); - let global = load_global_agent_config(app).unwrap_or_default(); - status_for_with( - app, - record, - key, - runtime, - requested_relay_url, - StatusInputs { - personas: &personas, - global: &global, - }, - ) +type RuntimeListResult = Result, String>; + +#[derive(Clone)] +struct RuntimeListFlight { + id: u64, + result: tokio::sync::watch::Receiver>, +} + +#[derive(Default)] +struct RuntimeListSingleFlight { + next_id: AtomicU64, + current: tokio::sync::Mutex>, +} + +impl RuntimeListSingleFlight { + async fn run(self: &Arc, compute: F) -> RuntimeListResult + where + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future + Send + 'static, + { + let mut receiver = { + let mut current = self.current.lock().await; + if let Some(flight) = current.as_ref() { + flight.result.clone() + } else { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let (sender, receiver) = tokio::sync::watch::channel(None); + *current = Some(RuntimeListFlight { + id, + result: receiver.clone(), + }); + let coordinator = Arc::clone(self); + tauri::async_runtime::spawn(async move { + let result = compute().await; + let _ = sender.send(Some(result)); + let mut current = coordinator.current.lock().await; + if current.as_ref().is_some_and(|flight| flight.id == id) { + *current = None; + } + }); + receiver + } + }; + + loop { + if let Some(result) = receiver.borrow().clone() { + return result; + } + receiver + .changed() + .await + .map_err(|_| "managed runtime status worker stopped unexpectedly".to_string())?; + } + } +} + +fn runtime_list_single_flight() -> &'static Arc { + static SINGLE_FLIGHT: OnceLock> = OnceLock::new(); + SINGLE_FLIGHT.get_or_init(|| Arc::new(RuntimeListSingleFlight::default())) } /// Preloaded per-call-site inputs for [`status_for_with`], so multi-row @@ -52,11 +94,37 @@ fn status_for_with( requested_relay_url: Option, inputs: StatusInputs<'_>, ) -> ManagedAgentRuntimeStatus { + let StatusInputs { personas, global } = inputs; + let local_setup = local_setup_for(record, StatusInputs { personas, global }); + status_for_with_local_setup(app, key, runtime, requested_relay_url, local_setup) +} + +fn local_setup_for(record: &super::ManagedAgentRecord, inputs: StatusInputs<'_>) -> bool { + local_setup_for_with(record, inputs, agent_readiness) +} + +fn local_setup_for_with( + record: &super::ManagedAgentRecord, + inputs: StatusInputs<'_>, + evaluate_readiness: F, +) -> bool +where + F: FnOnce(&super::readiness::EffectiveAgentEnv) -> AgentReadiness, +{ let StatusInputs { personas, global } = inputs; let command = record_agent_command(record, personas); let metadata = super::known_acp_runtime(&command); let effective = resolve_effective_agent_env(record, personas, metadata, global); - let local_setup = matches!(agent_readiness(&effective), AgentReadiness::Ready); + matches!(evaluate_readiness(&effective), AgentReadiness::Ready) +} + +fn status_for_with_local_setup( + app: &AppHandle, + key: &ManagedAgentRuntimeKey, + runtime: Option<&ManagedAgentPairRuntime>, + requested_relay_url: Option, + local_setup: bool, +) -> ManagedAgentRuntimeStatus { ManagedAgentRuntimeStatus { pubkey: key.pubkey.clone(), relay_url: key.relay_url.clone(), @@ -112,6 +180,15 @@ pub fn put_managed_agent_runtime_lifecycle( .iter() .find(|record| record.pubkey.eq_ignore_ascii_case(&key.pubkey)) .ok_or_else(|| format!("agent {} not found", key.pubkey))?; + let personas = load_personas(&app).unwrap_or_default(); + let global = load_global_agent_config(&app).unwrap_or_default(); + let local_setup = local_setup_for( + record, + StatusInputs { + personas: &personas, + global: &global, + }, + ); let mut runtimes = state .managed_agent_processes .lock() @@ -132,90 +209,271 @@ pub fn put_managed_agent_runtime_lifecycle( } runtime.lifecycle = payload.lifecycle; runtime.error = payload.error; - let status = status_for(&app, record, &key, Some(runtime), None); + let status = status_for_with_local_setup(&app, &key, Some(runtime), None, local_setup); emit_status(&app, &status); Ok(status) } #[tauri::command] -pub fn list_managed_agent_runtimes( +pub async fn list_managed_agent_runtimes( app: AppHandle, ) -> Result, String> { - // This command is polled whenever the members sidebar opens and refetched - // on every status event — load the per-row status inputs once, outside - // the locks, instead of hitting disk per row while holding them. - let personas = load_personas(&app).unwrap_or_default(); - let global = load_global_agent_config(&app).unwrap_or_default(); - let state = app.state::(); + // Runtime status is polled frequently by the desktop. The implementation + // reads agent configuration, inspects child processes, and may update the + // managed-agent store, so keep that blocking work off Tauri's IPC thread. + runtime_list_single_flight() + .run(move || async move { + tauri::async_runtime::spawn_blocking(move || list_managed_agent_runtimes_blocking(app)) + .await + .map_err(|error| format!("managed runtime status worker failed: {error}"))? + }) + .await +} + +#[derive(Clone)] +struct RuntimeProcessSnapshot { + key: ManagedAgentRuntimeKey, + lifecycle: ManagedAgentRuntimeLifecycle, + tracked_pid: u32, + status_pid: Option, + error: Option, + start_nonce: String, + exited: bool, + exit_code: Option, +} + +#[derive(Clone)] +struct RuntimeStatusSnapshot { + process: RuntimeProcessSnapshot, + record: super::ManagedAgentRecord, +} + +fn status_for_snapshot( + app: &AppHandle, + snapshot: &RuntimeStatusSnapshot, + inputs: StatusInputs<'_>, +) -> ManagedAgentRuntimeStatus { + let local_setup = local_setup_for(&snapshot.record, inputs); + ManagedAgentRuntimeStatus { + pubkey: snapshot.process.key.pubkey.clone(), + relay_url: snapshot.process.key.relay_url.clone(), + requested_relay_url: None, + local_setup, + lifecycle: snapshot.process.lifecycle.clone(), + pid: snapshot.process.status_pid, + error: snapshot.process.error.clone(), + log_path: managed_agent_runtime_log_path(app, &snapshot.process.key) + .ok() + .map(|path| path.display().to_string()), + } +} + +fn collect_runtime_process_snapshots( + state: &AppState, +) -> Result<(u64, Vec), String> { + // Phase 1: hold the runtime-management locks only long enough to inspect + // child generations and copy immutable scalar state. No file or command + // discovery and no authentication process is allowed in this scope. let _transition = state .managed_agent_runtime_transition .lock() - .map_err(|e| e.to_string())?; + .map_err(|error| error.to_string())?; let _store = state .managed_agents_store_lock .lock() - .map_err(|e| e.to_string())?; - let mut records = load_managed_agents(&app)?; + .map_err(|error| error.to_string())?; + let store_generation = super::managed_agent_store_generation(); let mut runtimes = state .managed_agent_processes .lock() - .map_err(|e| e.to_string())?; - let exited_keys: Vec<_> = runtimes + .map_err(|error| error.to_string())?; + let snapshots = runtimes .iter_mut() - .filter_map(|(key, runtime)| match runtime.child.try_wait() { - Ok(Some(_)) | Err(_) => Some(key.clone()), - Ok(None) => None, + .map(|(key, runtime)| { + let tracked_pid = runtime.child.id(); + let (exited, exit_code, error) = match runtime.child.try_wait() { + Ok(Some(status)) => ( + true, + status.code(), + Some(format!( + "managed agent runtime exited unexpectedly ({status})" + )), + ), + Err(error) => ( + true, + None, + Some(format!("failed to inspect managed agent runtime: {error}")), + ), + Ok(None) => (false, None, runtime.error.clone()), + }; + RuntimeProcessSnapshot { + key: key.clone(), + lifecycle: if exited { + ManagedAgentRuntimeLifecycle::Failed + } else { + runtime.lifecycle.clone() + }, + tracked_pid, + status_pid: (!exited).then_some(tracked_pid), + error, + start_nonce: runtime.start_nonce.clone(), + exited, + exit_code, + } }) .collect(); - let records_changed = !exited_keys.is_empty(); - let mut statuses = Vec::new(); - for key in exited_keys { - runtimes.remove(&key); - super::remove_agent_runtime_receipt(&app, &key); - state.clear_agent_session_cache(&key); + Ok((store_generation, snapshots)) +} + +fn ensure_store_generation_unchanged( + expected: u64, + current: u64, + context: &str, +) -> Result<(), String> { + if current == expected { + Ok(()) + } else { + Err(format!("managed-agent store changed {context}")) + } +} + +fn runtime_generation_matches( + expected_nonce: &str, + expected_pid: u32, + live_nonce: &str, + live_pid: u32, + live_exited: bool, +) -> bool { + expected_nonce == live_nonce && expected_pid == live_pid && live_exited +} + +fn persist_exited_runtime_snapshots( + app: &AppHandle, + state: &AppState, + store_generation: u64, + snapshots: &[RuntimeStatusSnapshot], + records: &mut [super::ManagedAgentRecord], +) -> Result<(), String> { + let exited: Vec<_> = snapshots + .iter() + .filter(|snapshot| snapshot.process.exited) + .collect(); + if exited.is_empty() { + return Ok(()); + } + + // Phase 4: only exited-runtime persistence reacquires the locks. Verify + // both the store and process generation before applying delayed results. + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|error| error.to_string())?; + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + ensure_store_generation_unchanged( + store_generation, + super::managed_agent_store_generation(), + "while status probes were in flight", + )?; + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|error| error.to_string())?; + for snapshot in &exited { + let process = &snapshot.process; + let unchanged = runtimes.get_mut(&process.key).is_some_and(|runtime| { + let live_nonce = runtime.start_nonce.clone(); + let live_pid = runtime.child.id(); + let live_exited = !matches!(runtime.child.try_wait(), Ok(None)); + runtime_generation_matches( + &process.start_nonce, + process.tracked_pid, + &live_nonce, + live_pid, + live_exited, + ) + }); + if !unchanged { + return Err("managed runtime changed while status probes were in flight".into()); + } + } + + for snapshot in &exited { + let process = &snapshot.process; if let Some(record) = records .iter_mut() - .find(|record| record.pubkey.eq_ignore_ascii_case(&key.pubkey)) + .find(|record| record.pubkey.eq_ignore_ascii_case(&process.key.pubkey)) { - record.updated_at = crate::util::now_iso(); - record.last_stopped_at = Some(record.updated_at.clone()); - let status = status_for_with( + record.updated_at = snapshot.record.updated_at.clone(); + record.last_stopped_at = snapshot.record.last_stopped_at.clone(); + record.last_exit_code = process.exit_code; + record.last_error = process.error.clone(); + record.last_error_code = process.exit_code.map(i64::from); + } + } + save_managed_agents(app, records)?; + for snapshot in exited { + runtimes.remove(&snapshot.process.key); + super::remove_agent_runtime_receipt(app, &snapshot.process.key); + state.clear_agent_session_cache(&snapshot.process.key); + } + Ok(()) +} + +fn list_managed_agent_runtimes_blocking( + app: AppHandle, +) -> Result, String> { + let state = app.state::(); + let (store_generation, process_snapshots) = collect_runtime_process_snapshots(&state)?; + + // Phase 3: all filesystem reads, command discovery, and authentication + // probes happen after every runtime-management lock has been released. + let personas = load_personas(&app).unwrap_or_default(); + let global = load_global_agent_config(&app).unwrap_or_default(); + let mut records = load_managed_agents(&app)?; + ensure_store_generation_unchanged( + store_generation, + super::managed_agent_store_generation(), + "during status snapshot", + )?; + let snapshots: Vec<_> = process_snapshots + .into_iter() + .filter_map(|process| { + let record = records + .iter() + .find(|record| record.pubkey.eq_ignore_ascii_case(&process.key.pubkey))?; + let mut record = record.clone(); + if process.exited { + record.updated_at = crate::util::now_iso(); + record.last_stopped_at = Some(record.updated_at.clone()); + } + Some(RuntimeStatusSnapshot { process, record }) + }) + .collect(); + + let statuses: Vec<_> = snapshots + .iter() + .map(|snapshot| { + status_for_snapshot( &app, - record, - &key, - None, - None, + snapshot, StatusInputs { personas: &personas, global: &global, }, - ); - emit_status(&app, &status); - statuses.push(status); + ) + }) + .collect(); + + persist_exited_runtime_snapshots(&app, &state, store_generation, &snapshots, &mut records)?; + + for (snapshot, status) in snapshots.iter().zip(&statuses) { + if snapshot.process.exited { + emit_status(&app, status); } } - statuses.extend(runtimes.iter().filter_map(|(key, runtime)| { - let record = records - .iter() - .find(|record| record.pubkey.eq_ignore_ascii_case(&key.pubkey))?; - Some(status_for_with( - &app, - record, - key, - Some(runtime), - None, - StatusInputs { - personas: &personas, - global: &global, - }, - )) - })); - drop(runtimes); - // Records are only mutated above when a runtime exited — skip the store - // rewrite on the common nothing-changed poll. - if records_changed { - save_managed_agents(&app, &records)?; - } Ok(statuses) } @@ -243,70 +501,179 @@ fn start_pair( expected_updated_at: Option<&str>, app: AppHandle, ) -> Result { - let state = app.state::(); - let _transition = state - .managed_agent_runtime_transition - .lock() - .map_err(|e| e.to_string())?; - if state.shutdown_started.load(Ordering::Acquire) { - return Err("desktop shutdown has started".into()); - } - let _store = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let mut records = load_managed_agents(&app)?; - let record = find_managed_agent_mut(&mut records, &pubkey)?; - if record.backend != BackendKind::Local { - return Err("managed runtime pairs require a local agent".into()); - } - if expected_updated_at.is_some_and(|expected| record.updated_at != expected) { - return Err("managed agent changed while runtime reconciliation was in flight".into()); - } + let personas = load_personas(&app).unwrap_or_default(); + let global = load_global_agent_config(&app).unwrap_or_default(); + let readiness_records = load_managed_agents(&app)?; + let readiness_record = readiness_records + .iter() + .find(|record| record.pubkey.eq_ignore_ascii_case(&pubkey)) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + let readiness_updated_at = readiness_record.updated_at.clone(); + // Authentication readiness may launch a CLI. Resolve it before runtime + // locks, then fence the result against the record generation below. + let local_setup = local_setup_for( + readiness_record, + StatusInputs { + personas: &personas, + global: &global, + }, + ); let key = ManagedAgentRuntimeKey::new(pubkey, &relay_url)?; - let mut runtimes = state - .managed_agent_processes - .lock() - .map_err(|e| e.to_string())?; - if runtimes - .get_mut(&key) - .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) - { - let status = status_for(&app, record, &key, runtimes.get(&key), None); - return Ok(status); - } - runtimes.remove(&key); - terminate_untracked_pair_runtime(&app, &key)?; - + let state = app.state::(); let owner = state .keys .lock() .ok() .map(|keys| keys.public_key().to_hex()); - let mut process = spawn_agent_child(&app, record, &key.relay_url, lazy, owner.as_deref())?; - let now = crate::util::now_iso(); - let receipt = ManagedAgentRuntimeReceipt { - key: key.clone(), - pid: process.child.id(), - desktop_instance_id: current_instance_id(&app), - started_at: now.clone(), + + // Phase A: validate and reserve this exact pair while holding locks only + // long enough to copy the immutable record snapshot. The reservation, not + // a held mutex, prevents a concurrent start from spawning a duplicate. + let (record_snapshot, _reservation) = { + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + if state.shutdown_started.load(Ordering::Acquire) { + return Err("desktop shutdown has started".into()); + } + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(&app)?; + let record = records + .iter() + .find(|record| record.pubkey.eq_ignore_ascii_case(&readiness_record.pubkey)) + .ok_or_else(|| format!("agent {} not found", readiness_record.pubkey))?; + if record.updated_at != readiness_updated_at { + return Err("managed agent changed while readiness was in flight".into()); + } + if record.backend != BackendKind::Local { + return Err("managed runtime pairs require a local agent".into()); + } + if expected_updated_at.is_some_and(|expected| record.updated_at != expected) { + return Err("managed agent changed while runtime reconciliation was in flight".into()); + } + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + return Ok(status_for_with_local_setup( + &app, + &key, + runtimes.get(&key), + None, + local_setup, + )); + } + runtimes.remove(&key); + let reservation = reserve_managed_agent_start(&state, &key)?; + (record.clone(), reservation) }; - if let Err(error) = write_agent_runtime_receipt(&app, &receipt) { - let _ = terminate_process(process.child.id()); + + // Phase B: command discovery, readiness, log setup, and both the login + // probe and buzz-acp spawn happen with every runtime lock released. + terminate_untracked_pair_runtime(&app, &key)?; + let spawned = spawn_agent_child( + &app, + &record_snapshot, + &key.relay_url, + lazy, + owner.as_deref(), + )?; + let mut spawned = Some(spawned); + let mut wrote_receipt = false; + + // Phase C: generation-fence the unlocked result and register it briefly. + // Any shutdown, record edit, or competing live runtime wins; the newly + // spawned child is then terminated only after these guards are dropped. + let registration = (|| -> Result { + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + if state.shutdown_started.load(Ordering::Acquire) { + return Err("desktop shutdown started while managed runtime was spawning".into()); + } + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let mut records = load_managed_agents(&app)?; + let record = find_managed_agent_mut(&mut records, &record_snapshot.pubkey)?; + if record.updated_at != record_snapshot.updated_at { + return Err("managed agent changed while runtime was spawning".into()); + } + if record.backend != BackendKind::Local { + return Err("managed runtime pairs require a local agent".into()); + } + if expected_updated_at.is_some_and(|expected| record.updated_at != expected) { + return Err("managed agent changed while runtime reconciliation was in flight".into()); + } + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + return Ok(status_for_with_local_setup( + &app, + &key, + runtimes.get(&key), + None, + local_setup, + )); + } + runtimes.remove(&key); + + let process = spawned + .as_ref() + .ok_or_else(|| "managed runtime spawn result was already consumed".to_string())?; + let now = crate::util::now_iso(); + let receipt = ManagedAgentRuntimeReceipt { + key: key.clone(), + pid: process.child.id(), + process_identity: managed_process_identity(process), + desktop_instance_id: current_instance_id(&app), + started_at: now.clone(), + }; + write_agent_runtime_receipt(&app, &receipt)?; + wrote_receipt = true; + record.runtime_pid = None; + record.updated_at = now.clone(); + record.last_started_at = Some(now); + record.last_stopped_at = None; + record.last_error = None; + let Some(process) = spawned.take() else { + return Err("managed runtime spawn result was already consumed".into()); + }; + runtimes.insert(key.clone(), ManagedAgentPairRuntime::starting(process)); + let status = status_for_with_local_setup(&app, &key, runtimes.get(&key), None, local_setup); + if let Err(error) = save_managed_agents(&app, &records) { + spawned = runtimes.remove(&key).map(|runtime| runtime.process); + return Err(error); + } + Ok(status) + })(); + + if let Some(mut process) = spawned { + let _ = terminate_managed_process(&mut process); let _ = process.child.wait(); - return Err(error); + if wrote_receipt { + super::remove_agent_runtime_receipt(&app, &key); + } } - record.runtime_pid = None; - record.updated_at = now.clone(); - record.last_started_at = Some(now); - record.last_stopped_at = None; - record.last_error = None; - runtimes.insert(key.clone(), ManagedAgentPairRuntime::starting(process)); - let status = status_for(&app, record, &key, runtimes.get(&key), None); - drop(runtimes); - save_managed_agents(&app, &records)?; - emit_status(&app, &status); - Ok(status) + if let Ok(status) = ®istration { + emit_status(&app, status); + } + registration } #[tauri::command] @@ -315,6 +682,21 @@ pub fn stop_managed_agent_runtime( relay_url: String, app: AppHandle, ) -> Result { + let personas = load_personas(&app).unwrap_or_default(); + let global = load_global_agent_config(&app).unwrap_or_default(); + let readiness_records = load_managed_agents(&app)?; + let readiness_record = readiness_records + .iter() + .find(|record| record.pubkey.eq_ignore_ascii_case(&pubkey)) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + let readiness_updated_at = readiness_record.updated_at.clone(); + let local_setup = local_setup_for( + readiness_record, + StatusInputs { + personas: &personas, + global: &global, + }, + ); let state = app.state::(); let _transition = state .managed_agent_runtime_transition @@ -326,18 +708,17 @@ pub fn stop_managed_agent_runtime( .map_err(|e| e.to_string())?; let mut records = load_managed_agents(&app)?; let record = find_managed_agent_mut(&mut records, &pubkey)?; + if record.updated_at != readiness_updated_at { + return Err("managed agent changed while readiness was in flight".into()); + } let key = ManagedAgentRuntimeKey::new(pubkey, &relay_url)?; let mut runtimes = state .managed_agent_processes .lock() .map_err(|e| e.to_string())?; if let Some(mut runtime) = runtimes.remove(&key) { - let stop_result = if process_is_running(runtime.child.id()) { - terminate_process(runtime.child.id()) - } else { - Ok(()) - } - .and_then(|()| runtime.child.wait().map_err(|e| e.to_string())); + let stop_result = terminate_managed_process(&mut runtime.process) + .and_then(|()| runtime.child.wait().map_err(|e| e.to_string())); match stop_result { Ok(status) => { record.last_exit_code = status.code(); @@ -369,7 +750,7 @@ pub fn stop_managed_agent_runtime( record.runtime_pid = None; record.updated_at = crate::util::now_iso(); record.last_stopped_at = Some(record.updated_at.clone()); - let status = status_for(&app, record, &key, None, None); + let status = status_for_with_local_setup(&app, &key, None, None, local_setup); drop(runtimes); save_managed_agents(&app, &records)?; emit_status(&app, &status); @@ -572,6 +953,236 @@ pub async fn reconcile_managed_agent_runtimes( mod tests { use super::*; + #[test] + fn stale_runtime_discovery_is_rejected_after_store_generation_changes() { + use std::sync::{ + atomic::{AtomicU64, Ordering}, + Arc, Barrier, + }; + + let generation = Arc::new(AtomicU64::new(41)); + let discovery_started = Arc::new(Barrier::new(2)); + let allow_persist = Arc::new(Barrier::new(2)); + let worker = std::thread::spawn({ + let generation = Arc::clone(&generation); + let discovery_started = Arc::clone(&discovery_started); + let allow_persist = Arc::clone(&allow_persist); + move || { + let captured = generation.load(Ordering::SeqCst); + discovery_started.wait(); + allow_persist.wait(); + ensure_store_generation_unchanged( + captured, + generation.load(Ordering::SeqCst), + "while a delayed runtime result was in flight", + ) + } + }); + discovery_started.wait(); + generation.store(42, Ordering::SeqCst); + allow_persist.wait(); + let result = worker.join().expect("delayed discovery worker"); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("managed-agent store changed")); + } + + #[test] + fn stale_runtime_discovery_is_rejected_after_process_generation_changes() { + assert!(runtime_generation_matches( + "generation-a", + 100, + "generation-a", + 100, + true + )); + assert!(!runtime_generation_matches( + "generation-a", + 100, + "generation-b", + 100, + true + )); + assert!(!runtime_generation_matches( + "generation-a", + 100, + "generation-a", + 101, + true + )); + assert!(!runtime_generation_matches( + "generation-a", + 100, + "generation-a", + 100, + false + )); + } + + #[tokio::test] + async fn runtime_list_single_flight_shares_hundreds_of_callers() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let coordinator = Arc::new(RuntimeListSingleFlight::default()); + let barrier = Arc::new(tokio::sync::Barrier::new(201)); + let computations = Arc::new(AtomicUsize::new(0)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let mut callers = Vec::new(); + + for _ in 0..200 { + let coordinator = Arc::clone(&coordinator); + let barrier = Arc::clone(&barrier); + let computations = Arc::clone(&computations); + let active = Arc::clone(&active); + let max_active = Arc::clone(&max_active); + callers.push(tokio::spawn(async move { + barrier.wait().await; + coordinator + .run(move || async move { + computations.fetch_add(1, Ordering::SeqCst); + let now_active = active.fetch_add(1, Ordering::SeqCst) + 1; + max_active.fetch_max(now_active, Ordering::SeqCst); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + active.fetch_sub(1, Ordering::SeqCst); + Ok(Vec::new()) + }) + .await + })); + } + barrier.wait().await; + for caller in callers { + assert!(caller.await.expect("caller task").is_ok()); + } + assert_eq!(computations.load(Ordering::SeqCst), 1); + assert_eq!(max_active.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn runtime_list_computation_survives_first_caller_cancellation() { + use std::sync::atomic::{AtomicBool, Ordering}; + + let coordinator = Arc::new(RuntimeListSingleFlight::default()); + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let finished = Arc::new(AtomicBool::new(false)); + let first = { + let coordinator = Arc::clone(&coordinator); + let started = Arc::clone(&started); + let release = Arc::clone(&release); + let finished = Arc::clone(&finished); + tokio::spawn(async move { + coordinator + .run(move || async move { + started.notify_one(); + release.notified().await; + finished.store(true, Ordering::SeqCst); + Ok(Vec::new()) + }) + .await + }) + }; + started.notified().await; + first.abort(); + + let second = coordinator.run(|| async { panic!("a second computation must not start") }); + tokio::pin!(second); + tokio::select! { + result = &mut second => panic!("shared computation finished too early: {result:?}"), + () = tokio::time::sleep(std::time::Duration::from_millis(10)) => {} + } + release.notify_one(); + assert!(second.await.is_ok()); + assert!(finished.load(Ordering::SeqCst)); + } + + #[test] + fn runtime_lifecycle_locks_are_available_while_readiness_is_blocked() { + use std::sync::{mpsc, Arc}; + + let state = Arc::new(crate::app_state::build_app_state()); + let (probe_started_tx, probe_started_rx) = mpsc::channel(); + let (release_probe_tx, release_probe_rx) = mpsc::channel(); + let worker_state = Arc::clone(&state); + let worker = std::thread::spawn(move || { + let _snapshot = collect_runtime_process_snapshots(&worker_state) + .expect("runtime snapshot should succeed"); + let record = record_with_relay(""); + let personas = Vec::new(); + let global = super::super::GlobalAgentConfig::default(); + local_setup_for_with( + &record, + StatusInputs { + personas: &personas, + global: &global, + }, + |_| { + probe_started_tx.send(()).expect("announce fake probe"); + release_probe_rx.recv().expect("release fake probe"); + AgentReadiness::Ready + }, + ) + }); + + probe_started_rx.recv().expect("fake readiness started"); + assert!(state.managed_agent_runtime_transition.try_lock().is_ok()); + assert!(state.managed_agents_store_lock.try_lock().is_ok()); + assert!(state.managed_agent_processes.try_lock().is_ok()); + release_probe_tx.send(()).expect("release fake readiness"); + assert!(worker.join().expect("snapshot worker")); + } + + #[test] + fn runtime_start_reservation_spans_unlocked_spawn_without_holding_lifecycle_locks() { + use std::sync::{mpsc, Arc}; + + let state = Arc::new(crate::app_state::build_app_state()); + let key = ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3000") + .expect("valid runtime key"); + let reservation = + reserve_managed_agent_start(&state, &key).expect("first start reserves the pair"); + let (spawn_started_tx, spawn_started_rx) = mpsc::channel(); + let (release_spawn_tx, release_spawn_rx) = mpsc::channel(); + let worker = std::thread::spawn(move || { + spawn_started_tx.send(()).expect("announce fake spawn"); + release_spawn_rx.recv().expect("release fake spawn"); + }); + + spawn_started_rx.recv().expect("fake spawn started"); + assert!(state.managed_agent_runtime_transition.try_lock().is_ok()); + assert!(state.managed_agents_store_lock.try_lock().is_ok()); + assert!(state.managed_agent_processes.try_lock().is_ok()); + assert!(reserve_managed_agent_start(&state, &key).is_err()); + + release_spawn_tx.send(()).expect("release fake spawn"); + worker.join().expect("fake spawn worker"); + drop(reservation); + assert!(reserve_managed_agent_start(&state, &key).is_ok()); + } + + #[test] + fn windows_recovery_teardown_never_launches_an_external_helper() { + let source = include_str!("process_lifecycle.rs"); + assert!(!source.contains("Command::new")); + assert!(!source.to_ascii_lowercase().contains("taskkill")); + } + + #[test] + fn windows_recovery_teardown_finds_descendants_by_generation() { + let entries = [ + (10, 1), + (11, 10), + (12, 10), + (13, 11), + (14, 13), + (99, 1), + (11, 10), + ]; + assert_eq!( + super::super::runtime::descendant_process_waves(&entries, 10), + vec![vec![11, 12], vec![13], vec![14]] + ); + } + fn payload( relay_url: &str, lifecycle: ManagedAgentRuntimeLifecycle, diff --git a/desktop/src-tauri/src/managed_agents/runtime_types.rs b/desktop/src-tauri/src/managed_agents/runtime_types.rs index 4862cedbae..b28e230a2c 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_types.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_types.rs @@ -115,6 +115,11 @@ pub struct ManagedAgentCommunityTarget { pub struct ManagedAgentRuntimeReceipt { pub key: ManagedAgentRuntimeKey, pub pid: u32, + /// Stable Win32 creation timestamp for this exact process instance. Older + /// receipts and non-Windows builds omit it; Windows teardown then fails + /// closed rather than acting on a PID alone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub process_identity: Option, pub desktop_instance_id: String, pub started_at: String, } diff --git a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs index a7a8cab93e..ac14624f94 100644 --- a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs +++ b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs @@ -443,6 +443,8 @@ fn no_sentinel_reaches_the_owning_process_debug_output() { adapter_availability: None, start_nonce: "test-nonce".to_string(), #[cfg(windows)] + process_identity: 0, + #[cfg(windows)] job: None, }; let rendered = format!("{process:?}"); diff --git a/desktop/src-tauri/src/managed_agents/storage.rs b/desktop/src-tauri/src/managed_agents/storage.rs index 652bb9b9ea..bab4200a4f 100644 --- a/desktop/src-tauri/src/managed_agents/storage.rs +++ b/desktop/src-tauri/src/managed_agents/storage.rs @@ -3,6 +3,7 @@ use std::{ fs::{self, File, OpenOptions}, io::{Read as _, Seek, SeekFrom, Write}, path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, }; use tauri::{AppHandle, Manager}; @@ -13,6 +14,12 @@ use crate::managed_agents::{ }; use crate::secret_store::{KeyringProbe, SecretStore}; +static MANAGED_AGENT_STORE_GENERATION: AtomicU64 = AtomicU64::new(0); + +pub(crate) fn managed_agent_store_generation() -> u64 { + MANAGED_AGENT_STORE_GENERATION.load(Ordering::Acquire) +} + /// Keyring key name for an agent's nsec, namespaced from the human identity /// key (`"identity"`) which shares the service. fn agent_keyring_name(pubkey: &str) -> String { @@ -378,7 +385,9 @@ pub fn save_managed_agents(app: &AppHandle, records: &[ManagedAgentRecord]) -> R // keyring is unreachable, the key stays inline. persist_agent_keys(&mut sorted); - write_agent_store(app, definitions, sorted) + write_agent_store(app, definitions, sorted)?; + crate::managed_agents::readiness::cli_probe::clear_login_probe_cache(); + Ok(()) } /// Save the key-less agent *definitions*, preserving the keyed instances — @@ -391,7 +400,9 @@ pub(crate) fn save_agent_definitions( instances.retain(|record| !record.pubkey.is_empty()); let mut definitions = definitions.to_vec(); definitions.retain(|record| record.pubkey.is_empty()); - write_agent_store(app, definitions, instances) + write_agent_store(app, definitions, instances)?; + crate::managed_agents::readiness::cli_probe::clear_login_probe_cache(); + Ok(()) } /// Serialize definitions + instances into the single unified store file. @@ -414,7 +425,9 @@ fn write_agent_store( // fallback. Write it owner-only (`0o600`) unconditionally — harmless for the // keyring-backed case (it is the user's own agent store) and closes the // umask window a post-write `chmod` would leave open. - atomic_write_json_restricted(&path, &payload) + atomic_write_json_restricted(&path, &payload)?; + MANAGED_AGENT_STORE_GENERATION.fetch_add(1, Ordering::AcqRel); + Ok(()) } /// Write each record's in-memory key to the keyring and blank the inline copy diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index c5bb6173d1..b4c43cde70 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -482,6 +482,10 @@ pub struct ManagedAgentProcess { pub adapter_availability: Option, /// Unpredictable identity shared only with this harness generation. pub start_nonce: String, + /// Win32 creation timestamp read from the stable `Child` process handle at + /// spawn time. Used to reject recycled PIDs before native teardown. + #[cfg(windows)] + pub process_identity: u64, /// Win32 Job Object owning the harness + its entire process tree. Closing /// the handle (via `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`) kills the whole /// tree — the Windows mirror of the Unix process-group teardown. `None` diff --git a/desktop/src/app/AppShell.tsx b/desktop/src/app/AppShell.tsx index fdb4907180..dca3306afe 100644 --- a/desktop/src/app/AppShell.tsx +++ b/desktop/src/app/AppShell.tsx @@ -42,6 +42,7 @@ import { requestOpenCreateAgent } from "@/features/agents/openCreateAgentEvent"; import { useAgentsDataRefresh } from "@/features/agents/lib/useAgentsDataRefresh"; import { useManagedAgentRuntimeReconciliation } from "@/features/agents/useManagedAgentRuntimeReconciliation"; import { useAutoRestartPolicy } from "@/features/agents/lib/useAutoRestartPolicy"; +import { useAgentRecoverySupervisor } from "@/features/agents/lib/useAgentRecoverySupervisor"; import { usePersonaSync } from "@/features/agents/lib/usePersonaSync"; import { useAgentObserverIngestion } from "@/features/agents/useAgentObserverIngestion"; import { AgentManagementDialogs } from "@/features/agents/ui/AgentManagementDialogs"; @@ -182,11 +183,10 @@ export function AppShell() { communitiesHook.activeCommunity?.relayUrl, ); useAgentsDataRefresh(); - // Chunk F: auto-restart drifted idle agents (per-agent opt-out, default ON). useAutoRestartPolicy(); - // Owner-global observer ingestion: receives + decrypts agent observer - // frames and keeps derived active-turn liveness in sync app-wide, so no - // individual screen/panel has to mount its own bridge for ingestion. + useAgentRecoverySupervisor(); + // Owner-global observer ingestion keeps active-turn liveness in sync app-wide, + // so individual screens do not need to mount their own ingestion bridge. // Intentionally mounted without a `startupReady`/identity guard: before // `currentPubkey` resolves the hook ingests managed agents only, and // relay-owned agents join automatically once identity arrives. Adding a diff --git a/desktop/src/features/agents/lib/agentRecoveryPolicy.test.mjs b/desktop/src/features/agents/lib/agentRecoveryPolicy.test.mjs new file mode 100644 index 0000000000..f51d6b2f32 --- /dev/null +++ b/desktop/src/features/agents/lib/agentRecoveryPolicy.test.mjs @@ -0,0 +1,36 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + AGENT_RECOVERY_BACKOFF_MS, + beginAgentRecovery, + recordFailedRecoveryAttempt, + recoveryAttemptDue, + recoveryExhausted, + recoveryLifecycleHealthy, +} from "./agentRecoveryPolicy.ts"; + +test("recovery uses bounded 5s, 30s, 2m backoff", () => { + let state = beginAgentRecovery(1_000, "crash"); + assert.equal(state.nextAttemptAt, 6_000); + assert.equal(recoveryAttemptDue(state, 5_999, false), false); + assert.equal(recoveryAttemptDue(state, 6_000, true), false); + assert.equal(recoveryAttemptDue(state, 6_000, false), true); + + state = recordFailedRecoveryAttempt(state, 6_000, "retry 1"); + assert.equal(state.nextAttemptAt, 6_000 + AGENT_RECOVERY_BACKOFF_MS[1]); + state = recordFailedRecoveryAttempt(state, state.nextAttemptAt, "retry 2"); + assert.equal(state.nextAttemptAt, 36_000 + AGENT_RECOVERY_BACKOFF_MS[2]); + state = recordFailedRecoveryAttempt(state, state.nextAttemptAt, "retry 3"); + assert.equal(recoveryExhausted(state), true); + assert.equal(state.nextAttemptAt, Number.POSITIVE_INFINITY); +}); + +test("recovery is reported only after a listener lifecycle is healthy", () => { + assert.equal(recoveryLifecycleHealthy("starting"), false); + assert.equal(recoveryLifecycleHealthy("failed"), false); + assert.equal(recoveryLifecycleHealthy("stopped"), false); + assert.equal(recoveryLifecycleHealthy("listening"), true); + assert.equal(recoveryLifecycleHealthy("waking"), true); + assert.equal(recoveryLifecycleHealthy("ready"), true); +}); diff --git a/desktop/src/features/agents/lib/agentRecoveryPolicy.ts b/desktop/src/features/agents/lib/agentRecoveryPolicy.ts new file mode 100644 index 0000000000..9926686ec4 --- /dev/null +++ b/desktop/src/features/agents/lib/agentRecoveryPolicy.ts @@ -0,0 +1,58 @@ +export const AGENT_RECOVERY_BACKOFF_MS = [5_000, 30_000, 120_000] as const; + +export type AgentRecoveryState = { + attempts: number; + firstFailedAt: number; + nextAttemptAt: number; + lastError: string | null; +}; + +export function beginAgentRecovery( + now: number, + error: string | null, +): AgentRecoveryState { + return { + attempts: 0, + firstFailedAt: now, + nextAttemptAt: now + AGENT_RECOVERY_BACKOFF_MS[0], + lastError: error, + }; +} + +export function recoveryAttemptDue( + state: AgentRecoveryState, + now: number, + agentIsWorking: boolean, +): boolean { + return ( + !agentIsWorking && + state.attempts < AGENT_RECOVERY_BACKOFF_MS.length && + now >= state.nextAttemptAt + ); +} + +export function recordFailedRecoveryAttempt( + state: AgentRecoveryState, + now: number, + error: string | null, +): AgentRecoveryState { + const attempts = state.attempts + 1; + const nextDelay = AGENT_RECOVERY_BACKOFF_MS[attempts]; + return { + ...state, + attempts, + nextAttemptAt: + nextDelay === undefined ? Number.POSITIVE_INFINITY : now + nextDelay, + lastError: error, + }; +} + +export function recoveryExhausted(state: AgentRecoveryState): boolean { + return state.attempts >= AGENT_RECOVERY_BACKOFF_MS.length; +} + +export function recoveryLifecycleHealthy(lifecycle: string): boolean { + return ( + lifecycle === "listening" || lifecycle === "waking" || lifecycle === "ready" + ); +} diff --git a/desktop/src/features/agents/lib/completionPollScheduler.test.mjs b/desktop/src/features/agents/lib/completionPollScheduler.test.mjs new file mode 100644 index 0000000000..9b30cae59a --- /dev/null +++ b/desktop/src/features/agents/lib/completionPollScheduler.test.mjs @@ -0,0 +1,129 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { createCompletionPollScheduler } from "./completionPollScheduler.ts"; + +function deferred() { + let resolve; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +function fakeTimers() { + let nextHandle = 1; + const callbacks = new Map(); + return { + scheduleTimer(callback) { + const handle = nextHandle++; + callbacks.set(handle, callback); + return handle; + }, + cancelTimer(handle) { + callbacks.delete(handle); + }, + runNext() { + const entry = callbacks.entries().next().value; + assert.ok(entry, "expected a scheduled timer"); + const [handle, callback] = entry; + callbacks.delete(handle); + callback(); + }, + pending() { + return callbacks.size; + }, + }; +} + +test("repeated timer ticks while a poll is unresolved produce exactly one call", async () => { + const gate = deferred(); + const timers = fakeTimers(); + let calls = 0; + const scheduler = createCompletionPollScheduler({ + poll: async () => { + calls += 1; + await gate.promise; + }, + delayMs: 5_000, + ...timers, + }); + + await Promise.resolve(); + const repeated = Array.from({ length: 100 }, () => scheduler.trigger()); + assert.equal(calls, 1); + assert.equal(timers.pending(), 0); + + gate.resolve(); + await Promise.all(repeated); + assert.equal(timers.pending(), 1); + scheduler.stop(); +}); + +test("the next poll starts only after completion and the delay", async () => { + const timers = fakeTimers(); + let calls = 0; + const scheduler = createCompletionPollScheduler({ + poll: async () => { + calls += 1; + }, + delayMs: 5_000, + ...timers, + }); + + await scheduler.trigger(); + assert.equal(calls, 1); + assert.equal(timers.pending(), 1); + + timers.runNext(); + await scheduler.trigger(); + assert.equal(calls, 2); + assert.equal(timers.pending(), 1); + scheduler.stop(); +}); + +test("cleanup cancels future polls and suppresses rescheduling", async () => { + const gate = deferred(); + const timers = fakeTimers(); + let calls = 0; + const scheduler = createCompletionPollScheduler({ + poll: async () => { + calls += 1; + await gate.promise; + }, + delayMs: 5_000, + ...timers, + }); + + await Promise.resolve(); + scheduler.stop(); + gate.resolve(); + await scheduler.trigger().catch(() => {}); + + assert.equal(calls, 1); + assert.equal(timers.pending(), 0); + await scheduler.trigger(); + assert.equal(calls, 1); +}); + +test("a failed automatic poll still schedules exactly one retry", async () => { + const timers = fakeTimers(); + let calls = 0; + const scheduler = createCompletionPollScheduler({ + poll: async () => { + calls += 1; + if (calls === 1) throw new Error("temporary failure"); + }, + delayMs: 5_000, + ...timers, + }); + + await scheduler.trigger().catch(() => {}); + assert.equal(calls, 1); + assert.equal(timers.pending(), 1); + timers.runNext(); + await scheduler.trigger(); + assert.equal(calls, 2); + assert.equal(timers.pending(), 1); + scheduler.stop(); +}); diff --git a/desktop/src/features/agents/lib/completionPollScheduler.ts b/desktop/src/features/agents/lib/completionPollScheduler.ts new file mode 100644 index 0000000000..8d37cdb390 --- /dev/null +++ b/desktop/src/features/agents/lib/completionPollScheduler.ts @@ -0,0 +1,64 @@ +export type TimerHandle = ReturnType; + +type CompletionPollSchedulerOptions = { + poll: () => Promise; + delayMs: number; + scheduleTimer?: (callback: () => void, delayMs: number) => TimerHandle; + cancelTimer?: (handle: TimerHandle) => void; +}; + +export type CompletionPollScheduler = { + /** Start a poll unless one is already running. Concurrent callers share it. */ + trigger: () => Promise; + /** Cancel future polls. An already-running poll is allowed to settle. */ + stop: () => void; +}; + +/** + * Run one poll immediately, then wait `delayMs` after each completed poll + * before starting the next one. This avoids the backlog created by intervals + * when native work takes longer than the nominal polling cadence. + */ +export function createCompletionPollScheduler({ + poll, + delayMs, + scheduleTimer = setTimeout, + cancelTimer = clearTimeout, +}: CompletionPollSchedulerOptions): CompletionPollScheduler { + let stopped = false; + let timer: TimerHandle | null = null; + let inFlight: Promise | null = null; + + const trigger = (): Promise => { + if (stopped) return Promise.resolve(); + if (inFlight) return inFlight; + if (timer !== null) { + cancelTimer(timer); + timer = null; + } + + const running = Promise.resolve() + .then(poll) + .finally(() => { + if (inFlight === running) inFlight = null; + if (stopped) return; + timer = scheduleTimer(() => { + timer = null; + void trigger().catch(() => {}); + }, delayMs); + }); + inFlight = running; + return running; + }; + + const stop = (): void => { + stopped = true; + if (timer !== null) { + cancelTimer(timer); + timer = null; + } + }; + + void trigger().catch(() => {}); + return { trigger, stop }; +} diff --git a/desktop/src/features/agents/lib/useAgentRecoverySupervisor.ts b/desktop/src/features/agents/lib/useAgentRecoverySupervisor.ts new file mode 100644 index 0000000000..03a42c09e3 --- /dev/null +++ b/desktop/src/features/agents/lib/useAgentRecoverySupervisor.ts @@ -0,0 +1,208 @@ +import * as React from "react"; +import { useQueryClient } from "@tanstack/react-query"; +import { toast } from "sonner"; + +import { useManagedAgentsQuery } from "@/features/agents/hooks"; +import { managedAgentRuntimesQueryKey } from "@/features/agents/managedAgentRuntimeHooks"; +import { getAgentWorkingState } from "@/features/agents/agentWorkingSignal"; +import { sendDesktopNotification } from "@/features/notifications/lib/desktop"; +import { + listManagedAgentRuntimes, + restartManagedAgentRuntime, +} from "@/shared/api/tauriManagedAgents"; +import type { ManagedAgentRuntimeStatus } from "@/shared/api/types"; +import { + beginAgentRecovery, + recordFailedRecoveryAttempt, + recoveryAttemptDue, + recoveryExhausted, + recoveryLifecycleHealthy, + type AgentRecoveryState, +} from "./agentRecoveryPolicy"; +import { createCompletionPollScheduler } from "./completionPollScheduler"; + +const HEALTH_POLL_MS = 5_000; + +type PairRecovery = AgentRecoveryState & { + pubkey: string; + relayUrl: string; + agentName: string; + inFlight: boolean; + awaitingHealth: boolean; + exhaustedNotified: boolean; +}; + +function pairKey(pubkey: string, relayUrl: string): string { + return JSON.stringify([pubkey, relayUrl]); +} + +function notifyDesktop(title: string, body: string): void { + void sendDesktopNotification({ title, body }); +} + +/** + * Keeps opted-in local agent listeners reachable after an unexpected exit. + * Recovery is bounded (5s, 30s, 2m), never interrupts an active turn, and + * reports both exhaustion and successful recovery in-app and on the desktop. + */ +export function useAgentRecoverySupervisor(): void { + const queryClient = useQueryClient(); + const agents = useManagedAgentsQuery().data; + const agentsRef = React.useRef(agents); + const recoveriesRef = React.useRef(new Map()); + agentsRef.current = agents; + + React.useEffect(() => { + let cancelled = false; + let pollInFlight = false; + + async function poll(): Promise { + if (pollInFlight || cancelled) return; + pollInFlight = true; + try { + const currentAgents = agentsRef.current; + if (!currentAgents) return; + const eligible = new Map( + currentAgents + .filter( + (agent) => + agent.backend.type === "local" && agent.startOnAppLaunch, + ) + .map((agent) => [agent.pubkey.toLowerCase(), agent]), + ); + + let runtimes: ManagedAgentRuntimeStatus[]; + try { + runtimes = await listManagedAgentRuntimes(); + } catch { + return; + } + if (cancelled) return; + queryClient.setQueryData(managedAgentRuntimesQueryKey, runtimes); + + const now = Date.now(); + const healthyPairs = new Set(); + for (const runtime of runtimes) { + const agent = eligible.get(runtime.pubkey.toLowerCase()); + if (!agent) continue; + const key = pairKey(runtime.pubkey, runtime.relayUrl); + if (recoveryLifecycleHealthy(runtime.lifecycle)) { + healthyPairs.add(key); + const prior = recoveriesRef.current.get(key); + if (prior) { + recoveriesRef.current.delete(key); + toast.success(`${agent.name} is reachable again`); + notifyDesktop( + "Buzz agent recovered", + `${agent.name} is listening again.`, + ); + } + continue; + } + if (runtime.lifecycle === "starting") continue; + const existingRecovery = recoveriesRef.current.get(key); + if (!existingRecovery) { + recoveriesRef.current.set(key, { + ...beginAgentRecovery(now, runtime.error), + pubkey: runtime.pubkey, + relayUrl: runtime.relayUrl, + agentName: agent.name, + inFlight: false, + awaitingHealth: false, + exhaustedNotified: false, + }); + toast.warning(`${agent.name} listener failed; recovery scheduled`); + } else if (existingRecovery.awaitingHealth) { + const updated = recordFailedRecoveryAttempt( + existingRecovery, + now, + runtime.error, + ); + recoveriesRef.current.set(key, { + ...existingRecovery, + ...updated, + inFlight: false, + awaitingHealth: false, + }); + } + } + + for (const [key, recovery] of recoveriesRef.current) { + if ( + healthyPairs.has(key) || + recovery.inFlight || + recovery.awaitingHealth + ) { + continue; + } + if (recoveryExhausted(recovery)) { + if (!recovery.exhaustedNotified) { + recovery.exhaustedNotified = true; + toast.error(`${recovery.agentName} could not be recovered`); + notifyDesktop( + "Buzz agent needs attention", + `${recovery.agentName} failed after 3 recovery attempts.`, + ); + } + continue; + } + if ( + !recoveryAttemptDue( + recovery, + now, + getAgentWorkingState(recovery.pubkey).working, + ) + ) { + continue; + } + + recovery.inFlight = true; + void restartManagedAgentRuntime(recovery.pubkey, recovery.relayUrl) + .then((runtime) => { + if (cancelled) return; + recoveriesRef.current.set(key, { + ...recovery, + inFlight: false, + awaitingHealth: true, + }); + queryClient.setQueryData( + managedAgentRuntimesQueryKey, + (current = []) => [ + ...current.filter( + (candidate) => + pairKey(candidate.pubkey, candidate.relayUrl) !== key, + ), + runtime, + ], + ); + }) + .catch((error: unknown) => { + if (cancelled) return; + const updated = recordFailedRecoveryAttempt( + recovery, + Date.now(), + error instanceof Error ? error.message : String(error), + ); + recoveriesRef.current.set(key, { + ...recovery, + ...updated, + inFlight: false, + awaitingHealth: false, + }); + }); + } + } finally { + pollInFlight = false; + } + } + + const scheduler = createCompletionPollScheduler({ + poll, + delayMs: HEALTH_POLL_MS, + }); + return () => { + cancelled = true; + scheduler.stop(); + }; + }, [queryClient]); +} diff --git a/docs/admin/managed-agent-status-freeze.md b/docs/admin/managed-agent-status-freeze.md new file mode 100644 index 0000000000..b6560dc9d9 --- /dev/null +++ b/docs/admin/managed-agent-status-freeze.md @@ -0,0 +1,100 @@ +# Managed-agent status freeze response + +## Incident signature + +On Linux, the desktop may be reported as unresponsive while managed-agent +status refreshes continue to accumulate. Confirm the signature before changing +state: + +- `buzz-desktop` has an unusually high or steadily increasing thread count. +- Several identical CLI login probes are live at once. +- The UI requests managed-agent runtime status faster than prior requests + finish. + +The August 2026 incident reached 544 desktop threads, including 521 Tokio +workers. A five-second UI polling cadence repeatedly entered a runtime listing +path that performed multiple roughly two-second CLI readiness probes while +runtime-management locks were held. + +## Safeguards + +Runtime status polling is completion-based and single-flight in both the UI and +backend. The backend snapshots runtime generations under its locks, releases +them before any readiness process starts, then revalidates exited generations +before persisting their delayed state. Lifecycle status paths also resolve +readiness before taking runtime locks and reject a result if the managed-agent +record generation changes in flight. Login probes are cached by effective +command and relevant environment, invalidated on successful configuration +writes, limited to one active probe per effective key while different keys +remain independent, and killed and reaped after five seconds. Probe output is +written to anonymous regular files, so a descendant cannot hold pipe EOF open; +after process-group cleanup, at most 64 KiB from each stream is read and +retained. + +Managed-agent starts, launch restoration, and pair reconciliation use a +three-phase transition: snapshot and reserve an exact agent/relay pair, release +all lifecycle/store/process-map mutexes before discovery, readiness, and child +spawn, then generation-check and register the child under the locks. The +per-pair reservation rejects a duplicate start during the unlocked phase. A +shutdown, concurrent record edit, receipt-write failure, or store-save failure +wins the race and the unregistered child is terminated and reaped after every +runtime lock has been released. No external process may be spawned while a +runtime-management lock is held. On Windows, each runtime receipt records the +process creation time captured from the spawned child handle. Live children +are owned by Job Objects; after a desktop restart, recovered process trees are +enumerated through Win32 APIs, opened, and creation-time checked immediately +before termination. Handles remain open throughout bounded descendant sweeps +so PIDs cannot be recycled underneath teardown. An absent, inaccessible, or +mismatched identity fails closed without terminating that PID. The recovery +path does not launch `taskkill` or any other helper executable. + +Windows readiness probes use a separate containment boundary from managed +runtimes: each probe is spawned suspended, assigned to a fresh kill-on-close +Job Object, and only then resumed. Assignment failure kills and reaps the still +suspended child. Timeout and completion cleanup terminate the job, wait for its +active-process count to reach zero, stop output readers, and reap the direct +child, preventing probe descendants from escaping the timeout. + +Unexpected local listener exits use three bounded recovery attempts after 5 +seconds, 30 seconds, and 2 minutes. Recovery is suppressed while an agent has +active work. The desktop reports both confirmed recovery and retry exhaustion; +a successful restart command alone is not considered recovery. + +## Release procedure + +1. Record the current executable checksum, process thread count, configured + listener count, and listener health. +2. Build from the Hermit environment through Tauri's production pipeline. For + an official release, first provide `BUZZ_UPDATER_PUBLIC_KEY` and + `BUZZ_UPDATER_ENDPOINT`, run + `cd desktop && node scripts/build-release-config.mjs`, then run + `pnpm tauri build --verbose --ci --bundles deb,appimage --features mesh-llm --config src-tauri/tauri.release.conf.json`. + For a local incident rollout that must not claim updater signing metadata, + use the base configuration with + `cd desktop && pnpm tauri build --verbose --ci --bundles deb --features mesh-llm`. + Both commands run the configured frontend build and compile the desktop with + Tauri's production custom protocol. A plain `cargo build --release` is not a + releasable artifact because it does not prove that the production frontend + was embedded. +3. Run frontend tests and type checking, Rust formatting, the complete Rust + suite, and repository checks. Stop if a new failure appears. +4. Copy the installed executable to a uniquely timestamped rollback path. +5. Copy the verified artifact beside the installed executable, verify its + checksum, then rename it over the destination on the same filesystem. +6. Restart Buzz once. Do not modify agent stores, keys, relay configuration, or + listener data during the cutover. +7. Confirm all expected listeners and relays, fewer than 80 desktop threads, + and no upward thread trend under repeated status refreshes. + +Rollback by stopping Buzz, atomically restoring the timestamped executable, +and restarting it. Retain the incident artifact and monitor log until the +replacement has passed the 24-hour checkpoint. + +## Monitoring checkpoints + +At 15 minutes, 2 hours, and 24 hours after deployment, record the executable +checksum, PID, thread count, file-descriptor count, listener count, unhealthy +listeners, and duplicated readiness processes. Roll back immediately if the UI +hang recurs, thread count reaches 80, thread or descriptor counts trend upward, +listeners are missing, or the same effective readiness command runs +concurrently.