diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index 050c3af89db..f41fa2d6e39 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -63,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/managed_agents/discovery.rs b/desktop/src-tauri/src/managed_agents/discovery.rs index 78592357c9b..1ee7e6e5562 100644 --- a/desktop/src-tauri/src/managed_agents/discovery.rs +++ b/desktop/src-tauri/src/managed_agents/discovery.rs @@ -1,8 +1,7 @@ -use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Command; use std::sync::OnceLock; -use std::time::{Duration, Instant}; +use std::time::Duration; use crate::managed_agents::{ buzz_managed_command_path, buzz_managed_node_bin_dir, buzz_managed_npm_bin_dir, @@ -10,6 +9,7 @@ use crate::managed_agents::{ HarnessSource, }; mod auth_status_cache; +mod bounded_command; mod login_shell; mod presets; mod runtime_metadata; @@ -593,6 +593,16 @@ pub fn resolve_command_cached(command: &str) -> Option { if let Some(managed) = resolve_buzz_managed_command(command) { return Some(managed); } + // Bundled sidecars (e.g. `buzz-agent`) ship next to the app executable, so + // `resolve_workspace_command` finds them with a filesystem stat and no + // login-shell spawn — the same class of work the managed-shim check above + // already performs. Without this the cheap path could never see the sidecar + // until a forced discovery warmed the resolve cache, so `buzz-agent` (which + // cannot legitimately be missing) reported "not installed" at every cold + // launch across the create/edit and agent-defaults surfaces. + if let Some(workspace) = resolve_workspace_command(command) { + return Some(workspace); + } resolve_cache() .lock() .ok() @@ -822,10 +832,9 @@ pub(crate) fn is_npm_global_install(cmd: &str) -> bool { /// 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. +/// On timeout or spawn failure the child is killed and `Unknown` is returned; +/// no orphaned threads or processes are left behind (see +/// [`bounded_command::output_with_timeout`]). fn probe_auth_status(binary_path: &Path, probe_args: &[&str]) -> AuthStatus { use crate::managed_agents::readiness::cli_probe; @@ -836,81 +845,17 @@ fn probe_auth_status(binary_path: &Path, probe_args: &[&str]) -> AuthStatus { 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, + // Window suppression is owned by `output_with_timeout`'s spawn + // (`BOUNDED_CREATION_FLAGS` carries `CREATE_NO_WINDOW`); a + // `configure_no_window` call here would be clobbered by that later + // `creation_flags` set, so it is deliberately omitted. + + let Some(output) = bounded_command::output_with_timeout(command, Duration::from_secs(10)) + else { + 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::classify_probe_output(&output.stderr, output.status.success()) { cli_probe::ProbeOutcome::LoggedIn => AuthStatus::LoggedIn, cli_probe::ProbeOutcome::LoggedOut => AuthStatus::LoggedOut, cli_probe::ProbeOutcome::ConfigInvalid { stderr_excerpt } => AuthStatus::ConfigInvalid { diff --git a/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs b/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs new file mode 100644 index 00000000000..18286d62b1e --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs @@ -0,0 +1,999 @@ +//! Run a child process to completion under a hard wall-clock deadline. +//! +//! Every spawn on the discovery path — the CLI auth probes and the login-shell +//! PATH lookups — must return in bounded time no matter how the child behaves. +//! A login shell that blocks on an interactive prompt, a child that traps +//! `SIGTERM`, or a forked descendant that keeps a pipe open must not be able to +//! stall discovery; that stall is what left "Check again" spinning forever. + +use std::io::{ErrorKind, Read}; +use std::process::{ChildStderr, ChildStdout, Command, ExitStatus, Output, Stdio}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +/// Poll interval while waiting for the child to exit. +const POLL_INTERVAL: Duration = Duration::from_millis(50); + +/// Idle backoff for a nonblocking Unix drain that has no bytes available and +/// has not yet been told to stop. Short so a running child's output is pulled +/// promptly and the post-teardown join returns quickly. +#[cfg(unix)] +const DRAIN_IDLE_POLL: Duration = Duration::from_millis(5); + +/// Maximum bytes retained across stdout + stderr for one bounded probe. +/// +/// Discovery output is tiny — a version string, an auth-status word, a PATH +/// lookup. A probe that emits more than this is noisy or hostile. The ceiling +/// is enforced *in the drain sink* (see [`spawn_drain`]): each stream is pulled +/// on its own thread into a capped buffer, the limit is checked the moment a +/// bounded read crosses it, and the probe is failed closed — so an over-cap +/// payload is never retained in memory (and, since output goes to pipes not +/// temp files, never written to disk). The ceiling is *aggregate*, not +/// per-stream, so a probe cannot double it by splitting output across stdout +/// and stderr. +const CAPTURE_LIMIT: u64 = 1 << 20; // 1 MiB + +/// Grace period between the initial `SIGTERM` and the escalating `SIGKILL` for a +/// timed-out process group. Long enough for a well-behaved child to flush and +/// exit cleanly, short enough that a signal-ignoring one is reaped promptly. +#[cfg(unix)] +const KILL_GRACE: Duration = Duration::from_millis(500); + +/// Freeze the child so the Job Object can take ownership before any child code +/// runs (see [`BoundedChild::spawn`]). +#[cfg(windows)] +const CREATE_SUSPENDED: u32 = 0x0000_0004; + +/// Suppress the console window a GUI-spawned console child would otherwise +/// flash — the same suppression [`crate::util::configure_no_window`] applies to +/// non-bounded spawns. +#[cfg(windows)] +const CREATE_NO_WINDOW: u32 = 0x0800_0000; + +/// The exact creation flags every bounded child is spawned with. +/// +/// `Command::creation_flags` *replaces* rather than accumulates (std ORs only +/// `CREATE_UNICODE_ENVIRONMENT` afterward), and [`BoundedChild::spawn`] is the +/// last writer before spawn, so a caller's earlier `configure_no_window` is +/// wiped. This constant therefore has to carry every flag a bounded child +/// needs, and owning both here keeps the window-suppression contract in one +/// place instead of split between the caller and the helper. +#[cfg(windows)] +const BOUNDED_CREATION_FLAGS: u32 = CREATE_SUSPENDED | CREATE_NO_WINDOW; + +/// Compile-time guard: the bounded flags must always carry *both* bits. A +/// future edit that drops `CREATE_NO_WINDOW` (reintroducing the console-flash +/// regression) or `CREATE_SUSPENDED` (reopening the spawn-to-assign race) fails +/// the build on Windows rather than shipping silently. +#[cfg(windows)] +const _: () = { + assert!(BOUNDED_CREATION_FLAGS & CREATE_SUSPENDED == CREATE_SUSPENDED); + assert!(BOUNDED_CREATION_FLAGS & CREATE_NO_WINDOW == CREATE_NO_WINDOW); +}; + +/// A spawned child plus ownership of its descendant tree, torn down on *every* +/// exit path — timeout, error, or successful exit. The two platforms establish +/// ownership differently, and the guarantee is deliberately asymmetric — the +/// adjudicated design, not an oversight: +/// +/// - **Unix:** the child leads its own process group (`process_group(0)`), so +/// `killpg` reaches every descendant that has not left the group. A +/// `setsid`/`setpgid` escapee holding a pipe is *not* owned and may survive +/// one probe, yet never hangs the helper (see [`output_with_timeout`]). +/// - **Windows:** the child is spawned `CREATE_SUSPENDED`, assigned to a +/// kill-on-close Job Object while frozen, then resumed. The job owns the root +/// before any descendant can exist and is created without breakaway, so no +/// writer can escape it — a hard whole-tree guarantee. Closing that job reaps +/// the whole tree *even after the root has exited* — the distinction that +/// makes `taskkill /T ` (a live-root lookup) unfit for the success path. +/// This mirrors the Job Object discipline the harness uses to reap its 24 +/// agent workers (`process_lifecycle.rs`). +struct BoundedChild { + child: std::process::Child, + /// The kill-on-close job that owns the whole tree. Taken and dropped by + /// `kill_tree` so the reap happens exactly once. Spawn is fail-closed: if + /// the job cannot be created, assigned, or the child resumed, the child is + /// terminated and `spawn` returns `None` rather than running unowned. + #[cfg(windows)] + job: Option, +} + +impl BoundedChild { + /// Spawn `command`, establishing tree ownership before the child can run. + /// Returns `None` if the spawn fails or — on Windows — if the job cannot be + /// created, assigned, or the frozen child resumed; in every such case the + /// child is terminated and reaped before returning, so no unowned process + /// survives. + fn spawn(mut command: Command) -> Option { + // Run the child in its own process group so the whole tree can be torn + // down as a unit, not just a direct child that may have forked workers. + #[cfg(unix)] + { + use std::os::unix::process::CommandExt as _; + command.process_group(0); + } + + // Spawn frozen so the Job Object can take ownership before any child + // code runs and forks a descendant that would escape the job. The flags + // are set here as the last writer before spawn; `Command::creation_flags` + // replaces rather than ORs, so `BOUNDED_CREATION_FLAGS` must itself carry + // `CREATE_NO_WINDOW` — a caller's earlier `configure_no_window` would be + // clobbered otherwise, flashing a console window on GUI discovery. + #[cfg(windows)] + { + use std::os::windows::process::CommandExt as _; + command.creation_flags(BOUNDED_CREATION_FLAGS); + } + + // `mut` is used only on the Windows fail-closed path (kill/wait on the + // frozen child); Unix moves the child unmodified into `Self`. + #[cfg_attr(not(windows), allow(unused_mut))] + let mut child = command.spawn().ok()?; + + #[cfg(windows)] + let job = { + // Assign the frozen child to a kill-on-close job, then resume it. + // Any failure is fail-closed: terminate + reap the still-owned + // child and abort the spawn, never run it unowned to the deadline. + let Some(job) = crate::managed_agents::create_job_for_child(child.id()) else { + let _ = child.kill(); + let _ = child.wait(); + return None; + }; + if !crate::managed_agents::resume_process(child.id()) { + // Dropping the job kills the still-suspended child via + // kill-on-close; reap it so no zombie lingers. + drop(job); + let _ = child.wait(); + return None; + } + job + }; + + Some(Self { + child, + #[cfg(windows)] + job: Some(job), + }) + } + + fn try_wait(&mut self) -> std::io::Result> { + self.child.try_wait() + } + + /// Timeout teardown: a graceful `SIGTERM` to the group and a bounded grace + /// period for a clean flush on Unix, then the unconditional forced kill. + /// Windows has no group signal, so it goes straight to the forced kill. + fn terminate_timed_out(&mut self) { + #[cfg(unix)] + { + // SAFETY: `killpg` on the group led by the child; an ignored result + // is intentional — the group may already be gone (ESRCH). + unsafe { + libc::killpg(self.child.id() as i32, libc::SIGTERM); + } + std::thread::sleep(KILL_GRACE); + } + self.kill_tree(); + } + + /// Forcibly reap the whole tree. Idempotent and safe on an already-exited + /// tree. Runs on every exit path — including success, because a login shell + /// or auth CLI can background a descendant that outlives the leader while + /// still holding the captured-output descriptors. + fn kill_tree(&mut self) { + #[cfg(unix)] + // SAFETY: `killpg` on the group led by the child; ignored result is + // intentional — `ESRCH` on a dead group is the success case. + unsafe { + libc::killpg(self.child.id() as i32, libc::SIGKILL); + } + #[cfg(windows)] + // Closing the kill-on-close job reaps every descendant, even once the + // root has exited — which `taskkill /T ` cannot. `spawn` is + // fail-closed, so the job is always present until this first take; + // a later take is a no-op (the tree is already reaped). + if let Some(job) = self.job.take() { + drop(job); + } + } + + /// Reap the direct child so no zombie lingers after the tree is killed. + fn reap(&mut self) { + let _ = self.child.wait(); + } + + /// Take the captured stdout pipe. `Some` because [`output_with_timeout`] + /// configures `Stdio::piped()` before spawn. + fn take_stdout(&mut self) -> Option { + self.child.stdout.take() + } + + /// Take the captured stderr pipe. + fn take_stderr(&mut self) -> Option { + self.child.stderr.take() + } +} + +/// Set a file descriptor nonblocking so a read on it returns `WouldBlock` +/// instead of parking when no bytes are available. Returns `false` on any +/// `fcntl` failure, which the caller treats as fail-closed. +#[cfg(unix)] +fn set_nonblocking(f: &F) -> bool { + let fd = f.as_raw_fd(); + // SAFETY: `fd` is owned by `f` for the duration of this call; `F_GETFL` / + // `F_SETFL` read and set the descriptor's flags without transferring + // ownership or touching any other resource. + unsafe { + let flags = libc::fcntl(fd, libc::F_GETFL); + if flags < 0 { + return false; + } + libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) == 0 + } +} + +/// Drain one child stream on its own thread into a buffer capped by the shared +/// aggregate budget, so the sink itself — not a post-hoc size sample — enforces +/// [`CAPTURE_LIMIT`]. +/// +/// Continuous draining keeps the pipe buffer from filling, so the child can +/// never block on a full pipe while we poll it. Retention is bounded: `total` +/// reserves a disjoint byte range per chunk across both streams, so the sum of +/// both buffers never exceeds the aggregate cap. The moment a read crosses the +/// cap, `overflow` is set and the drain returns immediately — it does not keep +/// reading, so a writer that keeps the pipe continuously readable cannot spin +/// this loop forever (it must cross the finite cap). A read error other than +/// `Interrupted`/`WouldBlock` returns `Err`, which the caller treats as +/// fail-closed. +/// +/// **Bounded completion differs by platform, because tree ownership does.** +/// - **Unix:** the read end is nonblocking (see [`set_nonblocking`]). A killed +/// in-group writer's descriptors close, so the read reaches EOF (`Ok(0)`) and +/// the thread returns normally. But `kill_tree` is a `killpg` on the child's +/// group, which does *not* reach a descendant that left the group via +/// `setsid`/`setpgid` while retaining the pipe; that writer keeps the write +/// end open and EOF never comes. So once teardown has set `stop`, a +/// `WouldBlock` (nothing more buffered) ends the drain rather than waiting on +/// that escaped writer forever. This is what makes bounded return hold +/// *without* depending on every inherited writer exiting — the correction to +/// the round-8 blocking-EOF design. +/// - **Windows:** the read blocks to EOF. That is sound because the whole tree +/// is owned by a kill-on-close Job Object created without +/// `JOB_OBJECT_LIMIT_BREAKAWAY_OK`, so no descendant can escape the job; job +/// close reaps every writer and the read reaches EOF. `stop` is unused there. +fn spawn_drain( + mut reader: R, + total: Arc, + overflow: Arc, + stop: Arc, +) -> JoinHandle>> { + // `stop` gates only the nonblocking Unix drain; the Windows path blocks to + // the job-close EOF and never consults it. + #[cfg(windows)] + let _ = &stop; + std::thread::spawn(move || { + let mut buf = Vec::new(); + let mut chunk = [0u8; 8192]; + loop { + match reader.read(&mut chunk) { + Ok(0) => return Ok(buf), + Ok(n) => { + // Atomically reserve [prev, prev + n) of the shared budget; + // `prev` is unique per call, so the two streams keep + // disjoint ranges and their retained bytes sum to <= cap. + let prev = total.fetch_add(n as u64, Ordering::Relaxed); + if prev.saturating_add(n as u64) > CAPTURE_LIMIT { + overflow.store(true, Ordering::Relaxed); + let keep = CAPTURE_LIMIT.saturating_sub(prev).min(n as u64) as usize; + buf.extend_from_slice(&chunk[..keep]); + // Overflow: the result is already fail-closed, so nothing + // still in the pipe is worth preserving. Return NOW rather + // than draining to EOF — this is what bounds the `Ok(n)` + // path against a writer that keeps the pipe continuously + // readable, which would otherwise never reach the + // `WouldBlock`/`stop` check below and hang the join. It is + // safe to stop draining: the poll loop sees `overflow` and + // kills the tree, and a writer that then blocks on a full + // pipe dies to `killpg`/job-close. Do NOT "fix" that + // blocked-writer case by resuming an unbounded drain here. + return Ok(buf); + } + buf.extend_from_slice(&chunk[..n]); + } + Err(e) if e.kind() == ErrorKind::Interrupted => continue, + // Nonblocking read (Unix only): no bytes available right now. + // After teardown, an escaped out-of-group writer is the only + // thing that could still hold the pipe open, so stop draining it + // rather than block the join forever; otherwise back off and + // retry so a running child's later output is still captured. + #[cfg(unix)] + Err(e) if e.kind() == ErrorKind::WouldBlock => { + if stop.load(Ordering::Relaxed) { + return Ok(buf); + } + std::thread::sleep(DRAIN_IDLE_POLL); + } + Err(e) => return Err(e), + } + } + }) +} + +/// Run `command` to completion, bounded by `timeout`. +/// +/// Returns `Some(output)` when the child exits within the deadline, `None` when +/// it fails to spawn, exceeds the deadline, or breaches the capture ceiling. +/// Guarantees a bounded return regardless of child cooperation: +/// +/// - **Sink-enforced capture bound.** Stdout and stderr are piped to two drain +/// threads that read into buffers capped by a shared aggregate budget +/// ([`spawn_drain`]); nothing over [`CAPTURE_LIMIT`] is ever retained. On a +/// breach the poll loop fails closed — kill the tree, return `None` — so a +/// noisy or hostile probe cannot force unbounded memory (and, with pipes +/// rather than temp files, cannot fill the disk either). Continuous draining +/// also keeps the pipe buffer from filling, so the child can never block on a +/// full pipe while we poll. +/// - **Bounded drain completion without depending on writer death.** Tree +/// teardown runs on *every* exit path before the drains are joined — +/// [`BoundedChild::kill_tree`] on timeout, error, cap breach, *and* success. +/// But teardown alone does not guarantee EOF on Unix: `kill_tree` is a +/// `killpg` on the child's process group, and a descendant that left the +/// group (`setsid`/`setpgid`) while retaining the pipe survives it and keeps +/// the write end open. So the drains do not rely on EOF from every writer: +/// the Unix reads are nonblocking, and after teardown sets the shared `stop` +/// flag a `WouldBlock` (no more buffered bytes) ends each drain. An escaped +/// writer is allowed to survive; the join still returns promptly. On Windows +/// the reads block to EOF, which is sound because the kill-on-close Job Object +/// is created without breakaway, so no writer can escape the job. This is the +/// correction to the round-8 design, whose blocking Unix reads could hang the +/// join forever on a group-escaping writer. +/// - **No wait hang.** The child is polled with [`Child::try_wait`] against the +/// deadline rather than blocked on with `wait()`. +/// - **Tree termination on every exit path.** [`BoundedChild`] tears the tree +/// down whether the child times out, errors, breaches the cap, *or exits +/// successfully* — a login-shell rc file or auth CLI can legitimately +/// background a descendant (`worker &`) that would outlive discovery. +/// Ownership is a hard whole-tree guarantee on Windows but only the child's +/// process group on Unix (the group-escapee case bounded by the drain rule +/// above) — the adjudicated asymmetry. The timeout path additionally sends a +/// graceful `SIGTERM` and a grace period before the kill. +pub(crate) fn output_with_timeout(mut command: Command, timeout: Duration) -> Option { + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let mut child = BoundedChild::spawn(command)?; + + let stdout_pipe = child.take_stdout(); + let stderr_pipe = child.take_stderr(); + + // Unix: make the parent read ends nonblocking so a drain can be told to stop + // (post-teardown) instead of parking forever on a group-escaping writer that + // still holds the pipe. Fail closed if the fd cannot be reconfigured — the + // child is still fully owned here, so cleanup is just kill + reap. + #[cfg(unix)] + { + let stdout_ok = match stdout_pipe.as_ref() { + Some(p) => set_nonblocking(p), + None => true, + }; + let stderr_ok = match stderr_pipe.as_ref() { + Some(p) => set_nonblocking(p), + None => true, + }; + if !(stdout_ok && stderr_ok) { + child.kill_tree(); + child.reap(); + return None; + } + } + + // Shared drain state: one aggregate byte budget across both streams, an + // overflow flag the poll loop watches so a streaming producer that never + // exits is failed closed the moment it crosses the cap, and a stop flag that + // teardown raises to end the nonblocking Unix drains. + let total = Arc::new(AtomicU64::new(0)); + let overflow = Arc::new(AtomicBool::new(false)); + let stop = Arc::new(AtomicBool::new(false)); + let stdout_drain = + stdout_pipe.map(|s| spawn_drain(s, total.clone(), overflow.clone(), stop.clone())); + let stderr_drain = + stderr_pipe.map(|s| spawn_drain(s, total.clone(), overflow.clone(), stop.clone())); + + let deadline = Instant::now() + timeout; + let status = loop { + match child.try_wait() { + Ok(Some(status)) => break Some(status), + Ok(None) => { + if Instant::now() >= deadline { + child.terminate_timed_out(); + break None; + } + // Fail closed on a capture breach *while the child runs*: the + // drain kept nothing over the cap; teardown below ends the + // drains so the join cannot hang. + if overflow.load(Ordering::Relaxed) { + child.kill_tree(); + break None; + } + std::thread::sleep(POLL_INTERVAL); + } + Err(_) => { + child.kill_tree(); + break None; + } + } + }; + + // Tree down on every path (timeout/error/overflow killed it above; a clean + // exit may still have backgrounded a descendant holding the pipe). Kill is + // idempotent, so calling it here on the success path is safe. Then raise + // `stop`: a killed in-group writer's pipe reaches EOF and ends its drain on + // its own, but a group-escaping writer never will — `stop` ends that drain + // on the next `WouldBlock` so the joins below return promptly. + child.kill_tree(); + child.reap(); + stop.store(true, Ordering::Relaxed); + + let stdout = join_drain(stdout_drain); + let stderr = join_drain(stderr_drain); + + // Fail closed if the child exited within the deadline but overran the cap in + // a final burst, or if either drain hit a read error (join_drain -> None). + let (status, stdout, stderr) = (status?, stdout?, stderr?); + if overflow.load(Ordering::Relaxed) { + return None; + } + + Some(Output { + status, + stdout, + stderr, + }) +} + +/// Join a drain thread, returning its captured bytes. `None` (fail closed) if +/// the stream was absent, the thread panicked, or the read errored. +fn join_drain(drain: Option>>>) -> Option> { + match drain { + Some(handle) => handle.join().ok()?.ok(), + None => Some(Vec::new()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + /// Drive `output_with_timeout` on its own thread under an independent + /// wall-clock `bound` — the real outer bound, unreachable by an inline `elapsed()` assertion if the helper hangs. + /// The raw result lets the Windows sites fold transcripts into the expiry panic. + #[cfg(any(unix, windows))] + fn run_watchdogged_raw( + cmd: Command, + timeout: Duration, + bound: Duration, + ) -> Result, mpsc::RecvTimeoutError> { + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(output_with_timeout(cmd, timeout)); + }); + rx.recv_timeout(bound) + } + #[cfg(unix)] + fn run_watchdogged(cmd: Command, timeout: Duration, bound: Duration) -> Option { + run_watchdogged_raw(cmd, timeout, bound) + .unwrap_or_else(|_| panic!("output_with_timeout did not return within {bound:?}")) + } + + /// True while a Unix process (or a reaped-but-not-waited zombie under this + /// test process) still exists. `kill(pid, 0)` probes existence without + /// signalling. Descendants reparent to init on exit, so a survivor stays + /// probeable; once `kill_tree` reaps it, the pid is gone (ESRCH). + #[cfg(unix)] + fn pid_alive(pid: i32) -> bool { + unsafe { libc::kill(pid, 0) == 0 } + } + + #[cfg(unix)] + #[test] + fn returns_output_for_fast_command() { + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", "printf hi; printf oops 1>&2"]); + let out = run_watchdogged(cmd, Duration::from_secs(5), Duration::from_secs(10)) + .expect("a fast command must complete within the timeout"); + assert!(out.status.success()); + assert_eq!(out.stdout, b"hi"); + assert_eq!(out.stderr, b"oops"); + } + + // Adversarial: a child that traps and ignores SIGTERM. The old + // wait-thread + lone-SIGTERM helper never returned for this input; the + // process-group SIGKILL escalation must reap it inside the grace period. + // The watchdog thread is the real bound — the helper hanging fails the + // test rather than hanging it. + #[cfg(unix)] + #[test] + fn kills_sigterm_ignoring_child_within_bound() { + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", "trap '' TERM; while :; do sleep 1; done"]); + let result = run_watchdogged(cmd, Duration::from_millis(200), Duration::from_secs(5)); + assert!(result.is_none(), "a timed-out child must yield None"); + } + + // Adversarial (success path): the direct child exits 0 but backgrounds a + // descendant that keeps writing to the inherited stdout/stderr forever. + // Two guarantees under test: (1) the drain returns rather than blocking on + // the descendant, and (2) `kill_tree` reaps that descendant before + // returning, so no survivor keeps consuming CPU after discovery reports + // success. This is the pass-2 leak Thufir proved with `(yes) & exit 0`. + #[cfg(unix)] + #[test] + fn reaps_backgrounded_descendant_on_success() { + let pid_file = tempfile::NamedTempFile::new().expect("temp file for descendant pid"); + let pid_path = pid_file + .path() + .to_str() + .expect("utf-8 temp path") + .to_string(); + // Background a real child process (`sleep`), record ITS pid via `$!` + // (not `$$`, which in a subshell is the invoking shell), then exit 0. + // The leader waits until the pid is recorded so the test can read it + // deterministically even though the success path kills the group at + // once. `$!` is the pass-2 `(yes) & exit 0` survivor, made observable. + let script = format!( + "sleep 30 & echo $! > '{pid_path}'; \ + until [ -s '{pid_path}' ]; do :; done; printf done; exit 0" + ); + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", &script]); + let out = run_watchdogged(cmd, Duration::from_secs(5), Duration::from_secs(10)) + .expect("the direct child exits, so this must return its output"); + assert!(out.status.success()); + + let descendant_pid: i32 = std::fs::read_to_string(&pid_path) + .expect("descendant must have recorded its PID") + .trim() + .parse() + .expect("descendant PID must be numeric"); + // Give the reaped group a moment to fully disappear, then assert dead. + std::thread::sleep(Duration::from_millis(200)); + assert!( + !pid_alive(descendant_pid), + "backgrounded descendant {descendant_pid} must be reaped on success, but it survived" + ); + } + + // Adversarial (timeout path): a SIGTERM-ignoring leader that backgrounds a + // descendant, both looping forever. The leader's process group is killed on + // timeout, so the descendant (same group) must die too. The descendant is a + // real child process whose PID is recorded via `$!`, so the test proves the + // actual descendant — not the already-reaped leader — reaches ESRCH. + #[cfg(unix)] + #[test] + fn reaps_descendant_on_timeout() { + let pid_file = tempfile::NamedTempFile::new().expect("temp file for descendant pid"); + let pid_path = pid_file + .path() + .to_str() + .expect("utf-8 temp path") + .to_string(); + let script = format!( + "trap '' TERM; sleep 300 & echo $! > '{pid_path}'; \ + while :; do sleep 1; done" + ); + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", &script]); + let result = run_watchdogged(cmd, Duration::from_millis(300), Duration::from_secs(5)); + assert!(result.is_none(), "a timed-out tree must yield None"); + + let descendant_pid: i32 = std::fs::read_to_string(&pid_path) + .expect("descendant must have written its PID") + .trim() + .parse() + .expect("descendant PID must be numeric"); + std::thread::sleep(Duration::from_millis(200)); + assert!( + !pid_alive(descendant_pid), + "backgrounded descendant {descendant_pid} must be group-killed on timeout, but it survived" + ); + } + + // Deterministic seam regression (Thufir's finding): a drain fed a reader + // that stays continuously readable — every `read` returns `Ok(8192)`, never + // `WouldBlock` — must still complete, because the `stop`/`WouldBlock` check + // alone never fires on such a reader. The bound comes from the `Ok(n)` path + // returning the instant the aggregate cap is crossed. No real process and no + // scheduler timing: the reader is a pure in-test `Read` impl, so this pins + // the control flow rather than relying on a descendant eventually blocking. + // With the round-9-initial code (which kept reading after overflow) the + // drain never returns and the join below hangs past the watchdog. + #[test] + fn overflow_bounds_a_continuously_readable_drain() { + /// A reader that is always ready with a full 8192-byte chunk. It never + /// returns 0 (EOF) or `WouldBlock`, so only the overflow return can end + /// a drain reading it. + struct AlwaysReady; + impl Read for AlwaysReady { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + for b in buf.iter_mut() { + *b = b'x'; + } + Ok(buf.len()) + } + } + + let total = Arc::new(AtomicU64::new(0)); + let overflow = Arc::new(AtomicBool::new(false)); + // `stop` set from the start: a correct drain must NOT depend on it here, + // since a continuously-ready reader never hits the `WouldBlock` arm that + // consults it. The overflow return is the only thing that can bound it. + let stop = Arc::new(AtomicBool::new(true)); + let drain = spawn_drain(AlwaysReady, total.clone(), overflow.clone(), stop); + + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(drain.join()); + }); + let joined = rx + .recv_timeout(Duration::from_secs(2)) + .expect("a continuously-readable drain must be bounded by the capture cap"); + let buf = joined + .expect("drain thread must not panic") + .expect("drain read must not error"); + assert!( + overflow.load(Ordering::Relaxed), + "the drain must have tripped overflow" + ); + assert!( + buf.len() as u64 <= CAPTURE_LIMIT, + "retained bytes {} must not exceed the cap {CAPTURE_LIMIT}", + buf.len() + ); + } + + // Adversarial (group escape): the leader backgrounds a descendant that + // calls `setsid()` — leaving the leader's process group while retaining the + // inherited stdout — then sleeps 300s; the leader itself loops forever, so + // the helper times out. `kill_tree` is a `killpg` on the leader's group and + // cannot reach the escaped descendant, so its pipe write end stays open and + // never reaches EOF. The helper must still return within the outer watchdog + // and fail closed: the nonblocking drains stop on `WouldBlock` after + // teardown rather than blocking on that surviving writer. This is the exact + // primitive Thufir reproduced against the round-8 blocking-read design; with + // blocking reads the drain join hangs forever and `run_watchdogged` panics. + // + // Non-vacuous: the descendant is asserted *alive* after the helper returns, + // proving it genuinely escaped the `killpg` (so it was still holding the + // pipe at join time) — the return therefore came from the stop path, not + // from an EOF the kill happened to produce. The test then reaps it. + #[cfg(unix)] + #[test] + fn returns_when_escaped_descendant_retains_pipe() { + let pid_file = tempfile::NamedTempFile::new().expect("temp file for descendant pid"); + let pid_path = pid_file + .path() + .to_str() + .expect("utf-8 temp path") + .to_string(); + // The perl descendant `setsid()`s out of the leader's group, records its + // PID, writes a few bytes to the retained stdout, then sleeps. The + // leader waits until the PID is recorded (so the test can read it) and + // then loops forever, forcing the timeout path. + let script = format!( + "perl -MPOSIX -e 'POSIX::setsid() or die; open(my $f,\">\",$ARGV[0]) or die; \ + print $f $$; close $f; print \"x\" x 4096; sleep 300;' '{pid_path}' & \ + until [ -s '{pid_path}' ]; do :; done; while :; do sleep 1; done" + ); + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", &script]); + let result = run_watchdogged(cmd, Duration::from_millis(300), Duration::from_secs(5)); + assert!( + result.is_none(), + "a timed-out probe must fail closed even when an escaped writer holds the pipe" + ); + + let descendant_pid: i32 = std::fs::read_to_string(&pid_path) + .expect("escaped descendant must have recorded its PID") + .trim() + .parse() + .expect("descendant PID must be numeric"); + assert!( + pid_alive(descendant_pid), + "descendant {descendant_pid} was expected to survive the group kill (proving it escaped)" + ); + // Reap the escaped writer so the test leaves nothing behind. + unsafe { + libc::kill(descendant_pid, libc::SIGKILL); + } + } + + // Adversarial (capture bound): a producer that streams zero bytes + // *indefinitely* — it never exits and never stops writing on its own, so + // the only thing that can end the probe is the in-flight ceiling check + // tripping `overflow`, killing the tree, and failing closed (None). + // + // The discriminator is `timeout >> bound`: the deadline is 60s but the + // watchdog fails the test at 10s, so a return within the bound proves the + // *cap* ended the probe, not the timeout. Neuter the overflow check and the + // helper runs until the 60s deadline, blowing the 10s watchdog. Pipe + // backpressure cannot end it either: the drains pull continuously, so `cat` + // would keep writing forever. Retention stays bounded by construction — + // `spawn_drain` reserves a disjoint byte range per chunk against the shared + // budget and discards everything past `CAPTURE_LIMIT` — so no over-cap + // payload is ever materialized even though the producer is infinite. + #[cfg(unix)] + #[test] + fn fails_closed_when_capture_exceeds_limit() { + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", "exec cat /dev/zero"]); + let result = run_watchdogged(cmd, Duration::from_secs(60), Duration::from_secs(10)); + assert!( + result.is_none(), + "an unbounded producer must fail closed on the capture cap, well before the deadline" + ); + } + + // The complement of the bound: output at or under the ceiling still returns + // in full, so the limit rejects only genuine overruns. + #[cfg(unix)] + #[test] + fn returns_full_output_at_capture_limit() { + let mut cmd = Command::new("/bin/sh"); + // Comfortably under 1 MiB, emitted in one burst then a clean exit. + cmd.args(["-c", "head -c 4096 /dev/zero"]); + let out = run_watchdogged(cmd, Duration::from_secs(5), Duration::from_secs(10)) + .expect("output under the limit must be returned"); + assert!(out.status.success()); + assert_eq!(out.stdout.len(), 4096); + } + + // ---- Windows tree-ownership verification (Will's box) ---------------- + // + // No CI lane executes Windows tests for this helper, so these are + // `#[ignore]`-gated for a sanctioned local run on a real Windows machine: + // + // cargo test -p buzz-desktop --lib bounded_command -- --ignored --nocapture + // + // Both assert on the actual PowerShell-recorded descendant PID (not the + // already-exited root), so neutering the Job Object ownership leaves that + // PID alive and fails the test — the mutation is observable. + + /// True while a Windows process still exists. Opens with the minimal + /// query right and reads its exit code: `STILL_ACTIVE` (259) means running, + /// any other code means exited. A failed open means the PID is gone. + #[cfg(windows)] + fn pid_alive(pid: u32) -> bool { + use windows_sys::Win32::Foundation::{CloseHandle, STILL_ACTIVE}; + use windows_sys::Win32::System::Threading::{ + GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, + }; + unsafe { + let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid); + if handle.is_null() { + return false; + } + let mut code: u32 = 0; + let ok = GetExitCodeProcess(handle, &mut code); + CloseHandle(handle); + ok != 0 && code == STILL_ACTIVE as u32 + } + } + + /// Read a PID that a probe wrote to `path`, retrying briefly since the + /// descendant records it asynchronously. Dumps `logs` on failure so a remote + /// run diagnoses itself instead of panicking blind. + #[cfg(windows)] + fn read_recorded_pid(path: &str, logs: &[&str]) -> u32 { + for _ in 0..200 { + if let Ok(text) = std::fs::read_to_string(path) { + if let Ok(pid) = text.trim().parse::() { + return pid; + } + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!( + "descendant never recorded its PID at {path}\n{}", + dump_logs(logs) + ); + } + + /// Write a PowerShell payload to `path` as a `.ps1` file. Invoking these via + /// `powershell -File` avoids the Rust-std → cmd.exe → powershell quoting + /// gauntlet that silently mangled the inline `-Command` fixtures (the root + /// exited without its payload ever running), so the payload reaches + /// PowerShell verbatim. + #[cfg(windows)] + fn write_ps1(path: &std::path::Path, body: &str) { + std::fs::write(path, body).expect("write .ps1 payload"); + } + + /// Collect the named transcript files (each written by the fixture's + /// PowerShell) into one string for a self-diagnosing assert message. Missing + /// files are reported as such rather than skipped. + #[cfg(windows)] + fn dump_logs(paths: &[&str]) -> String { + let mut out = String::from("---- fixture transcripts ----\n"); + for p in paths { + out.push_str(&format!("[{p}]\n")); + match std::fs::read_to_string(p) { + Ok(text) if text.is_empty() => out.push_str("(empty)\n"), + Ok(text) => { + out.push_str(&text); + if !text.ends_with('\n') { + out.push('\n'); + } + } + Err(e) => out.push_str(&format!("(unreadable: {e})\n")), + } + } + out + } + + // Success path, run in a loop to hammer the spawn/assign race. A PowerShell + // root (no cmd.exe anywhere) launches a hidden, detached PowerShell + // descendant via `Start-Process -WindowStyle Hidden`; the descendant records + // its own PID and sleeps. The root then waits synchronously until the PID + // file is non-empty before exiting 0 — without that wait the root would exit + // in the same tick, the success path would close the kill-on-close job + // immediately, and the descendant would be reaped mid-cold-start before it + // could record its PID, starving the test of its evidence. The descendant is + // still born inside the job (suspend → assign → resume, no breakaway), so the + // reaping guarantee under test is unchanged; only the delivery mechanism (a + // `.ps1` via `-File`, not a mangled inline `-Command`) is fixed. Every assert + // dumps the PowerShell transcripts so a remote failure is self-diagnosing. + #[cfg(windows)] + #[test] + #[ignore = "requires a Windows host; run manually with --ignored"] + fn reaps_backgrounded_descendant_on_success_windows() { + for iteration in 0..25 { + let dir = tempfile::tempdir().expect("temp dir for fixture scripts"); + let pid_path = dir.path().join("descendant.pid"); + let child_ps1 = dir.path().join("child.ps1"); + let root_ps1 = dir.path().join("root.ps1"); + let root_log = dir.path().join("root.log"); + let child_log = dir.path().join("child.log"); + let pid_s = pid_path.to_str().expect("utf-8 pid path"); + let root_log_s = root_log.to_str().expect("utf-8 root log"); + let child_log_s = child_log.to_str().expect("utf-8 child log"); + + write_ps1( + &child_ps1, + &format!( + "$PID | Set-Content -Encoding ascii -Path '{pid_s}'\n\ + Add-Content -Path '{child_log_s}' -Value \"descendant $PID started\"\n\ + Start-Sleep -Seconds 30\n" + ), + ); + write_ps1( + &root_ps1, + &format!( + "Add-Content -Path '{root_log_s}' -Value \"root $PID launching descendant\"\n\ + Start-Process -FilePath 'powershell' -WindowStyle Hidden -ArgumentList \ + '-NoProfile','-ExecutionPolicy','Bypass','-File','{child}'\n\ + $deadline = (Get-Date).AddSeconds(15)\n\ + while (((-not (Test-Path '{pid_s}')) -or ((Get-Item '{pid_s}').Length -eq 0)) \ + -and (Get-Date) -lt $deadline) {{ Start-Sleep -Milliseconds 50 }}\n\ + Add-Content -Path '{root_log_s}' -Value \"root observed pid file, exiting\"\n\ + exit 0\n", + child = child_ps1.to_str().expect("utf-8 child path"), + ), + ); + + let mut cmd = Command::new("powershell"); + cmd.args([ + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + root_ps1.to_str().expect("utf-8 root path"), + ]); + let out = run_watchdogged_raw(cmd, Duration::from_secs(20), Duration::from_secs(40)) + .ok() + .flatten() + .unwrap_or_else(|| { + panic!( + "iteration {iteration}: root exits, so this must return output\n{}", + dump_logs(&[root_log_s, child_log_s]) + ) + }); + assert!( + out.status.success(), + "iteration {iteration}: root must exit 0\nstdout={}\nstderr={}\n{}", + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr), + dump_logs(&[root_log_s, child_log_s]) + ); + + let descendant_pid = read_recorded_pid(pid_s, &[root_log_s, child_log_s]); + std::thread::sleep(Duration::from_millis(300)); + assert!( + !pid_alive(descendant_pid), + "iteration {iteration}: descendant {descendant_pid} must be reaped on success, but it survived\n{}", + dump_logs(&[root_log_s, child_log_s]) + ); + } + } + + // Timeout path: a PowerShell root launches a hidden, detached PowerShell + // descendant (records its PID, sleeps 300s), waits synchronously until the + // PID file is non-empty, then enters its own 300s block so the helper's + // deadline fires inside it. The helper must time out and close the job, + // reaping both. The synchronous wait is the evidence — the descendant's PID + // is recorded before the root reaches the block the deadline fires in, so the + // reap cannot kill it mid-cold-start and starve the assert. Same `.ps1` + // delivery as the success fixture (no cmd tokenizer), and every assert dumps + // the transcripts. + #[cfg(windows)] + #[test] + #[ignore = "requires a Windows host; run manually with --ignored"] + fn reaps_descendant_on_timeout_windows() { + let dir = tempfile::tempdir().expect("temp dir for fixture scripts"); + let pid_path = dir.path().join("descendant.pid"); + let child_ps1 = dir.path().join("child.ps1"); + let root_ps1 = dir.path().join("root.ps1"); + let root_log = dir.path().join("root.log"); + let child_log = dir.path().join("child.log"); + let pid_s = pid_path.to_str().expect("utf-8 pid path"); + let root_log_s = root_log.to_str().expect("utf-8 root log"); + let child_log_s = child_log.to_str().expect("utf-8 child log"); + + write_ps1( + &child_ps1, + &format!( + "$PID | Set-Content -Encoding ascii -Path '{pid_s}'\n\ + Add-Content -Path '{child_log_s}' -Value \"descendant $PID started\"\n\ + Start-Sleep -Seconds 300\n" + ), + ); + write_ps1( + &root_ps1, + &format!( + "Add-Content -Path '{root_log_s}' -Value \"root $PID launching descendant\"\n\ + Start-Process -FilePath 'powershell' -WindowStyle Hidden -ArgumentList \ + '-NoProfile','-ExecutionPolicy','Bypass','-File','{child}'\n\ + $deadline = (Get-Date).AddSeconds(15)\n\ + while (((-not (Test-Path '{pid_s}')) -or ((Get-Item '{pid_s}').Length -eq 0)) \ + -and (Get-Date) -lt $deadline) {{ Start-Sleep -Milliseconds 50 }}\n\ + Add-Content -Path '{root_log_s}' -Value \"root observed pid file, blocking\"\n\ + Start-Sleep -Seconds 300\n", + child = child_ps1.to_str().expect("utf-8 child path"), + ), + ); + + let mut cmd = Command::new("powershell"); + cmd.args([ + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + root_ps1.to_str().expect("utf-8 root path"), + ]); + let result = run_watchdogged_raw(cmd, Duration::from_secs(20), Duration::from_secs(40)) + .unwrap_or_else(|_| { + panic!( + "watchdog expired — output_with_timeout hung on the timeout path\n{}", + dump_logs(&[root_log_s, child_log_s]) + ) + }); + assert!( + result.is_none(), + "a timed-out tree must yield None\n{}", + dump_logs(&[root_log_s, child_log_s]) + ); + + let descendant_pid = read_recorded_pid(pid_s, &[root_log_s, child_log_s]); + std::thread::sleep(Duration::from_millis(300)); + assert!( + !pid_alive(descendant_pid), + "descendant {descendant_pid} must be job-killed on timeout, but it survived\n{}", + dump_logs(&[root_log_s, child_log_s]) + ); + } +} diff --git a/desktop/src-tauri/src/managed_agents/discovery/login_shell.rs b/desktop/src-tauri/src/managed_agents/discovery/login_shell.rs index d8f8e603546..c9109184d5b 100644 --- a/desktop/src-tauri/src/managed_agents/discovery/login_shell.rs +++ b/desktop/src-tauri/src/managed_agents/discovery/login_shell.rs @@ -6,9 +6,15 @@ use std::path::{Path, PathBuf}; use std::process::Command; +use std::time::Duration; use super::is_executable_file; +/// Per-candidate wall-clock bound for a login-shell spawn. Matches the auth +/// probe's 10s discipline: long enough for a healthy interactive shell to +/// source its rc files, short enough that a wedged shell can't stall discovery. +const LOGIN_SHELL_TIMEOUT: Duration = Duration::from_secs(10); + /// Test-only spawn counter lives beside `discovery.rs`; import it here so the /// spawn-record call site stays byte-identical to the pre-extraction source. #[cfg(test)] @@ -34,14 +40,25 @@ pub(crate) fn login_shell_candidates() -> Vec { /// Run a command in a login shell (tries zsh then bash on Unix, Git Bash on Windows). /// Returns trimmed stdout if the command succeeds with non-empty output. +/// +/// Each candidate shell is bounded by [`LOGIN_SHELL_TIMEOUT`]: a shell whose +/// startup blocks (an interactive prompt in `.zshrc`, a stalled network mount, +/// a credential helper waiting on input) is killed and treated as a miss so the +/// loop falls through to the next candidate rather than hanging the whole +/// discovery. Without this bound a single slow login shell froze the forced +/// pipeline indefinitely, which is what left "Check again" spinning forever. fn run_in_login_shell(args: &[&str]) -> Option { #[cfg(test)] login_shell_spawn_probe::record(); for shell in login_shell_candidates() { let mut cmd = Command::new(&shell); cmd.args(args); - crate::util::configure_no_window(&mut cmd); - let Ok(output) = cmd.output() else { + // Window suppression is owned by `output_with_timeout`'s spawn + // (`BOUNDED_CREATION_FLAGS` carries `CREATE_NO_WINDOW`); a + // `configure_no_window` call here would be clobbered by that later + // `creation_flags` set, so it is deliberately omitted. + let Some(output) = super::bounded_command::output_with_timeout(cmd, LOGIN_SHELL_TIMEOUT) + else { continue; }; if !output.status.success() { @@ -72,10 +89,25 @@ enum LoginShellPath { Probed(Option), } -fn path_cache() -> &'static std::sync::Mutex { +/// Cache plus a monotonic generation counter. `refresh_login_shell_path` bumps +/// the generation and resets the state together; a probe records the generation +/// it started under and may only publish its result while that generation is +/// still current. This stops a slow, pre-refresh probe from committing a stale +/// (often false-negative) PATH over the fresh value a post-refresh probe wrote. +struct PathCache { + generation: u64, + state: LoginShellPath, +} + +fn path_cache() -> &'static std::sync::Mutex { use std::sync::{Mutex, OnceLock}; - static CACHE: OnceLock> = OnceLock::new(); - CACHE.get_or_init(|| Mutex::new(LoginShellPath::Uninit)) + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(|| { + Mutex::new(PathCache { + generation: 0, + state: LoginShellPath::Uninit, + }) + }) } fn fetch_login_shell_path_inner() -> Option { @@ -103,45 +135,112 @@ fn fetch_login_shell_path_inner() -> Option { /// to invalidate the cache so the next call re-fetches — e.g. after the user /// installs Node.js mid-session and clicks Retry. /// -/// The lock is never held while the login shell spawns: we check for a cached -/// value, release the lock, run the shell, then re-lock to write. Two concurrent -/// callers may both run the shell (last-writer-wins is fine — both produce the -/// same result), but neither blocks a concurrent agent spawn on the Mutex. +/// The lock is never held while the login shell spawns: we read the cached +/// value and the current generation, release the lock, run the shell, then +/// re-lock to publish. Publication is generation-guarded so a probe that +/// started before a [`refresh_login_shell_path`] can never overwrite the fresh +/// value: if the generation moved while the probe ran, its result is discarded. +/// Within one generation two callers may both probe; a failure/timeout result +/// (`None`) never clobbers an already-committed success, so a slow timeout can't +/// undo a peer's fresh PATH. +/// +/// The caller never returns its own local probe result: after publishing it +/// returns the value now in the cache. This closes two divergences where a +/// caller's own result contradicted the authoritative cache: +/// - same-generation timeout-vs-success — a peer committed a success while +/// our probe timed out (`None`); we return the peer's success, not `None`; +/// - a pre-refresh probe whose writeback was generation-rejected — its local +/// value is stale, so we re-probe under the new generation instead. pub fn login_shell_path() -> Option { - // Fast path: return cached result without spawning a shell. - { - let guard = path_cache().lock().unwrap_or_else(|e| e.into_inner()); - if let LoginShellPath::Probed(ref result) = *guard { - return result.clone(); + loop { + // Fast path: return the cached result and capture the generation the + // probe will run under, all under a single lock. + let generation = { + let guard = path_cache().lock().unwrap_or_else(|e| e.into_inner()); + if let LoginShellPath::Probed(ref result) = guard.state { + return result.clone(); + } + guard.generation + }; + + // Slow path: spawn shell outside any lock. + let result = probe_login_shell_path(); + + // Publish under our generation, then return whatever value is now + // authoritative. `None` means a refresh invalidated our generation + // mid-probe and no fresh value is cached yet, so our `result` is stale + // by definition — discard it and re-probe under the new generation. + // + // Termination: another lap requires another [`refresh_login_shell_path`] + // to land during a probe. Refreshes come only from discrete human + // actions (install/retry/Doctor re-run) and one-shot boot warm, so the + // loop cannot spin unbounded. + if let Some(committed) = publish_probe_result(generation, result) { + return committed; } } +} - // Slow path: spawn shell outside any lock. - let result = fetch_login_shell_path_inner(); +/// Real login-shell probe. A `cfg(test)` seam lets the race tests inject +/// deterministic probe results (and side effects) without spawning shells. +#[cfg(not(test))] +fn probe_login_shell_path() -> Option { + fetch_login_shell_path_inner() +} - // Write back; last-writer-wins is safe here. - { - let mut guard = path_cache().lock().unwrap_or_else(|e| e.into_inner()); - *guard = LoginShellPath::Probed(result.clone()); +#[cfg(test)] +fn probe_login_shell_path() -> Option { + match path_cache_race_tests::take_injected_probe() { + Some(injected) => injected(), + None => fetch_login_shell_path_inner(), } +} - result +/// Commit a probe's `result` under the generation it started with, then report +/// the value the caller should return. +/// +/// A probe whose generation is stale (a [`refresh_login_shell_path`] ran while +/// it was probing) does not commit. Within a live generation a failure/timeout +/// (`None`) never overwrites an already-committed success. This is the sole +/// writer of a probed value, so the two race outcomes are decided here. +/// +/// Returns `Some(v)` — the now-cached probed value the caller must return +/// (its own commit, or a peer's success that superseded it) — or `None` when +/// the cache is `Uninit` because a refresh landed mid-probe, signalling the +/// caller to re-probe under the new generation. Commit and re-read happen under +/// one lock so no refresh can slip between them. +fn publish_probe_result(generation: u64, result: Option) -> Option> { + let mut guard = path_cache().lock().unwrap_or_else(|e| e.into_inner()); + if guard.generation == generation { + let keep_committed_success = + result.is_none() && matches!(guard.state, LoginShellPath::Probed(Some(_))); + if !keep_committed_success { + guard.state = LoginShellPath::Probed(result); + } + } + match guard.state { + LoginShellPath::Probed(ref v) => Some(v.clone()), + LoginShellPath::Uninit => None, + } } /// Invalidate the login-shell PATH cache so the next [`login_shell_path`] call /// re-fetches from a fresh login shell. /// /// Called before every install/retry operation and on Doctor Re-run so a -/// newly-installed tool becomes visible without restarting the app. +/// newly-installed tool becomes visible without restarting the app. Bumping the +/// generation revokes any in-flight probe's writeback, so a shell that started +/// before this refresh cannot recache its now-stale result. pub(crate) fn refresh_login_shell_path() { let mut guard = path_cache().lock().unwrap_or_else(|e| e.into_inner()); - *guard = LoginShellPath::Uninit; + guard.generation = guard.generation.wrapping_add(1); + guard.state = LoginShellPath::Uninit; } #[cfg(test)] pub(crate) fn is_login_shell_path_uninit() -> bool { matches!( - *path_cache().lock().unwrap_or_else(|e| e.into_inner()), + path_cache().lock().unwrap_or_else(|e| e.into_inner()).state, LoginShellPath::Uninit ) } @@ -234,3 +333,178 @@ pub(crate) fn parse_semver_tag(s: &str) -> Option<(u64, u64, u64)> { let patch = patch_str.split('-').next()?.parse::().ok()?; Some((major, minor, patch)) } + +#[cfg(test)] +mod path_cache_race_tests { + use super::*; + use std::collections::VecDeque; + use std::sync::{Mutex, OnceLock}; + + /// A deterministic stand-in for one login-shell spawn. Returning it lets a + /// test drive `login_shell_path`'s slow path without a real shell, and run + /// side effects (a peer commit, a mid-probe refresh) at the exact moment a + /// probe would be executing. + pub(super) type InjectedProbe = Box Option + Send>; + + fn probe_queue() -> &'static Mutex> { + static Q: OnceLock>> = OnceLock::new(); + Q.get_or_init(|| Mutex::new(VecDeque::new())) + } + + /// Consumed by the `cfg(test)` `probe_login_shell_path` seam: each slow-path + /// probe pops the next injected result, falling back to the real shell when + /// the queue is empty (so unrelated cache tests still exercise real probing). + pub(super) fn take_injected_probe() -> Option { + probe_queue() + .lock() + .unwrap_or_else(|e| e.into_inner()) + .pop_front() + } + + fn inject_probes(probes: Vec) { + let mut q = probe_queue().lock().unwrap_or_else(|e| e.into_inner()); + q.clear(); + q.extend(probes); + } + + fn cached_probe() -> Option> { + match path_cache().lock().unwrap_or_else(|e| e.into_inner()).state { + LoginShellPath::Uninit => None, + LoginShellPath::Probed(ref v) => Some(v.clone()), + } + } + + fn generation() -> u64 { + path_cache() + .lock() + .unwrap_or_else(|e| e.into_inner()) + .generation + } + + /// A probe that started before a refresh must not recache its stale result. + /// Models the P1 interleaving: probe A captures generation G; a forced + /// refresh bumps to G+1 and (via probe B) commits a fresh PATH; then A + /// finishes late and tries to publish. A publishes a non-empty *success* + /// (`/stale/bin`), which the same-generation `None`-over-`Some` rule would + /// accept — so only the generation guard can reject it. This keeps the test + /// non-vacuous: delete the generation comparison and stale overwrites fresh. + #[test] + fn stale_probe_cannot_commit_after_refresh() { + let _guard = crate::managed_agents::lock_path_mutex(); + refresh_login_shell_path(); + + // Probe A starts here. + let gen_a = generation(); + + // A forced refresh invalidates the cache; probe B (new generation) then + // commits a fresh PATH. + refresh_login_shell_path(); + let gen_b = generation(); + assert_ne!(gen_a, gen_b, "refresh must bump the generation"); + publish_probe_result(gen_b, Some("/fresh/bin".to_string())); + + // Probe A finishes late and tries to publish a *stale success* under + // its old generation. Only the generation guard can reject this — the + // same-generation success-retention rule would let a `Some` through. + publish_probe_result(gen_a, Some("/stale/bin".to_string())); + + assert_eq!( + cached_probe(), + Some(Some("/fresh/bin".to_string())), + "a pre-refresh probe must not overwrite the post-refresh fresh PATH" + ); + + // Restore the shared cache so sibling tests re-probe a real PATH rather + // than reading this fixture value. + refresh_login_shell_path(); + } + + /// Within one generation a slow failure/timeout must not clobber a peer's + /// already-committed success. Two cold callers race under generation G: the + /// success lands first, the timeout (`None`) lands second and is dropped. + #[test] + fn timeout_does_not_clobber_committed_success() { + let _guard = crate::managed_agents::lock_path_mutex(); + refresh_login_shell_path(); + let gen = generation(); + + // Caller 1 succeeds. + publish_probe_result(gen, Some("/usr/local/bin".to_string())); + // Caller 2 times out later in the same generation. + publish_probe_result(gen, None); + + assert_eq!( + cached_probe(), + Some(Some("/usr/local/bin".to_string())), + "a same-generation timeout must not overwrite a committed success" + ); + + // Restore the shared cache so sibling tests re-probe a real PATH rather + // than reading this fixture value. + refresh_login_shell_path(); + } + + /// P1 #2, divergence (a): same-generation timeout-vs-success. A caller + /// whose own probe times out (`None`) must still return the success a peer + /// committed under the same generation — never its own `None`, which would + /// let a forced discovery on this thread settle a PATH-missing UI while the + /// authoritative cache holds the peer's success. + /// + /// Injected probe: commit the peer's `/peer/bin` success, then return `None` + /// (this caller's timeout). Non-vacuous for the "return authoritative value" + /// rule: return the local result instead and this yields `None`. + #[test] + fn caller_returns_peer_success_not_own_timeout() { + let _guard = crate::managed_agents::lock_path_mutex(); + refresh_login_shell_path(); + let gen = generation(); + + inject_probes(vec![Box::new(move || { + // A peer probe finishes first and commits a success under gen. + publish_probe_result(gen, Some("/peer/bin".to_string())); + // Our probe then times out. + None + })]); + + assert_eq!( + login_shell_path(), + Some("/peer/bin".to_string()), + "a timed-out caller must return the peer's committed success, not its own None" + ); + + refresh_login_shell_path(); + } + + /// P1 #2, divergence (b): a pre-refresh probe whose writeback is + /// generation-rejected must not return its stale local value; the caller + /// re-probes under the new generation and returns the fresh result. + /// + /// First injected probe refreshes mid-flight (bumping the generation) and + /// returns a stale `/stale/bin`; publication is rejected, so the caller + /// loops and the second probe returns the fresh `/fresh/bin`. Non-vacuous + /// for the re-probe rule: return the stale local value on a rejected commit + /// instead and this yields `/stale/bin`. + #[test] + fn caller_reprobes_after_midprobe_refresh() { + let _guard = crate::managed_agents::lock_path_mutex(); + refresh_login_shell_path(); + + inject_probes(vec![ + Box::new(|| { + // A forced refresh lands while this probe runs, invalidating the + // generation it started under; its result is stale by definition. + refresh_login_shell_path(); + Some("/stale/bin".to_string()) + }), + Box::new(|| Some("/fresh/bin".to_string())), + ]); + + assert_eq!( + login_shell_path(), + Some("/fresh/bin".to_string()), + "a generation-rejected probe must re-probe, never return its stale local value" + ); + + refresh_login_shell_path(); + } +} diff --git a/desktop/src-tauri/src/managed_agents/discovery/tests/managed_path_resolution.rs b/desktop/src-tauri/src/managed_agents/discovery/tests/managed_path_resolution.rs index 5369b6321b7..aab5cd45298 100644 --- a/desktop/src-tauri/src/managed_agents/discovery/tests/managed_path_resolution.rs +++ b/desktop/src-tauri/src/managed_agents/discovery/tests/managed_path_resolution.rs @@ -187,3 +187,30 @@ fn cheap_discovery_never_spawns_login_shell_even_when_cold() { "the forced path must probe the absent command via login shell at least once, got {forced}" ); } + +/// Regression: `resolve_command_cached` (the cheap discovery path) must find a +/// bundled sidecar sitting next to the executable via a filesystem stat, even +/// with a cold resolve cache. Before the fix it consulted only the managed-shim +/// dirs + cache, so `buzz-agent` reported "not installed" at every cold launch. +/// Here the path form exercises the same `resolve_workspace_command` stat the +/// cheap path now shares. +#[cfg(unix)] +#[test] +fn cheap_path_resolves_workspace_sidecar_without_cache() { + use crate::managed_agents::discovery::resolve_command_cached; + use std::os::unix::fs::PermissionsExt; + + let dir = std::env::temp_dir().join(format!("buzz-sidecar-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("create temp dir"); + let bin = dir.join("buzz-agent"); + std::fs::write(&bin, "#!/bin/sh\n").expect("write sidecar"); + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).expect("chmod"); + + assert_eq!( + resolve_command_cached(bin.to_str().expect("utf8 path")), + Some(bin.clone()), + "cheap path must resolve a bundled sidecar by path with a cold cache" + ); + + let _ = std::fs::remove_dir_all(dir); +} diff --git a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs index 479d6ec913e..8e27ba1031d 100644 --- a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs +++ b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs @@ -45,20 +45,19 @@ impl Drop for JobHandle { /// the caller can fall back to `Child::kill()` — a degraded teardown beats a /// failed spawn. /// -/// 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. +/// For the harness spawn path ([`finish_spawn`]) assignment happens immediately +/// after a normal spawn. The child (buzz-acp) must init tokio, parse its config, +/// and spawn 24 children (tens-to-hundreds of ms) before any descendant exists, +/// so the microsecond `OpenProcess` + `AssignProcessToJobObject` reliably wins +/// that race. 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 { +/// The discovery path (`bounded_command`) runs arbitrary probe commands that +/// can background a descendant and exit in the same tick, so it cannot rely on +/// assign-latency. It spawns with `CREATE_SUSPENDED`, assigns the frozen child +/// here, then calls [`resume_process`] — no descendant can exist until the job +/// owns the root, closing the race by construction. +pub(crate) fn create_job_for_child(pid: u32) -> Option { use std::ptr::null; use windows_sys::Win32::Foundation::{CloseHandle, FALSE}; use windows_sys::Win32::System::JobObjects::{ @@ -105,6 +104,56 @@ fn create_job_for_child(pid: u32) -> Option { } } +/// Resume a process spawned with `CREATE_SUSPENDED` by resuming every thread it +/// owns. A fresh `CREATE_SUSPENDED` process has exactly one thread suspended at +/// its entry point; resuming it lets the process run. We enumerate via a +/// ToolHelp thread snapshot filtered to `pid` rather than tracking the initial +/// thread id (`std::process::Command` does not expose it), and resume each so +/// the walk is correct even in the pathological multi-thread case. +/// +/// Returns `true` only if at least one owned thread was resumed. `false` means +/// no thread could be resumed — the caller must treat the child as unusable and +/// tear it down, since a still-suspended root would otherwise hang to the +/// deadline. +pub(crate) fn resume_process(pid: u32) -> bool { + use windows_sys::Win32::Foundation::{CloseHandle, 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 false; + } + + let mut entry: THREADENTRY32 = std::mem::zeroed(); + entry.dwSize = std::mem::size_of::() as u32; + + let mut resumed_any = false; + let mut has_entry = Thread32First(snapshot, &mut entry); + while has_entry != 0 { + if entry.th32OwnerProcessID == pid { + let thread = OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID); + if !thread.is_null() { + // ResumeThread returns u32::MAX on failure; any other value + // is the thread's previous suspend count. + if ResumeThread(thread) != u32::MAX { + resumed_any = true; + } + CloseHandle(thread); + } + } + entry.dwSize = std::mem::size_of::() as u32; + has_entry = Thread32Next(snapshot, &mut entry); + } + + CloseHandle(snapshot); + resumed_any + } +} + /// 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 diff --git a/desktop/src/app/useAppShellLifecycleEffects.ts b/desktop/src/app/useAppShellLifecycleEffects.ts index 71db0523695..fcdc29fc5fd 100644 --- a/desktop/src/app/useAppShellLifecycleEffects.ts +++ b/desktop/src/app/useAppShellLifecycleEffects.ts @@ -1,5 +1,7 @@ import * as React from "react"; +import { useQueryClient } from "@tanstack/react-query"; +import { startBootWarm } from "@/features/agents/acpRuntimesQuery"; import { setDesktopAppBadge } from "@/features/notifications/lib/desktop"; import { useForegroundQueryRefresh } from "@/features/workflows/hooks"; import { relayClient } from "@/shared/api/relayClient"; @@ -23,6 +25,22 @@ export function useAppShellLifecycleEffects({ useRelayResumeTriggers(); useForegroundQueryRefresh(); + // Warm the ACP runtime catalog once at app launch. The shared runtime-catalog + // cache is in-memory only, so it starts cold every boot; the cheap discovery + // path reports every harness as "(not installed)" until a forced pass warms + // it. The create/edit picker and Agents > Agent defaults surfaces read that + // cheap path, so without this warm they render all-missing (and block agent + // save) until the user visits Settings > Agents — the accidental workaround. + // `startBootWarm` drives the module-level boot-warm gate (once per launch, so + // this remounting effect never re-fires the probe) which makes those cheap + // surfaces show loading/retryable-error instead of blessing the cold catalog, + // and swallows the probe's own errors so a failure leaves the last good + // catalog in place without an unhandled rejection. + const queryClient = useQueryClient(); + React.useEffect(() => { + void startBootWarm(queryClient); + }, [queryClient]); + // Prevent webview file:/// navigation on file drop outside the composer. // Scoped to file drags only (text drag-and-drop into inputs still works). // Composer's onDrop fires first (React synthetic before window bubble). diff --git a/desktop/src/features/agents/acpRuntimesQuery.test.mjs b/desktop/src/features/agents/acpRuntimesQuery.test.mjs index c51dea05b8f..e9320bb6e99 100644 --- a/desktop/src/features/agents/acpRuntimesQuery.test.mjs +++ b/desktop/src/features/agents/acpRuntimesQuery.test.mjs @@ -179,12 +179,15 @@ globalThis.__TAURI_INTERNALS__ = { import React from "react"; import { createRoot } from "react-dom/client"; import { act } from "react"; -import { QueryClient } from "@tanstack/react-query"; +import { QueryClient, QueryObserver } from "@tanstack/react-query"; import { QueryClientProvider } from "@tanstack/react-query"; import { acpRuntimesQueryKey, + applyBootWarmGate, + getBootWarmSnapshot, refreshAcpRuntimes, + startBootWarm, useAcpRuntimesQueryForced, } from "./acpRuntimesQuery.ts"; import { discoverAcpRuntimes } from "@/shared/api/tauriAcpDiscovery.ts"; @@ -230,6 +233,144 @@ afterEach(() => { discoverHandler = () => Promise.resolve([]); }); +// Runs FIRST so the process-global boot-warm gate is observed from `idle`. +// Covers Carl's ask: the cheap/forced race (a cold cheap catalog must read as +// loading, not authoritative, while the first forced pass is in flight) and the +// failure state (a failed forced pass must surface a retryable error carrying +// the real reason, not a silent empty catalog), plus recovery on retry. +describe("boot-warm gate drives cheap consumers through the initial pass", () => { + it("applyBootWarmGate: a non-empty cold catalog is not authoritative while pending or failed", () => { + // The real cold cheap response is NEVER empty: discovery always emits the + // known runtimes as not_installed/cli_missing rows plus presets. Model that + // wire shape so the gate is exercised against the payload it exists to + // gate, not a `[]` that never occurs in production. + const coldCatalog = { + data: [ + rawEntry("codex", "unknown"), + rawEntry("goose", "unknown"), + rawEntry("claude-code", "unknown"), + ], + error: null, + isLoading: false, + isPending: false, + isFetching: false, + isError: false, + }; + // A consumer maps `isLoading -> "loading"`, `isError -> "error"`, else + // `"ready"`. "Ready" is what blesses the cold rows as authoritative — the + // exact P2 defect. Assert neither pending nor failed reads as ready. + const readsAsReady = (q) => !q.isLoading && !q.isError; + + const pending = applyBootWarmGate(coldCatalog, { + status: "pending", + error: null, + }); + assert.equal(pending.isLoading, true); + assert.equal(pending.isPending, true); + assert.equal( + readsAsReady(pending), + false, + "pending must not read as ready", + ); + // The catalog rows are preserved so a consumer reading `data ?? []` keeps + // them; only the lifecycle flags are overlaid. + assert.equal(pending.data.length, 3); + + const reason = new Error("PATH probe timed out"); + const failed = applyBootWarmGate(coldCatalog, { + status: "failed", + error: reason, + }); + assert.equal(failed.isError, true); + assert.equal(failed.error, reason); + assert.equal(readsAsReady(failed), false, "failed must not read as ready"); + assert.equal(failed.data.length, 3); + + // idle/settled pass through untouched: onboarding renders before the warm + // starts (idle) and the warmed hot path (settled) must both read as ready. + for (const status of ["idle", "settled"]) { + const passed = applyBootWarmGate(coldCatalog, { status, error: null }); + assert.equal(passed.isLoading, false); + assert.equal(passed.isError, false); + assert.equal(readsAsReady(passed), true, `${status} must read as ready`); + } + }); + + it("applyBootWarmGate: a warmed non-empty catalog reads as ready once settled", () => { + const warm = { + data: [rawEntry("codex", "logged_in")], + error: null, + isLoading: false, + isPending: false, + isFetching: false, + isError: false, + }; + const settled = applyBootWarmGate(warm, { status: "settled", error: null }); + assert.equal(settled.isLoading, false); + assert.equal(settled.isError, false); + assert.equal(settled.data.length, 1); + }); + + it("applyBootWarmGate: failed reads as a retryable error with the real reason", () => { + const cold = { + data: [], + error: null, + isLoading: true, + isPending: true, + isFetching: true, + isError: false, + }; + const reason = new Error("PATH probe timed out"); + const failed = applyBootWarmGate(cold, { status: "failed", error: reason }); + assert.equal(failed.isError, true); + assert.equal(failed.error, reason); + assert.equal(failed.isLoading, false, "a failed warm is not still loading"); + }); + + it("startBootWarm: failure marks the gate failed, a retry settles it", async () => { + assert.equal( + getBootWarmSnapshot().status, + "idle", + "gate must start idle before any warm", + ); + + const queryClient = makeQueryClient(); + queryClient.mount(); + + // 1. First forced pass fails: the gate goes `failed` and captures the + // reason, so cold cheap surfaces can show a retryable error. + let failForced = true; + discoverHandler = (args) => + args?.force === true && failForced + ? Promise.reject(new Error("discovery boom")) + : Promise.resolve([]); + await startBootWarm(queryClient); + assert.equal(getBootWarmSnapshot().status, "failed"); + assert.equal(getBootWarmSnapshot().error?.message, "discovery boom"); + + // 2. A retry that succeeds settles the gate and clears the error, so cheap + // consumers stop overlaying and render the warmed catalog. + failForced = false; + discoverHandler = () => Promise.resolve([rawEntry("codex", "logged_in")]); + await startBootWarm(queryClient); + assert.equal(getBootWarmSnapshot().status, "settled"); + assert.equal(getBootWarmSnapshot().error, null); + + // 3. Once settled, further boot warms are no-ops (fixes the per-remount + // re-fire): no additional forced probe fires. + const before = calls.filter( + (c) => c.command === "discover_acp_providers" && c.args?.force === true, + ).length; + await startBootWarm(queryClient); + const after = calls.filter( + (c) => c.command === "discover_acp_providers" && c.args?.force === true, + ).length; + assert.equal(after, before, "a settled gate must not re-fire the probe"); + + queryClient.unmount(); + }); +}); + describe("refreshAcpRuntimes cannot dedup onto an in-flight cheap request", () => { it("runs a distinct force:true probe and writes it into the shared cache", async () => { const queryClient = makeQueryClient(); @@ -275,6 +416,59 @@ describe("refreshAcpRuntimes cannot dedup onto an in-flight cheap request", () = queryClient.unmount(); }); + + it("an in-flight cheap query cannot clobber the forced result after refresh", async () => { + // Carl's settle-order finding: a cheap query in flight on the shared key + // must not land its (older) result after the forced catalog is written. + // `refreshAcpRuntimes` cancels the shared-key query before settling; this + // proves the cancel is load-bearing by holding a real cheap observer + // fetching, running the forced refresh, then resolving the cheap request + // late — its result must not overwrite the forced catalog, and the gate + // must settle on the forced state. (Removing the `cancelQueries` call makes + // the late cheap result win and fails this test.) + const queryClient = makeQueryClient(); + queryClient.mount(); + + // Seed a pre-existing cold catalog, then start a mounted cheap observer that + // refetches and is held pending — the real in-flight shape. + queryClient.setQueryData(acpRuntimesQueryKey, [ + rawEntry("codex", "unknown"), + ]); + const cheap = deferred(); + discoverHandler = (args) => { + if (args?.force === false) return cheap.promise; + return Promise.resolve([rawEntry("codex", "logged_in")]); + }; + const observer = new QueryObserver(queryClient, { + queryKey: acpRuntimesQueryKey, + queryFn: () => discoverAcpRuntimes(), + staleTime: 0, + }); + const unsubscribe = observer.subscribe(() => {}); + await new Promise((r) => setImmediate(r)); + + // Forced refresh completes and settles while the cheap observer is fetching. + await refreshAcpRuntimes(queryClient); + + // The cheap request resolves afterward; its result must be dropped. + cheap.resolve([rawEntry("codex", "unknown")]); + await new Promise((r) => setImmediate(r)); + await new Promise((r) => setImmediate(r)); + + assert.equal( + queryClient.getQueryData(acpRuntimesQueryKey)?.[0]?.authStatus.status, + "logged_in", + "shared cache must remain the forced result after a late cheap resolution", + ); + assert.equal( + getBootWarmSnapshot().status, + "settled", + "the gate must settle on the forced catalog, not the stale cheap state", + ); + + unsubscribe(); + queryClient.unmount(); + }); }); describe("useAcpRuntimesQueryForced surfaces forced-probe failures", () => { diff --git a/desktop/src/features/agents/acpRuntimesQuery.ts b/desktop/src/features/agents/acpRuntimesQuery.ts index 0e76e25ee76..16f82a0aa95 100644 --- a/desktop/src/features/agents/acpRuntimesQuery.ts +++ b/desktop/src/features/agents/acpRuntimesQuery.ts @@ -19,6 +19,153 @@ export const acpRuntimesQueryKey = ["acp-runtimes"] as const; */ export const acpRuntimesForcedQueryKey = ["acp-runtimes", "forced"] as const; +/** + * Boot-warm gate for the *initial* forced discovery pass. + * + * The shared runtime catalog is in-memory only, so it starts cold every launch: + * the cheap discovery path reports every harness `(not installed)` until a + * forced pass warms it. Without a gate, the create/edit picker and Agents > + * Agent defaults surfaces read that cheap path and present the cold catalog as + * *authoritative* — blessing every harness as unavailable and blocking save — + * during the 20–65s boot probe, and forever if that probe fails. + * + * This module-level state lets cheap consumers (`useAcpRuntimesQuery`) treat the + * catalog as still-loading while the first forced pass is in flight and as a + * retryable error if it failed, instead of authoritative. It is process-global + * (one launch), so `startBootWarm` runs the warm exactly once no matter how many + * times `AppShell` mounts — that also fixes the per-remount re-fire. + * + * The seam that protects onboarding (which renders before `AppShell` fires the + * warm): the gate only overlays loading/error once the warm has *started* + * (`pending`/`failed`). While `idle` — no warm yet, e.g. the onboarding flow — + * cheap consumers behave exactly as before. A successful forced refresh from any + * surface settles the gate, so onboarding's own forced warm clears it too. + */ +export type AcpBootWarmStatus = "idle" | "pending" | "settled" | "failed"; + +/** + * A stable snapshot object for `useSyncExternalStore`: `getSnapshot` must return + * a referentially-stable value between changes, so the object is rebuilt only in + * `setBootWarm`, never per read. + */ +let bootWarmSnapshot: { status: AcpBootWarmStatus; error: Error | null } = { + status: "idle", + error: null, +}; +const bootWarmListeners = new Set<() => void>(); + +function setBootWarm(status: AcpBootWarmStatus, error: Error | null) { + if (bootWarmSnapshot.status === status && bootWarmSnapshot.error === error) { + return; + } + bootWarmSnapshot = { status, error }; + for (const listener of bootWarmListeners) listener(); +} + +export function subscribeBootWarm(listener: () => void) { + bootWarmListeners.add(listener); + return () => { + bootWarmListeners.delete(listener); + }; +} + +export function getBootWarmSnapshot() { + return bootWarmSnapshot; +} + +/** + * Overlay the launch boot-warm gate onto a cheap-path query result so cheap + * consumers never present a cold catalog as authoritative. Pure so it can be + * unit-tested without a mounted hook. + * + * The cheap backend response is *never* empty on a cold cache — discovery + * always emits the full set of known runtimes (as `not_installed`/`cli_missing` + * rows) plus presets. Gating on `data.length` would therefore be a no-op for the + * exact payload this exists to gate, so the gate keys on the boot-warm state + * instead and always preserves `query.data`: + * + * - `pending` (first forced pass in flight) reads as loading, so a cold catalog + * is presented as still-loading rather than a settled "everything + * unavailable" list — even though those cold rows are non-empty. + * - `failed` (forced pass rejected) reads as a retryable error carrying the + * probe's real reason. + * - `idle`/`settled` pass the query through unchanged, so onboarding (which + * renders before the warm starts) and the warmed hot path are untouched. + * + * `query.data` is preserved on every branch: overlaying only the lifecycle + * flags means a consumer that reads `data ?? []` keeps its rows while a + * status-driven consumer correctly treats them as not-yet-authoritative. + */ +export function applyBootWarmGate< + Q extends { + data?: unknown[]; + error: Error | null; + isLoading: boolean; + isPending: boolean; + isFetching: boolean; + isError: boolean; + }, +>(query: Q, bootWarm: { status: AcpBootWarmStatus; error: Error | null }): Q { + if (bootWarm.status === "pending") { + return { ...query, isLoading: true, isPending: true, isFetching: true }; + } + if (bootWarm.status === "failed") { + return { + ...query, + isError: true, + error: bootWarm.error ?? query.error, + isLoading: false, + }; + } + return query; +} + +/** + * Run the initial forced discovery pass once per launch and drive the boot-warm + * gate. `AppShell` calls this on mount; the `pending`/`settled` short-circuit + * makes remounts no-ops (fixing the re-fire) while still retrying after a prior + * failure. Success is recorded by `refreshAcpRuntimes` itself (any forced + * success settles the gate); this only has to mark its own failure. + */ +export async function startBootWarm( + queryClient: ReturnType, +) { + const status: AcpBootWarmStatus = bootWarmSnapshot.status; + if (status === "pending" || status === "settled") { + return; + } + setBootWarm("pending", null); + const result = await refreshAcpRuntimes(queryClient); + // A concurrent forced success may have already settled the gate; only mark + // failed if this pass is still the pending one and it returned no catalog. + if (result === undefined && bootWarmSnapshot.status === "pending") { + setBootWarm("failed", lastForcedError); + } +} + +/** + * A stable callback that re-runs the boot warm after it failed, for the retry + * affordance the cheap-path surfaces (create/edit picker, Agent defaults) show + * when the gate is in its `failed` state. `startBootWarm` is the retry + * primitive: from `failed` it transitions back through `pending` (so the + * surface shows loading again) to `settled` on success or `failed` with a fresh + * reason on another rejection. It no-ops while `pending`/`settled`, so a + * double-click cannot stack probes. + */ +export function useRetryBootWarm() { + const queryClient = useQueryClient(); + return React.useCallback(() => { + void startBootWarm(queryClient); + }, [queryClient]); +} + +/** + * The error from the most recent failed forced probe, surfaced through the + * boot-warm `failed` state so a cold catalog shows a real reason rather than a + * silent empty list. Cleared on the next forced success. + */ +let lastForcedError: Error | null = null; + /** * Run a forced (full re-discovery) refresh and write the result into the shared * runtime-catalog cache. @@ -48,13 +195,20 @@ export async function refreshAcpRuntimes( staleTime: 0, gcTime: 0, }); - queryClient.setQueryData(acpRuntimesQueryKey, result); - // A hot-surface cheap fetch may already be in flight on the shared key; cancel - // it so its (older, cached) result cannot land after and clobber the fresh - // forced catalog we just wrote. + // Cancel and *await* the in-flight cheap query on the shared key BEFORE + // writing the forced result. `cancelQueries` defaults to `revert: true`, so + // cancellation restores the cheap query's pre-fetch state; doing it after + // `setQueryData` would let that revert land last and clobber the fresh + // forced catalog, and the gate would then settle on the stale state. With + // the cancel awaited first, our `setQueryData` is the final write. await queryClient.cancelQueries({ queryKey: acpRuntimesQueryKey }); + queryClient.setQueryData(acpRuntimesQueryKey, result); + // Any forced success proves the catalog is warm: settle the boot-warm gate + // and clear the last error, so cheap consumers stop overlaying loading/error. + lastForcedError = null; + setBootWarm("settled", null); return result; - } catch { + } catch (error) { // The forced probe rejected. `fetchQuery` has already recorded the error in // the forced key's query state, where `useAcpRuntimesQueryForced` projects // it into the hook's returned `error`/`isError`. Swallow the rejection here @@ -63,7 +217,9 @@ export async function refreshAcpRuntimes( // paths) can keep `void refreshAcpRuntimes(...)` without ever leaking an // unhandled rejection, and a new call site can never reintroduce one. The // shared cache is left untouched so consumers keep the last good catalog - // alongside the surfaced error. + // alongside the surfaced error. Record the error so a failed boot warm can + // surface a real reason on the cheap-path surfaces (via the boot-warm gate). + lastForcedError = error instanceof Error ? error : new Error(String(error)); return undefined; } } diff --git a/desktop/src/features/agents/hooks.ts b/desktop/src/features/agents/hooks.ts index b7dd7667334..3daf4fa78cc 100644 --- a/desktop/src/features/agents/hooks.ts +++ b/desktop/src/features/agents/hooks.ts @@ -52,9 +52,15 @@ import { import { bootstrapManagedAgentRuntimePairs } from "@/features/agents/managedAgentRuntimeHooks"; import { acpRuntimesQueryKey, + applyBootWarmGate, + getBootWarmSnapshot, refreshAcpRuntimes, + subscribeBootWarm, +} from "@/features/agents/acpRuntimesQuery"; +export { + useAcpRuntimesQueryForced, + useRetryBootWarm, } from "@/features/agents/acpRuntimesQuery"; -export { useAcpRuntimesQueryForced } from "@/features/agents/acpRuntimesQuery"; import { createPersona, deletePersona, @@ -218,12 +224,23 @@ function invalidateManagedAgentQueriesInBackground( * probe pipeline. */ export function useAcpRuntimesQuery(options?: { enabled?: boolean }) { - return useQuery({ + const query = useQuery({ enabled: options?.enabled ?? true, queryKey: acpRuntimesQueryKey, queryFn: () => discoverAcpRuntimes(), staleTime: 30 * 60_000, }); + // Overlay the launch boot-warm gate so cheap consumers never present a cold + // catalog as authoritative: until the first forced pass settles, an un-warmed + // catalog reads as loading (`pending`) or a retryable error (`failed`) rather + // than "every harness not installed". `applyBootWarmGate` preserves an + // already-good list and passes through untouched while idle/settled. + const bootWarm = React.useSyncExternalStore( + subscribeBootWarm, + getBootWarmSnapshot, + getBootWarmSnapshot, + ); + return applyBootWarmGate(query, bootWarm); } export function useAvailableAcpRuntimes(options?: { enabled?: boolean }) { diff --git a/desktop/src/features/agents/ui/AgentDefaultsEditor.tsx b/desktop/src/features/agents/ui/AgentDefaultsEditor.tsx index 69e8b3a5b43..3564f31cb50 100644 --- a/desktop/src/features/agents/ui/AgentDefaultsEditor.tsx +++ b/desktop/src/features/agents/ui/AgentDefaultsEditor.tsx @@ -33,6 +33,7 @@ import { sortPersonaRuntimes, } from "@/features/agents/ui/agentConfigOptions"; import { AgentDropdownSelect } from "@/features/agents/ui/agentConfigControls"; +import { HarnessCatalogRetryNotice } from "@/features/agents/ui/HarnessCatalogRetryNotice"; import { AgentConfigFields, EMPTY_GLOBAL_CONFIG, @@ -171,10 +172,12 @@ export function AgentDefaultsEditor({ [sortedRuntimes], ); const configSurfaceLoading = isLoading || runtimesQuery.isLoading; - const configSurfaceError = - loadError || + // The runtime catalog failing to warm is retryable in-place (re-run the boot + // probe); a global-config load failure is not, so it keeps the restart copy. + const runtimeCatalogError = runtimesQuery.isError || - (!configSurfaceLoading && sortedRuntimes.length === 0); + (!configSurfaceLoading && !loadError && sortedRuntimes.length === 0); + const configSurfaceError = loadError || runtimeCatalogError; function handleConfigChange(next: GlobalAgentConfig) { configRef.current = next; @@ -268,10 +271,16 @@ export function AgentDefaultsEditor({ Loading… ) : configSurfaceError ? ( -
- - Couldn't load agent defaults. Restart the app to try again. -
+ runtimeCatalogError && !loadError ? ( +
+ +
+ ) : ( +
+ + Couldn't load agent defaults. Restart the app to try again. +
+ ) ) : ( <>
diff --git a/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx b/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx index 06f41667b09..81033f7d928 100644 --- a/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx +++ b/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx @@ -823,6 +823,7 @@ export function AgentDefinitionDialog({ > {aiConfigurationMode === "custom" ? ( ) : null} - {llmProviderFieldVisible && aiConfigurationMode === "custom" ? (
void; options: PersonaDropdownOption[]; @@ -34,7 +37,7 @@ export function AgentHarnessField({ placeholder={placeholder} value={value} /> - {warning} + {catalogStatus === "error" ? : warning}
); } diff --git a/desktop/src/features/agents/ui/HarnessCatalogRetryNotice.tsx b/desktop/src/features/agents/ui/HarnessCatalogRetryNotice.tsx new file mode 100644 index 00000000000..34efb54c94d --- /dev/null +++ b/desktop/src/features/agents/ui/HarnessCatalogRetryNotice.tsx @@ -0,0 +1,24 @@ +import { AlertCircle } from "lucide-react"; + +import { useRetryBootWarm } from "@/features/agents/hooks"; +import { Button } from "@/shared/ui/button"; + +/** + * Inline error affordance shown when the launch runtime-catalog warm failed + * (the boot-warm gate's `failed` state). Unlike a global-config load failure — + * which is not retryable and keeps the "restart the app" copy — a failed + * harness probe re-runs in place via `useRetryBootWarm`, so the create/edit + * picker and Agent defaults surfaces both render this instead of a dead end. + */ +export function HarnessCatalogRetryNotice() { + const retryBootWarm = useRetryBootWarm(); + return ( +
+ + Couldn't detect agent harnesses. + +
+ ); +}