From a40114c54199c718de0506aa8bda839518cd0a6a Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Sat, 5 Sep 2026 15:31:05 +0800 Subject: [PATCH 1/2] Harden Desktop lifecycle deadlines and persistence --- apps/desktop/src-tauri/Cargo.lock | 1 + apps/desktop/src-tauri/Cargo.toml | 5 +- apps/desktop/src-tauri/src/activity.rs | 1 + apps/desktop/src-tauri/src/deadline.rs | 57 ++ apps/desktop/src-tauri/src/lib.rs | 3 +- apps/desktop/src-tauri/src/platform/macos.rs | 44 - apps/desktop/src-tauri/src/platform/mod.rs | 63 -- .../desktop/src-tauri/src/platform/windows.rs | 25 - apps/desktop/src-tauri/src/process/mod.rs | 3 +- apps/desktop/src-tauri/src/process/owned.rs | 52 -- .../src-tauri/src/process/supervisor.rs | 456 ++++++++--- apps/desktop/src-tauri/src/state.rs | 750 ++++++++++++++++-- .../desktop/src-tauri/src/webcodex/adapter.rs | 207 +++-- apps/desktop/src-tauri/src/webcodex/cli.rs | 221 +++++- .../src/bin/process_tree_helper.rs | 30 +- .../webcodex-process/tests/managed_child.rs | 43 + crates/webcodex-runner/src/main.rs | 24 +- .../src/main_tests/runner_config.rs | 22 + .../src/webcodex_runner/transport.rs | 37 + src/bin/webcodex-server.rs | 5 +- src/lib.rs | 37 +- src/server_shutdown.rs | 49 +- 22 files changed, 1656 insertions(+), 479 deletions(-) create mode 100644 apps/desktop/src-tauri/src/deadline.rs delete mode 100644 apps/desktop/src-tauri/src/platform/macos.rs delete mode 100644 apps/desktop/src-tauri/src/process/owned.rs diff --git a/apps/desktop/src-tauri/Cargo.lock b/apps/desktop/src-tauri/Cargo.lock index fee859338..d66d84d3d 100644 --- a/apps/desktop/src-tauri/Cargo.lock +++ b/apps/desktop/src-tauri/Cargo.lock @@ -3999,6 +3999,7 @@ dependencies = [ "tokio", "url", "webcodex-process", + "windows-sys 0.61.2", ] [[package]] diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 353638b05..9ad603519 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -24,6 +24,9 @@ tokio = { version = "1", features = ["fs", "io-util", "macros", "process", "rt-m url = "2" webcodex-process = { path = "../../../crates/webcodex-process" } -[target.'cfg(target_os = "macos")'.dependencies] +[target.'cfg(unix)'.dependencies] libc = "0.2" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61.2", features = ["Win32_Storage_FileSystem"] } + diff --git a/apps/desktop/src-tauri/src/activity.rs b/apps/desktop/src-tauri/src/activity.rs index b75cbdef5..f46a55b78 100644 --- a/apps/desktop/src-tauri/src/activity.rs +++ b/apps/desktop/src-tauri/src/activity.rs @@ -33,6 +33,7 @@ pub enum ActivityEventKind { RegularTunnelReady, RegularTunnelStopped, RuntimeStopped, + StateRecovered, OperationStarted, OperationCancelRequested, OperationCancelled, diff --git a/apps/desktop/src-tauri/src/deadline.rs b/apps/desktop/src-tauri/src/deadline.rs new file mode 100644 index 000000000..980db340d --- /dev/null +++ b/apps/desktop/src-tauri/src/deadline.rs @@ -0,0 +1,57 @@ +use std::time::Duration; +use tokio::time::Instant; + +/// One absolute operation deadline. Nested work may consume the remaining +/// budget but must never manufacture a fresh duration-based timeout. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Deadline { + at: Instant, +} + +impl Deadline { + pub(crate) fn after(duration: Duration) -> Self { + Self { + at: Instant::now() + duration, + } + } + + pub(crate) fn at(at: Instant) -> Self { + Self { at } + } + + pub(crate) fn instant(self) -> Instant { + self.at + } + + pub(crate) fn is_elapsed(self) -> bool { + Instant::now() >= self.at + } + + /// Cleanup may use one small, explicit post-deadline slack window. Before + /// the business deadline expires cleanup remains inside the same budget. + pub(crate) fn cleanup_deadline(self, slack: Duration) -> Instant { + let now = Instant::now(); + if now < self.at { + std::cmp::min(self.at, now + slack) + } else { + now + slack + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn nested_deadline_never_resets_outer_budget() { + let deadline = Deadline::after(Duration::from_millis(80)); + tokio::time::sleep(Duration::from_millis(30)).await; + let nested = Deadline::at(deadline.instant()); + assert!( + nested.instant().saturating_duration_since(Instant::now()) <= Duration::from_millis(60) + ); + tokio::time::sleep_until(nested.instant()).await; + assert!(deadline.is_elapsed()); + } +} diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index 2ef2bb752..703f7a4d0 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -1,5 +1,6 @@ mod activity; mod commands; +mod deadline; mod error; mod models; mod operation; @@ -17,7 +18,7 @@ pub fn run() { .setup(|app| { let data_dir = app.path().app_local_data_dir()?; let resource_dir = app.path().resource_dir()?; - app.manage(AppState::new(data_dir, resource_dir)); + app.manage(AppState::new(data_dir, resource_dir)?); Ok(()) }) .invoke_handler(tauri::generate_handler![ diff --git a/apps/desktop/src-tauri/src/platform/macos.rs b/apps/desktop/src-tauri/src/platform/macos.rs deleted file mode 100644 index cb7272c1c..000000000 --- a/apps/desktop/src-tauri/src/platform/macos.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::io; -use std::os::unix::process::CommandExt as _; -use tokio::process::Command; - -pub fn configure_child(command: &mut Command) { - // Create a new process group whose pgid is the spawned child's pid. If the - // process-group setup cannot be performed, spawning fails rather than - // leaving a Desktop-owned child without a safe tree identity. - command.as_std_mut().process_group(0); -} - -pub fn terminate_owned_tree(root_pid: u32) -> bool { - signal_owned_group(root_pid, libc::SIGTERM) -} - -pub fn force_stop_owned_tree(root_pid: u32) -> bool { - signal_owned_group(root_pid, libc::SIGKILL) -} - -pub fn owned_tree_is_running(root_pid: u32) -> bool { - let Some(pgid) = process_group_id(root_pid) else { - return false; - }; - let result = unsafe { libc::killpg(pgid, 0) }; - if result == 0 { - return true; - } - !matches!(io::Error::last_os_error().raw_os_error(), Some(libc::ESRCH)) -} - -fn signal_owned_group(root_pid: u32, signal: libc::c_int) -> bool { - let Some(pgid) = process_group_id(root_pid) else { - return false; - }; - let result = unsafe { libc::killpg(pgid, signal) }; - if result == 0 { - return true; - } - matches!(io::Error::last_os_error().raw_os_error(), Some(libc::ESRCH)) -} - -fn process_group_id(root_pid: u32) -> Option { - i32::try_from(root_pid).ok().filter(|value| *value > 0) -} diff --git a/apps/desktop/src-tauri/src/platform/mod.rs b/apps/desktop/src-tauri/src/platform/mod.rs index b3b968d14..156eb5fa7 100644 --- a/apps/desktop/src-tauri/src/platform/mod.rs +++ b/apps/desktop/src-tauri/src/platform/mod.rs @@ -1,34 +1,8 @@ -#[cfg(target_os = "macos")] -mod macos; #[cfg(target_os = "windows")] mod windows; -use tokio::process::Command; use webcodex_process::SpawnOptions; -#[derive(Debug, Clone, Copy)] -pub struct OwnedProcessTree { - root_pid: u32, -} - -impl OwnedProcessTree { - pub fn from_spawned_root(root_pid: u32) -> Option { - if root_pid == 0 { - return None; - } - #[cfg(target_os = "macos")] - i32::try_from(root_pid).ok()?; - Some(Self { root_pid }) - } -} - -pub fn configure_child(command: &mut Command) { - #[cfg(target_os = "windows")] - windows::configure_child(command); - #[cfg(target_os = "macos")] - macos::configure_child(command); -} - pub fn managed_spawn_options() -> SpawnOptions { #[cfg(target_os = "windows")] { @@ -40,43 +14,6 @@ pub fn managed_spawn_options() -> SpawnOptions { } } -pub async fn terminate_owned_tree(tree: OwnedProcessTree) -> bool { - #[cfg(target_os = "windows")] - { - return windows::force_stop_owned_tree(tree.root_pid).await; - } - #[cfg(target_os = "macos")] - { - return macos::terminate_owned_tree(tree.root_pid); - } - #[cfg(not(any(target_os = "windows", target_os = "macos")))] - { - let _ = tree; - false - } -} - -pub async fn force_stop_owned_tree(tree: OwnedProcessTree) -> bool { - #[cfg(target_os = "windows")] - { - return windows::force_stop_owned_tree(tree.root_pid).await; - } - #[cfg(target_os = "macos")] - { - return macos::force_stop_owned_tree(tree.root_pid); - } - #[cfg(not(any(target_os = "windows", target_os = "macos")))] - { - let _ = tree; - false - } -} - -#[cfg(target_os = "macos")] -pub fn owned_tree_is_running(tree: OwnedProcessTree) -> bool { - macos::owned_tree_is_running(tree.root_pid) -} - pub fn current_username() -> String { std::env::var("USERNAME") .or_else(|_| std::env::var("USER")) diff --git a/apps/desktop/src-tauri/src/platform/windows.rs b/apps/desktop/src-tauri/src/platform/windows.rs index 3d3f189b7..76d5e8776 100644 --- a/apps/desktop/src-tauri/src/platform/windows.rs +++ b/apps/desktop/src-tauri/src/platform/windows.rs @@ -1,34 +1,9 @@ -use std::os::windows::process::CommandExt; -use std::process::Stdio; -use std::time::Duration; -use tokio::process::Command; use webcodex_process::SpawnOptions; const CREATE_NO_WINDOW: u32 = 0x0800_0000; -pub fn configure_child(command: &mut Command) { - command.as_std_mut().creation_flags(CREATE_NO_WINDOW); -} - pub fn managed_spawn_options() -> SpawnOptions { SpawnOptions { windows_creation_flags: CREATE_NO_WINDOW, } } - -pub async fn force_stop_owned_tree(pid: u32) -> bool { - let mut command = Command::new("taskkill.exe"); - command - .arg("/PID") - .arg(pid.to_string()) - .arg("/T") - .arg("/F") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()); - configure_child(&mut command); - matches!( - tokio::time::timeout(Duration::from_secs(5), command.status()).await, - Ok(Ok(status)) if status.success() - ) -} diff --git a/apps/desktop/src-tauri/src/process/mod.rs b/apps/desktop/src-tauri/src/process/mod.rs index 8e0610f5a..c4dc73d47 100644 --- a/apps/desktop/src-tauri/src/process/mod.rs +++ b/apps/desktop/src-tauri/src/process/mod.rs @@ -1,8 +1,7 @@ -mod owned; mod supervisor; #[cfg(test)] mod tests; -pub(crate) use owned::reclaim_owned_tree; +pub(crate) use supervisor::MachineEventReceiver; pub use supervisor::{ProcessKind, ProcessPhase, ProcessSnapshot, ProcessSupervisor}; diff --git a/apps/desktop/src-tauri/src/process/owned.rs b/apps/desktop/src-tauri/src/process/owned.rs deleted file mode 100644 index 5ce78bd50..000000000 --- a/apps/desktop/src-tauri/src/process/owned.rs +++ /dev/null @@ -1,52 +0,0 @@ -use crate::platform; -use std::time::Duration; -use tokio::process::Child; -use tokio::time::Instant; - -pub(crate) async fn reclaim_owned_tree( - child: &mut Child, - tree: platform::OwnedProcessTree, - deadline: Instant, -) { - let _ = platform::terminate_owned_tree(tree).await; - if !wait_for_owned_tree_stop(child, tree, deadline).await { - let _ = platform::force_stop_owned_tree(tree).await; - let _ = child.start_kill(); - let _ = wait_for_owned_tree_stop(child, tree, deadline).await; - } -} - -#[cfg(target_os = "macos")] -async fn wait_for_owned_tree_stop( - child: &mut Child, - tree: platform::OwnedProcessTree, - deadline: Instant, -) -> bool { - loop { - let _ = child.try_wait(); - if !platform::owned_tree_is_running(tree) { - return true; - } - let now = Instant::now(); - if now >= deadline { - return false; - } - tokio::time::sleep(std::cmp::min(Duration::from_millis(25), deadline - now)).await; - } -} - -#[cfg(not(target_os = "macos"))] -async fn wait_for_owned_tree_stop( - child: &mut Child, - _tree: platform::OwnedProcessTree, - deadline: Instant, -) -> bool { - let now = Instant::now(); - if now >= deadline { - return child.try_wait().ok().flatten().is_some(); - } - matches!( - tokio::time::timeout_at(deadline, child.wait()).await, - Ok(Ok(_)) - ) -} diff --git a/apps/desktop/src-tauri/src/process/supervisor.rs b/apps/desktop/src-tauri/src/process/supervisor.rs index 90188cb76..17bc5418d 100644 --- a/apps/desktop/src-tauri/src/process/supervisor.rs +++ b/apps/desktop/src-tauri/src/process/supervisor.rs @@ -1,22 +1,25 @@ use crate::activity::{sanitize_message, ActivityEventKind, ActivityLevel, ActivityLog}; +use crate::deadline::Deadline; use crate::error::{DesktopError, DesktopResult}; use crate::platform; -use crate::process::reclaim_owned_tree; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::{HashMap, VecDeque}; -use std::process::Stdio; +use std::io::Read; +use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; -use tokio::io::{AsyncRead, AsyncReadExt}; -use tokio::process::{Child, Command}; -use tokio::sync::mpsc; +use tokio::sync::Notify; use tokio::task::JoinHandle; +use webcodex_process::{GracefulTermination, ManagedChild}; const LOG_LINES: usize = 80; const LOG_LINE_BYTES: usize = 2048; const MACHINE_LINE_BYTES: usize = 16 * 1024; +const MACHINE_EVENT_CAPACITY: usize = 64; +const MACHINE_CRITICAL_RESERVE: usize = 8; const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); -const STREAM_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); +const LOCAL_EOF_GRACE: std::time::Duration = std::time::Duration::from_millis(250); +const PROCESS_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(20); #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)] #[serde(rename_all = "snake_case")] @@ -58,8 +61,7 @@ pub struct ProcessSnapshot { } struct ManagedProcess { - child: Child, - owned_tree: platform::OwnedProcessTree, + child: ManagedChild, phase: ProcessPhase, exit_code: Option, logs: Arc>>, @@ -67,6 +69,141 @@ struct ManagedProcess { stderr_task: JoinHandle<()>, } +#[derive(Default)] +struct MachineEventState { + queue: VecDeque, + closed: bool, + dropped_progress: u64, + dropped_critical: u64, +} + +#[derive(Clone)] +struct MachineEventSender { + state: Arc>, + notify: Arc, +} + +pub(crate) struct MachineEventReceiver { + state: Arc>, + notify: Arc, +} + +fn machine_event_channel() -> (MachineEventSender, MachineEventReceiver) { + let state = Arc::new(Mutex::new(MachineEventState::default())); + let notify = Arc::new(Notify::new()); + ( + MachineEventSender { + state: Arc::clone(&state), + notify: Arc::clone(¬ify), + }, + MachineEventReceiver { state, notify }, + ) +} + +impl MachineEventReceiver { + pub(crate) async fn recv(&mut self) -> Option { + loop { + let notified = self.notify.notified(); + { + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.dropped_critical > 0 { + let dropped = std::mem::take(&mut state.dropped_critical); + return Some(serde_json::json!({ + "event": "machine_event_overflow", + "dropped_critical": dropped, + })); + } + if let Some(value) = state.queue.pop_front() { + return Some(value); + } + if state.closed { + return None; + } + } + notified.await; + } + } +} + +impl MachineEventSender { + fn send(&self, value: Value) { + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.closed { + return; + } + if machine_event_is_progress(&value) { + let progress_limit = MACHINE_EVENT_CAPACITY.saturating_sub(MACHINE_CRITICAL_RESERVE); + if state.queue.len() >= progress_limit { + if let Some(existing) = state + .queue + .iter_mut() + .rev() + .find(|event| machine_event_is_progress(event)) + { + *existing = value; + } else { + state.dropped_progress = state.dropped_progress.saturating_add(1); + } + return; + } + state.queue.push_back(value); + } else { + if state.queue.len() >= MACHINE_EVENT_CAPACITY { + if let Some(index) = state.queue.iter().position(machine_event_is_progress) { + state.queue.remove(index); + state.dropped_progress = state.dropped_progress.saturating_add(1); + } else if machine_event_is_terminal(&value) { + if let Some(index) = state + .queue + .iter() + .position(|event| !machine_event_is_terminal(event)) + { + state.queue.remove(index); + } else { + state.queue.pop_front(); + } + state.dropped_critical = state.dropped_critical.saturating_add(1); + } else { + state.dropped_critical = state.dropped_critical.saturating_add(1); + drop(state); + self.notify.notify_one(); + return; + } + } + state.queue.push_back(value); + } + drop(state); + self.notify.notify_one(); + } + + fn close(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + state.closed = true; + drop(state); + self.notify.notify_waiters(); + } +} + +fn machine_event_is_progress(value: &Value) -> bool { + value.get("event").and_then(Value::as_str) == Some("progress") +} + +fn machine_event_is_terminal(value: &Value) -> bool { + matches!( + value.get("event").and_then(Value::as_str), + Some("ready" | "error" | "failed" | "stopped" | "exit" | "exited" | "terminal") + ) +} + pub struct ProcessSupervisor { processes: HashMap, activity: ActivityLog, @@ -85,7 +222,7 @@ impl ProcessSupervisor { kind: ProcessKind, mut command: Command, machine_stdout: bool, - ) -> DesktopResult>> { + ) -> DesktopResult> { self.refresh(); if self.processes.get(&kind).is_some_and(|process| { matches!( @@ -99,57 +236,37 @@ impl ProcessSupervisor { "Stop the existing Desktop-owned process first.", )); } - #[cfg(target_os = "macos")] if self.processes.contains_key(&kind) { - // refresh() may have reaped a root that exited while one of its - // descendants is still alive. Preserve the recorded process-group - // authority until that terminal generation is fully reclaimed. + // A terminal direct child may still own live descendants. Keep the + // exact ManagedChild generation until its whole tree is reclaimed; + // never retarget cleanup by a remembered numeric PID/PGID. self.stop(kind).await; } - #[cfg(not(target_os = "macos"))] - self.processes.remove(&kind); - if matches!(kind, ProcessKind::QuickShare | ProcessKind::RegularTunnel) { - command.stdin(Stdio::piped()); - } else { - command.stdin(Stdio::null()); - } + // stdin is the Desktop parent-liveness lease for every long-lived + // generation. Quick Share/Tunnel already consume EOF; Local Server and + // Runner do so only when Desktop adds their explicit opt-in CLI flag. + command.stdin(Stdio::piped()); command.stdout(Stdio::piped()).stderr(Stdio::piped()); - platform::configure_child(&mut command); - let mut child = command.spawn().map_err(|error| { - DesktopError::new( - "process_start_failed", - format!("Could not start the {kind:?} process"), - "Check the configured WebCodex binaries and retry.", - ) - .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) - })?; - let pid = child.id().ok_or_else(|| { - let _ = child.start_kill(); - DesktopError::new( - "process_start_failed", - format!("Could not establish ownership for the {kind:?} process"), - "Retry the operation.", - ) - })?; - let owned_tree = platform::OwnedProcessTree::from_spawned_root(pid).ok_or_else(|| { - let _ = child.start_kill(); - DesktopError::new( - "process_start_failed", - format!( - "Could not establish a safe process-tree identity for the {kind:?} process" - ), - "Retry the operation.", - ) - })?; - let stdout = child.stdout.take().ok_or_else(|| { + let mut child = + ManagedChild::spawn_with_options(&mut command, platform::managed_spawn_options()) + .map_err(|error| { + DesktopError::new( + "process_start_failed", + format!("Could not start the {kind:?} process"), + "Check the configured WebCodex binaries and retry.", + ) + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) + })?; + let pid = child.id(); + let stdout = child.child_mut().stdout.take().ok_or_else(|| { DesktopError::new( "process_start_failed", "Could not capture process output", "Retry the operation.", ) })?; - let stderr = child.stderr.take().ok_or_else(|| { + let stderr = child.child_mut().stderr.take().ok_or_else(|| { DesktopError::new( "process_start_failed", "Could not capture process diagnostics", @@ -159,20 +276,18 @@ impl ProcessSupervisor { let logs = Arc::new(Mutex::new(VecDeque::new())); let (machine_tx, machine_rx) = if machine_stdout { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = machine_event_channel(); (Some(tx), Some(rx)) } else { (None, None) }; let stdout_logs = Arc::clone(&logs); - let stdout_task = tokio::spawn(drain_stream( - stdout, - stdout_logs, - machine_tx, - machine_stdout, - )); + let stdout_task = tokio::task::spawn_blocking(move || { + drain_stream(stdout, stdout_logs, machine_tx, machine_stdout) + }); let stderr_logs = Arc::clone(&logs); - let stderr_task = tokio::spawn(drain_stream(stderr, stderr_logs, None, false)); + let stderr_task = + tokio::task::spawn_blocking(move || drain_stream(stderr, stderr_logs, None, false)); self.activity.push( ActivityEventKind::ProcessStarted, kind.source(), @@ -183,7 +298,6 @@ impl ProcessSupervisor { kind, ManagedProcess { child, - owned_tree, phase: ProcessPhase::Starting, exit_code: None, logs, @@ -240,7 +354,7 @@ impl ProcessSupervisor { self.processes.get(&kind).map(|process| ProcessSnapshot { kind, phase: process.phase, - pid: process.child.id(), + pid: Some(process.child.id()), exit_code: process.exit_code, owned_by_desktop: true, }) @@ -262,9 +376,18 @@ impl ProcessSupervisor { } pub async fn stop(&mut self, kind: ProcessKind) { + self.stop_until(kind, Deadline::after(GRACEFUL_STOP_TIMEOUT)) + .await; + } + + pub async fn stop_until(&mut self, kind: ProcessKind, deadline: Deadline) { let Some(mut process) = self.processes.remove(&kind) else { return; }; + // Closing the Desktop side of stdin is the generation-scoped parent + // lease. Do it even if the direct child was already observed terminal: + // a descendant may still hold the child side of the pipe. + drop(process.child.child_mut().stdin.take()); if matches!( process.phase, ProcessPhase::Starting | ProcessPhase::Running @@ -276,45 +399,38 @@ impl ProcessSupervisor { ActivityLevel::Info, "Stopping the Desktop-owned process", ); - let mut graceful = + + let now = tokio::time::Instant::now(); + let eof_deadline = if matches!(kind, ProcessKind::QuickShare | ProcessKind::RegularTunnel) { - drop(process.child.stdin.take()); - matches!( - tokio::time::timeout(GRACEFUL_STOP_TIMEOUT, process.child.wait()).await, - Ok(Ok(_)) - ) + deadline.instant() } else { - false + std::cmp::min(deadline.instant(), now + LOCAL_EOF_GRACE) }; - #[cfg(target_os = "macos")] - if graceful && platform::owned_tree_is_running(process.owned_tree) { - // EOF can let the root exit before a descendant. Exact ownership - // is complete only when the process group itself is gone. - graceful = false; + let graceful = wait_for_tree_exit(&mut process.child, eof_deadline).await; + if !graceful && tokio::time::Instant::now() < deadline.instant() { + if matches!( + process.child.request_terminate_tree(), + Ok(GracefulTermination::Requested) + ) { + let signal_deadline = std::cmp::min( + deadline.instant(), + tokio::time::Instant::now() + LOCAL_EOF_GRACE, + ); + let _ = wait_for_tree_exit(&mut process.child, signal_deadline).await; + } } - if !graceful { - reclaim_owned_tree( - &mut process.child, - process.owned_tree, - tokio::time::Instant::now() + GRACEFUL_STOP_TIMEOUT, - ) - .await; + if !process.child.try_tree_exit().unwrap_or(false) { + let _ = process.child.terminate_tree(); + let _ = wait_for_tree_exit(&mut process.child, deadline.instant()).await; } } - #[cfg(target_os = "macos")] - if platform::owned_tree_is_running(process.owned_tree) { - // A root may already be terminal when stop() is called while one of - // its descendants is still alive. The recorded process-group identity - // remains the authority for that cleanup. - reclaim_owned_tree( - &mut process.child, - process.owned_tree, - tokio::time::Instant::now() + GRACEFUL_STOP_TIMEOUT, - ) - .await; + if !process.child.try_tree_exit().unwrap_or(false) { + let _ = process.child.terminate_tree(); + let _ = wait_for_tree_exit(&mut process.child, deadline.instant()).await; } - finish_drain_task(process.stdout_task).await; - finish_drain_task(process.stderr_task).await; + finish_drain_task(process.stdout_task, deadline.instant()).await; + finish_drain_task(process.stderr_task, deadline.instant()).await; self.activity.push( ActivityEventKind::ProcessStopped, kind.source(), @@ -335,23 +451,36 @@ impl ProcessSupervisor { } } -async fn finish_drain_task(mut task: JoinHandle<()>) { - if tokio::time::timeout(STREAM_DRAIN_TIMEOUT, &mut task) - .await - .is_err() +async fn wait_for_tree_exit(child: &mut ManagedChild, deadline: tokio::time::Instant) -> bool { + loop { + let _ = child.try_wait(); + if child.try_tree_exit().unwrap_or(false) { + return true; + } + let now = tokio::time::Instant::now(); + if now >= deadline { + return false; + } + tokio::time::sleep_until(std::cmp::min(deadline, now + PROCESS_POLL_INTERVAL)).await; + } +} + +async fn finish_drain_task(mut task: JoinHandle<()>, deadline: tokio::time::Instant) { + if tokio::time::Instant::now() >= deadline + || tokio::time::timeout_at(deadline, &mut task).await.is_err() { task.abort(); let _ = task.await; } } -async fn drain_stream( +fn drain_stream( mut reader: R, logs: Arc>>, - machine_tx: Option>, + machine_tx: Option, machine_only: bool, ) where - R: AsyncRead + Unpin, + R: Read, { let mut buffer = [0_u8; 4096]; let mut line = Vec::with_capacity(4096); @@ -361,7 +490,7 @@ async fn drain_stream( LOG_LINE_BYTES }; loop { - let read = match reader.read(&mut buffer).await { + let read = match reader.read(&mut buffer) { Ok(0) | Err(_) => break, Ok(read) => read, }; @@ -377,12 +506,15 @@ async fn drain_stream( if !line.is_empty() { process_line(&line, &logs, machine_tx.as_ref(), machine_only); } + if let Some(tx) = machine_tx { + tx.close(); + } } fn process_line( line: &[u8], logs: &Arc>>, - machine_tx: Option<&mpsc::UnboundedSender>, + machine_tx: Option<&MachineEventSender>, machine_only: bool, ) { let text = String::from_utf8_lossy(line).trim().to_string(); @@ -391,7 +523,7 @@ fn process_line( } if machine_only { if let (Some(tx), Ok(value)) = (machine_tx, serde_json::from_str::(&text)) { - let _ = tx.send(value); + tx.send(value); } return; } @@ -406,6 +538,61 @@ fn process_line( mod tests { use super::*; + #[tokio::test] + async fn machine_event_progress_flood_stays_bounded_and_ready_is_observed() { + let (sender, mut receiver) = machine_event_channel(); + for sequence in 0..10_000_u64 { + sender.send(serde_json::json!({ + "event": "progress", + "sequence": sequence, + })); + } + sender.send(serde_json::json!({ "event": "ready", "schema_version": 1 })); + sender.close(); + + let queued = sender + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .queue + .len(); + assert!(queued <= MACHINE_EVENT_CAPACITY); + + let mut ready = false; + while let Some(value) = receiver.recv().await { + if value.get("event").and_then(Value::as_str) == Some("ready") { + ready = true; + break; + } + } + assert!(ready, "readiness event must survive a noisy progress flood"); + } + + #[tokio::test] + async fn critical_overflow_is_explicit_and_terminal_event_is_retained() { + let (sender, mut receiver) = machine_event_channel(); + for sequence in 0..MACHINE_EVENT_CAPACITY { + sender.send(serde_json::json!({ + "event": "diagnostic", + "sequence": sequence, + })); + } + sender.send(serde_json::json!({ "event": "terminal", "status": "failed" })); + sender.close(); + + let mut saw_overflow = false; + let mut saw_terminal = false; + while let Some(value) = receiver.recv().await { + match value.get("event").and_then(Value::as_str) { + Some("machine_event_overflow") => saw_overflow = true, + Some("terminal") => saw_terminal = true, + _ => {} + } + } + assert!(saw_overflow, "critical loss must never be silent"); + assert!(saw_terminal, "terminal event must remain observable"); + } + #[tokio::test] async fn supervisor_only_stops_children_it_owns() { let activity = ActivityLog::default(); @@ -425,6 +612,73 @@ mod tests { assert_eq!(kinds.len(), 4); } + #[cfg(unix)] + #[tokio::test] + async fn local_parent_liveness_lease_closes_on_desktop_stop() { + let marker = std::env::temp_dir().join(format!( + "webcodex-desktop-local-parent-eof-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + let marker_arg = marker.to_string_lossy().into_owned(); + let mut command = Command::new("/bin/sh"); + command.args([ + "-c", + "cat >/dev/null; printf eof > \"$1\"", + "webcodex-parent-eof", + marker_arg.as_str(), + ]); + + let activity = ActivityLog::default(); + let mut supervisor = ProcessSupervisor::new(activity); + supervisor + .spawn_owned(ProcessKind::LocalServer, command, false) + .await + .expect("start local parent-liveness fixture"); + supervisor.stop(ProcessKind::LocalServer).await; + + assert!( + marker.is_file(), + "local generation must observe stdin EOF before forced tree cleanup" + ); + let _ = std::fs::remove_file(marker); + } + + #[cfg(unix)] + #[tokio::test] + async fn quick_share_eof_stop_remains_green() { + let marker = std::env::temp_dir().join(format!( + "webcodex-desktop-quick-share-eof-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + let marker_arg = marker.to_string_lossy().into_owned(); + let mut command = Command::new("/bin/sh"); + command.args([ + "-c", + "cat >/dev/null; printf eof > \"$1\"", + "webcodex-quick-share-eof", + marker_arg.as_str(), + ]); + + let activity = ActivityLog::default(); + let mut supervisor = ProcessSupervisor::new(activity); + supervisor + .spawn_owned(ProcessKind::QuickShare, command, false) + .await + .expect("start Quick Share EOF fixture"); + supervisor.stop(ProcessKind::QuickShare).await; + + assert!(marker.is_file(), "Quick Share child must observe stdin EOF"); + let _ = std::fs::remove_file(marker); + } + #[cfg(windows)] #[tokio::test] async fn regular_tunnel_stop_closes_stdin_for_canonical_graceful_shutdown() { diff --git a/apps/desktop/src-tauri/src/state.rs b/apps/desktop/src-tauri/src/state.rs index 2b0e1ff92..cf9c86da0 100644 --- a/apps/desktop/src-tauri/src/state.rs +++ b/apps/desktop/src-tauri/src/state.rs @@ -1,4 +1,5 @@ use crate::activity::{ActivityEventKind, ActivityLevel, ActivityLog}; +use crate::deadline::Deadline; use crate::error::{DesktopError, DesktopResult}; use crate::models::{ aggregate_readiness, DesktopOperationKind, DesktopStateSnapshot, Enrollment, Experience, @@ -11,14 +12,19 @@ use crate::operation::{ cancelled_error, CancellationContext, CancellationSignal, OperationAdmission, OperationController, }; -use crate::process::{ProcessKind, ProcessPhase, ProcessSupervisor}; +use crate::process::{MachineEventReceiver, ProcessKind, ProcessPhase, ProcessSupervisor}; use crate::webcodex::{ inspect_project_path, ProjectRuntimeIdentity, QuickShareReadyEvent, RegularTunnelReadyEvent, WebCodexAdapter, }; use serde_json::Value; +#[cfg(unix)] +use std::fs::File; +use std::fs::OpenOptions; +use std::io::{self, Write}; use std::net::TcpListener; use std::path::{Path, PathBuf}; +use std::sync::atomic::AtomicU64; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, RwLock}; use std::time::Duration; @@ -30,7 +36,10 @@ const PROJECT_READY_TIMEOUT: Duration = Duration::from_secs(20); const QUICK_SHARE_READY_TIMEOUT: Duration = Duration::from_secs(90); const REGULAR_TUNNEL_READY_TIMEOUT: Duration = Duration::from_secs(90); const POLL_INTERVAL: Duration = Duration::from_millis(300); +const READINESS_CLEANUP_SLACK: Duration = Duration::from_secs(2); const SHUTDOWN_OPERATION_WAIT: Duration = Duration::from_secs(5); +const DESKTOP_STATE_MAX_BYTES: u64 = 256 * 1024; +static NEXT_STATE_TEMP_ID: AtomicU64 = AtomicU64::new(1); type SharedSupervisor = Arc>; @@ -45,12 +54,12 @@ pub struct AppState { } impl AppState { - pub fn new(data_dir: PathBuf, resource_dir: PathBuf) -> Self { - let core = DesktopCore::new(data_dir, resource_dir); + pub fn new(data_dir: PathBuf, resource_dir: PathBuf) -> DesktopResult { + let core = DesktopCore::new(data_dir, resource_dir)?; let published = Arc::clone(&core.published); let supervisor = Arc::clone(&core.supervisor); let activity = core.activity.clone(); - Self { + Ok(Self { core: Mutex::new(Some(core)), published, supervisor, @@ -58,7 +67,7 @@ impl AppState { activity, shutdown_signal: CancellationSignal::new(), shutdown_started: AtomicBool::new(false), - } + }) } pub fn get_state(&self) -> DesktopStateSnapshot { @@ -403,10 +412,10 @@ pub struct DesktopCore { } impl DesktopCore { - fn new(data_dir: PathBuf, resource_dir: PathBuf) -> Self { + fn new(data_dir: PathBuf, resource_dir: PathBuf) -> DesktopResult { let activity = ActivityLog::default(); let config_path = data_dir.join("desktop-state.json"); - let config = load_config(&config_path).unwrap_or_default(); + let config = load_config(&config_path, &activity)?; let mut snapshot = DesktopStateSnapshot::default(); snapshot.topology = config.topology.clone(); snapshot.project = project_snapshot(&config); @@ -414,7 +423,7 @@ impl DesktopCore { snapshot.regular_tunnel_available = true; let published = Arc::new(RwLock::new(snapshot.clone())); let supervisor = Arc::new(Mutex::new(ProcessSupervisor::new(activity.clone()))); - Self { + Ok(Self { data_dir, config_path, config, @@ -423,7 +432,7 @@ impl DesktopCore { supervisor, activity, published, - } + }) } pub async fn get_state(&mut self) -> DesktopResult { @@ -746,19 +755,43 @@ impl DesktopCore { self.save_config().await?; cancellation.check()?; + let server_deadline = Deadline::after(SERVER_READY_TIMEOUT); let running = self .adapter - .server_status(Some(&server_url), Some(&env_file), None, cancellation) + .server_status_until( + Some(&server_url), + Some(&env_file), + None, + cancellation, + server_deadline, + ) .await .is_ok_and(|status| status.http_reachable); cancellation.check()?; - if !running { + let server_started = if !running { + if server_deadline.is_elapsed() { + return Err(readiness_timeout_error( + "server_unreachable", + "WebCodex Service did not become ready", + "Check the local Service diagnostics and retry.", + )); + } let command = self.adapter.local_server_command(&env_file)?; self.spawn_owned(ProcessKind::LocalServer, command, false, cancellation) .await?; - } - self.wait_for_server(&server_url, Some(&env_file), None, cancellation) - .await?; + true + } else { + false + }; + self.wait_for_server( + &server_url, + Some(&env_file), + None, + cancellation, + server_deadline, + server_started, + ) + .await?; self.snapshot.readiness.server = ServerReadiness::Ready; self.publish_snapshot(); @@ -787,23 +820,36 @@ impl DesktopCore { } }; + let runner_deadline = Deadline::after(RUNNER_READY_TIMEOUT); let runner_ready = self .adapter - .runner_ready(&identity, cancellation) + .runner_ready_until(&identity, cancellation, runner_deadline) .await .unwrap_or(false); cancellation.check()?; - if !runner_ready { + let runner_started = if !runner_ready { + if runner_deadline.is_elapsed() { + return Err(readiness_timeout_error( + "runner_offline", + "Runner did not become connected", + "Check Server reachability and Runner diagnostics, then retry.", + )); + } self.snapshot.readiness.runner = RunnerReadiness::Connecting; self.publish_snapshot(); let command = self.adapter.local_runner_command(&identity.runner_config)?; self.spawn_owned(ProcessKind::LocalRunner, command, false, cancellation) .await?; - } - self.wait_for_runner(&identity, cancellation).await?; + true + } else { + false + }; + self.wait_for_runner(&identity, cancellation, runner_deadline, runner_started) + .await?; self.snapshot.readiness.runner = RunnerReadiness::Ready; self.publish_snapshot(); - self.wait_for_project(&identity, cancellation).await?; + self.wait_for_project(&identity, cancellation, runner_started) + .await?; cancellation.check()?; self.snapshot.readiness = aggregate_readiness( ServerReadiness::Ready, @@ -900,15 +946,31 @@ impl DesktopCore { } }; - let server_status = self + let server_deadline = Deadline::after(SERVER_READY_TIMEOUT); + let server_status = match self .adapter - .server_status( + .server_status_until( Some(&server_url), None, Some(&identity.user_token_file), cancellation, + server_deadline, ) - .await?; + .await + { + Ok(status) => status, + Err(error) => { + cancellation.check()?; + if server_deadline.is_elapsed() { + return Err(readiness_timeout_error( + "server_unreachable", + "The existing WebCodex Server did not respond before the readiness deadline", + "Check the Server URL and network path, then retry.", + )); + } + return Err(error); + } + }; cancellation.check()?; if !server_status.http_reachable { return Err(DesktopError::new( @@ -917,19 +979,32 @@ impl DesktopCore { "Check the Server URL and network path, then retry.", )); } + let runner_deadline = Deadline::after(RUNNER_READY_TIMEOUT); let runner_ready = self .adapter - .runner_ready(&identity, cancellation) + .runner_ready_until(&identity, cancellation, runner_deadline) .await .unwrap_or(false); cancellation.check()?; - if !runner_ready { + let runner_started = if !runner_ready { + if runner_deadline.is_elapsed() { + return Err(readiness_timeout_error( + "runner_offline", + "Runner did not become connected", + "Check Server reachability and Runner diagnostics, then retry.", + )); + } let command = self.adapter.local_runner_command(&identity.runner_config)?; self.spawn_owned(ProcessKind::LocalRunner, command, false, cancellation) .await?; - } - self.wait_for_runner(&identity, cancellation).await?; - self.wait_for_project(&identity, cancellation).await?; + true + } else { + false + }; + self.wait_for_runner(&identity, cancellation, runner_deadline, runner_started) + .await?; + self.wait_for_project(&identity, cancellation, runner_started) + .await?; cancellation.check()?; self.config.topology = Some(topology); self.save_config().await?; @@ -982,9 +1057,17 @@ impl DesktopCore { "Stop the current share before starting another one.", )); } + let deadline = Deadline::after(QUICK_SHARE_READY_TIMEOUT); let command = self .adapter .quick_share_command(Path::new(&project.path), provider)?; + if deadline.is_elapsed() { + return Err(readiness_timeout_error( + "quick_share_not_ready", + "Quick Share did not reach verified readiness", + "Check Activity and Tunnel prerequisites, then retry.", + )); + } let mut events = self .spawn_owned(ProcessKind::QuickShare, command, true, cancellation) .await? @@ -1016,7 +1099,6 @@ impl DesktopCore { "Starting the temporary Quick Share runtime", ); self.publish_snapshot(); - let event_wait = async { while let Some(value) = events.recv().await { if value.get("event").and_then(Value::as_str) == Some("ready") { @@ -1028,22 +1110,32 @@ impl DesktopCore { let event_value = tokio::select! { biased; _ = cancellation.cancelled() => { - self.stop_process(ProcessKind::QuickShare).await; + self.stop_process_until( + ProcessKind::QuickShare, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ).await; return Err(cancelled_error()); } - result = tokio::time::timeout(QUICK_SHARE_READY_TIMEOUT, event_wait) => { + result = tokio::time::timeout_at(deadline.instant(), event_wait) => { result.ok().flatten() } }; let Some(event_value) = event_value else { let logs = self.process_logs(ProcessKind::QuickShare).await; - self.stop_process(ProcessKind::QuickShare).await; + self.stop_process_until( + ProcessKind::QuickShare, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; return Err(DesktopError::new( "quick_share_not_ready", "Quick Share did not reach verified readiness", "Check Activity and Tunnel prerequisites, then retry.", ) - .with_details(serde_json::json!({ "diagnostic_lines": logs }))); + .with_details(serde_json::json!({ + "category": "readiness_timeout", + "diagnostic_lines": logs, + }))); }; let event: QuickShareReadyEvent = match serde_json::from_value(event_value) { Ok(event) => event, @@ -1218,9 +1310,17 @@ impl DesktopCore { ) })?; + let deadline = Deadline::after(REGULAR_TUNNEL_READY_TIMEOUT); let command = self .adapter .regular_tunnel_command(&env_file, &user_token_file)?; + if deadline.is_elapsed() { + return Err(readiness_timeout_error( + "tunnel_unavailable", + "OpenAI Secure Tunnel did not reach verified readiness", + "Check Activity and the canonical Tunnel prerequisites, then retry.", + )); + } let mut events = self .spawn_owned(ProcessKind::RegularTunnel, command, true, cancellation) .await? @@ -1246,7 +1346,6 @@ impl DesktopCore { "Starting the regular OpenAI Secure Tunnel", ); self.publish_snapshot(); - let event_wait = async { while let Some(value) = events.recv().await { if value.get("event").and_then(Value::as_str) == Some("ready") { @@ -1258,16 +1357,23 @@ impl DesktopCore { let event_value = tokio::select! { biased; _ = cancellation.cancelled() => { - self.stop_process(ProcessKind::RegularTunnel).await; + self.stop_process_until( + ProcessKind::RegularTunnel, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ).await; return Err(cancelled_error()); } - result = tokio::time::timeout(REGULAR_TUNNEL_READY_TIMEOUT, event_wait) => { + result = tokio::time::timeout_at(deadline.instant(), event_wait) => { result.ok().flatten() } }; let Some(event_value) = event_value else { let logs = self.process_logs(ProcessKind::RegularTunnel).await; - self.stop_process(ProcessKind::RegularTunnel).await; + self.stop_process_until( + ProcessKind::RegularTunnel, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; self.snapshot.regular_tunnel = Some(RegularTunnelState { provider: "openai".to_string(), status: RegularTunnelStatus::Error, @@ -1287,7 +1393,10 @@ impl DesktopCore { "OpenAI Secure Tunnel did not reach verified readiness", "Check Activity and the canonical Tunnel prerequisites, then retry.", ) - .with_details(serde_json::json!({ "diagnostic_lines": logs }))); + .with_details(serde_json::json!({ + "category": "readiness_timeout", + "diagnostic_lines": logs, + }))); }; let event: RegularTunnelReadyEvent = match serde_json::from_value(event_value) { Ok(event) => event, @@ -1393,10 +1502,10 @@ impl DesktopCore { async fn spawn_owned( &self, kind: ProcessKind, - command: tokio::process::Command, + command: std::process::Command, machine_stdout: bool, cancellation: &CancellationContext, - ) -> DesktopResult>> { + ) -> DesktopResult> { cancellation.check()?; let mut supervisor = self.supervisor.lock().await; cancellation.check()?; @@ -1407,18 +1516,46 @@ impl DesktopCore { self.supervisor.lock().await.stop(kind).await; } + async fn stop_process_until(&self, kind: ProcessKind, deadline: Deadline) { + self.supervisor + .lock() + .await + .stop_until(kind, deadline) + .await; + } + async fn wait_for_server( &mut self, server_url: &str, env_file: Option<&Path>, token_file: Option<&Path>, cancellation: &CancellationContext, + deadline: Deadline, + cleanup_owned_process: bool, ) -> DesktopResult<()> { - let deadline = tokio::time::Instant::now() + SERVER_READY_TIMEOUT; loop { cancellation.check()?; + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalServer, + deadline, + cleanup_owned_process, + ) + .await; + return Err(readiness_timeout_error( + "server_unreachable", + "WebCodex Service did not become ready", + "Check the local Service diagnostics and retry.", + )); + } if let Some(process) = self.process_snapshot(ProcessKind::LocalServer).await { if matches!(process.phase, ProcessPhase::Exited | ProcessPhase::Failed) { + self.cleanup_readiness_process( + ProcessKind::LocalServer, + deadline, + cleanup_owned_process, + ) + .await; return Err(DesktopError::new( "server_start_failed", "The Desktop-owned WebCodex Server exited during startup", @@ -1428,21 +1565,33 @@ impl DesktopCore { } if self .adapter - .server_status(Some(server_url), env_file, token_file, cancellation) + .server_status_until( + Some(server_url), + env_file, + token_file, + cancellation, + deadline, + ) .await .is_ok_and(|status| status.http_reachable) { return Ok(()); } cancellation.check()?; - if tokio::time::Instant::now() >= deadline { - return Err(DesktopError::new( + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalServer, + deadline, + cleanup_owned_process, + ) + .await; + return Err(readiness_timeout_error( "server_unreachable", "WebCodex Service did not become ready", "Check the local Service diagnostics and retry.", )); } - sleep_or_cancel(POLL_INTERVAL, cancellation).await?; + sleep_or_cancel_until(POLL_INTERVAL, cancellation, deadline).await?; } } @@ -1450,12 +1599,32 @@ impl DesktopCore { &mut self, identity: &ProjectRuntimeIdentity, cancellation: &CancellationContext, + deadline: Deadline, + cleanup_owned_process: bool, ) -> DesktopResult<()> { - let deadline = tokio::time::Instant::now() + RUNNER_READY_TIMEOUT; loop { cancellation.check()?; + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalRunner, + deadline, + cleanup_owned_process, + ) + .await; + return Err(readiness_timeout_error( + "runner_offline", + "Runner did not become connected", + "Check Server reachability and Runner diagnostics, then retry.", + )); + } if let Some(process) = self.process_snapshot(ProcessKind::LocalRunner).await { if matches!(process.phase, ProcessPhase::Exited | ProcessPhase::Failed) { + self.cleanup_readiness_process( + ProcessKind::LocalRunner, + deadline, + cleanup_owned_process, + ) + .await; return Err(DesktopError::new( "runner_offline", "The Desktop-owned Runner exited while connecting", @@ -1465,21 +1634,27 @@ impl DesktopCore { } if self .adapter - .runner_ready(identity, cancellation) + .runner_ready_until(identity, cancellation, deadline) .await .unwrap_or(false) { return Ok(()); } cancellation.check()?; - if tokio::time::Instant::now() >= deadline { - return Err(DesktopError::new( + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalRunner, + deadline, + cleanup_owned_process, + ) + .await; + return Err(readiness_timeout_error( "runner_offline", "Runner did not become connected", "Check Server reachability and Runner diagnostics, then retry.", )); } - sleep_or_cancel(POLL_INTERVAL, cancellation).await?; + sleep_or_cancel_until(POLL_INTERVAL, cancellation, deadline).await?; } } @@ -1487,27 +1662,62 @@ impl DesktopCore { &mut self, identity: &ProjectRuntimeIdentity, cancellation: &CancellationContext, + cleanup_owned_runner: bool, ) -> DesktopResult<()> { - let deadline = tokio::time::Instant::now() + PROJECT_READY_TIMEOUT; + let deadline = Deadline::after(PROJECT_READY_TIMEOUT); loop { cancellation.check()?; + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalRunner, + deadline, + cleanup_owned_runner, + ) + .await; + return Err(readiness_timeout_error( + "project_not_loaded", + "The selected project is registered but not loaded by the Runner", + "Restart the Runner or check the project registry, then retry.", + )); + } if self .adapter - .project_ready(identity, cancellation) + .project_ready_until(identity, cancellation, deadline) .await .unwrap_or(false) { return Ok(()); } cancellation.check()?; - if tokio::time::Instant::now() >= deadline { - return Err(DesktopError::new( + if deadline.is_elapsed() { + self.cleanup_readiness_process( + ProcessKind::LocalRunner, + deadline, + cleanup_owned_runner, + ) + .await; + return Err(readiness_timeout_error( "project_not_loaded", "The selected project is registered but not loaded by the Runner", "Restart the Runner or check the project registry, then retry.", )); } - sleep_or_cancel(POLL_INTERVAL, cancellation).await?; + sleep_or_cancel_until(POLL_INTERVAL, cancellation, deadline).await?; + } + } + + async fn cleanup_readiness_process( + &self, + kind: ProcessKind, + deadline: Deadline, + cleanup_owned_process: bool, + ) { + if cleanup_owned_process { + self.stop_process_until( + kind, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; } } @@ -1549,29 +1759,40 @@ impl DesktopCore { "Retry the setup operation.", ) })?; - tokio::fs::write(&self.config_path, encoded) + let config_path = self.config_path.clone(); + tokio::task::spawn_blocking(move || save_config_atomically(&config_path, &encoded)) .await - .map_err(|_| { - DesktopError::new( - "desktop_state_unavailable", - "Desktop could not persist its non-secret runtime state", - "Check local app-data permissions and retry.", - ) - }) + .map_err(|_| desktop_state_unavailable("Desktop state persistence worker stopped"))??; + Ok(()) } } -async fn sleep_or_cancel( +async fn sleep_or_cancel_until( duration: Duration, cancellation: &CancellationContext, + deadline: Deadline, ) -> DesktopResult<()> { + cancellation.check()?; + if deadline.is_elapsed() { + return Ok(()); + } + let wake_at = std::cmp::min(deadline.instant(), tokio::time::Instant::now() + duration); tokio::select! { biased; _ = cancellation.cancelled() => Err(cancelled_error()), - _ = tokio::time::sleep(duration) => Ok(()), + _ = tokio::time::sleep_until(wake_at) => Ok(()), } } +fn readiness_timeout_error( + code: &'static str, + message: &'static str, + action: &'static str, +) -> DesktopError { + DesktopError::new(code, message, action) + .with_details(serde_json::json!({ "category": "readiness_timeout" })) +} + fn project_snapshot(config: &StoredDesktopConfig) -> Option { let mut project = config.project.clone()?; if identity_from_config(config).is_none() { @@ -1604,13 +1825,238 @@ fn identity_from_config(config: &StoredDesktopConfig) -> Option Option { - let metadata = std::fs::metadata(path).ok()?; - if metadata.len() > 256 * 1024 { - return None; +#[derive(Debug)] +enum StoredConfigFile { + Missing, + Valid { + config: StoredDesktopConfig, + bytes: Vec, + }, + Corrupt, +} + +fn load_config(path: &Path, activity: &ActivityLog) -> DesktopResult { + let backup_path = desktop_state_backup_path(path); + match read_stored_config(path)? { + StoredConfigFile::Valid { config, .. } => Ok(config), + StoredConfigFile::Missing => match read_stored_config(&backup_path)? { + StoredConfigFile::Missing => Ok(StoredDesktopConfig::default()), + StoredConfigFile::Valid { config, bytes } => { + recover_config_from_backup(path, &bytes, activity)?; + Ok(config) + } + StoredConfigFile::Corrupt => Err(desktop_state_corrupt()), + }, + StoredConfigFile::Corrupt => match read_stored_config(&backup_path)? { + StoredConfigFile::Valid { config, bytes } => { + recover_config_from_backup(path, &bytes, activity)?; + Ok(config) + } + StoredConfigFile::Missing | StoredConfigFile::Corrupt => Err(desktop_state_corrupt()), + }, + } +} + +fn recover_config_from_backup( + primary_path: &Path, + bytes: &[u8], + activity: &ActivityLog, +) -> DesktopResult<()> { + write_atomic_file(primary_path, bytes).map_err(|error| { + desktop_state_unavailable("Desktop could not restore the previous known-good state") + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) + })?; + activity.push( + ActivityEventKind::StateRecovered, + "desktop_state", + ActivityLevel::Warning, + "Recovered Desktop state from the previous known-good snapshot", + ); + Ok(()) +} + +fn read_stored_config(path: &Path) -> DesktopResult { + let metadata = match std::fs::metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Ok(StoredConfigFile::Missing) + } + Err(error) => { + return Err( + desktop_state_unavailable("Desktop could not inspect its saved state") + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })), + ) + } + }; + if !metadata.is_file() || metadata.len() > DESKTOP_STATE_MAX_BYTES { + return Ok(StoredConfigFile::Corrupt); + } + let bytes = std::fs::read(path).map_err(|error| { + desktop_state_unavailable("Desktop could not read its saved state") + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) + })?; + match serde_json::from_slice::(&bytes) { + Ok(config) => Ok(StoredConfigFile::Valid { config, bytes }), + Err(_) => Ok(StoredConfigFile::Corrupt), + } +} + +fn save_config_atomically(path: &Path, encoded: &[u8]) -> DesktopResult<()> { + if encoded.len() as u64 > DESKTOP_STATE_MAX_BYTES { + return Err(DesktopError::new( + "desktop_state_invalid", + "Desktop state exceeded its bounded persistence size", + "Retry after reducing the saved Desktop configuration.", + )); } - let bytes = std::fs::read(path).ok()?; - serde_json::from_slice(&bytes).ok() + + if let StoredConfigFile::Valid { bytes, .. } = read_stored_config(path)? { + let backup = desktop_state_backup_path(path); + write_atomic_file(&backup, &bytes).map_err(|error| { + desktop_state_unavailable("Desktop could not preserve the previous known-good state") + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) + })?; + } + + write_atomic_file(path, encoded).map_err(|error| { + desktop_state_unavailable("Desktop could not persist its non-secret runtime state") + .with_details(serde_json::json!({ "io_kind": format!("{:?}", error.kind()) })) + }) +} + +fn desktop_state_backup_path(path: &Path) -> PathBuf { + let file_name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("desktop-state.json"); + path.with_file_name(format!("{file_name}.bak")) +} + +fn state_temp_path(path: &Path) -> PathBuf { + let id = NEXT_STATE_TEMP_ID.fetch_add(1, Ordering::Relaxed); + let file_name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("desktop-state.json"); + path.with_file_name(format!(".{file_name}.{}.{}.tmp", std::process::id(), id)) +} + +fn write_atomic_file(path: &Path, bytes: &[u8]) -> io::Result<()> { + write_atomic_file_with_hook(path, bytes, |_| Ok(())) +} + +fn write_atomic_file_with_hook(path: &Path, bytes: &[u8], before_replace: F) -> io::Result<()> +where + F: FnOnce(&Path) -> io::Result<()>, +{ + let parent = path + .parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "state path has no parent"))?; + std::fs::create_dir_all(parent)?; + let temp_path = state_temp_path(path); + let result = (|| { + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options.open(&temp_path)?; + file.write_all(bytes)?; + file.flush()?; + file.sync_all()?; + drop(file); + before_replace(&temp_path)?; + atomic_replace(&temp_path, path)?; + sync_state_directory(parent)?; + Ok(()) + })(); + if result.is_err() { + let _ = std::fs::remove_file(&temp_path); + } + result +} + +#[cfg(not(windows))] +fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> { + std::fs::rename(source, destination) +} + +#[cfg(windows)] +fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, + }; + + let source = source + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let destination = destination + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let result = unsafe { + MoveFileExW( + source.as_ptr(), + destination.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if result == 0 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(unix)] +fn sync_state_directory(path: &Path) -> io::Result<()> { + let directory = File::open(path)?; + match directory.sync_all() { + Ok(()) => Ok(()), + Err(error) + if error.kind() == io::ErrorKind::Unsupported + || error.raw_os_error() == Some(libc::EINVAL) => + { + // Some Unix filesystems (notably macOS variants) do not support + // directory fsync. The file itself has already been synced and the + // same-directory rename is atomic, so treat this specific platform + // limitation as best-effort durability rather than a false save + // failure after replacement has already succeeded. + Ok(()) + } + Err(error) => Err(error), + } +} + +#[cfg(not(unix))] +fn sync_state_directory(_path: &Path) -> io::Result<()> { + // Windows uses MOVEFILE_WRITE_THROUGH for the replacement. Opening a + // directory for FlushFileBuffers would require broader sharing semantics + // than the app-data policy needs here. + Ok(()) +} + +fn desktop_state_corrupt() -> DesktopError { + DesktopError::new( + "desktop_state_corrupt", + "Desktop saved state is corrupt and no valid recovery snapshot is available", + "Restore or remove the Desktop state files explicitly, then restart WebCodex Desktop.", + ) + .with_details(serde_json::json!({ "category": "state_corrupt" })) +} + +fn desktop_state_unavailable(message: &'static str) -> DesktopError { + DesktopError::new( + "desktop_state_unavailable", + message, + "Check local app-data permissions and retry.", + ) } fn reserve_loopback_address() -> DesktopResult { @@ -1721,6 +2167,124 @@ fn same_project(left: &str, right: &str) -> bool { mod tests { use super::*; + fn unique_state_dir(name: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "webcodex-desktop-state-{name}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )) + } + + fn test_stored_config(label: &str) -> StoredDesktopConfig { + StoredDesktopConfig { + topology: None, + project: Some(ProjectSelection { + path: format!("/{label}"), + allowed_root: "/".to_string(), + is_git_repository: false, + runtime_project_id: None, + }), + runtime: None, + } + } + + #[test] + fn atomic_save_interruption_keeps_prior_valid_state() { + let dir = unique_state_dir("interrupted-save"); + std::fs::create_dir_all(&dir).expect("create state fixture dir"); + let path = dir.join("desktop-state.json"); + let previous = test_stored_config("previous"); + let replacement = test_stored_config("replacement"); + let previous_bytes = serde_json::to_vec_pretty(&previous).unwrap(); + let replacement_bytes = serde_json::to_vec_pretty(&replacement).unwrap(); + write_atomic_file(&path, &previous_bytes).expect("write previous state"); + + let interrupted = write_atomic_file_with_hook(&path, &replacement_bytes, |_| { + Err(io::Error::other("injected interruption before replace")) + }); + assert!(interrupted.is_err()); + match read_stored_config(&path).expect("read state after interruption") { + StoredConfigFile::Valid { config, .. } => assert_eq!(config, previous), + other => panic!("previous state was not preserved: {other:?}"), + } + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn atomic_save_preserves_previous_known_good_backup() { + let dir = unique_state_dir("known-good-backup"); + std::fs::create_dir_all(&dir).expect("create state fixture dir"); + let path = dir.join("desktop-state.json"); + let previous = test_stored_config("previous"); + let replacement = test_stored_config("replacement"); + save_config_atomically(&path, &serde_json::to_vec_pretty(&previous).unwrap()) + .expect("initial atomic save"); + save_config_atomically(&path, &serde_json::to_vec_pretty(&replacement).unwrap()) + .expect("replacement atomic save"); + + match read_stored_config(&path).expect("read primary") { + StoredConfigFile::Valid { config, .. } => assert_eq!(config, replacement), + other => panic!("replacement state was not valid: {other:?}"), + } + match read_stored_config(&desktop_state_backup_path(&path)).expect("read backup") { + StoredConfigFile::Valid { config, .. } => assert_eq!(config, previous), + other => panic!("previous snapshot was not valid: {other:?}"), + } + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn corrupt_primary_with_valid_backup_recovers_explicitly() { + let dir = unique_state_dir("recover-backup"); + std::fs::create_dir_all(&dir).expect("create state fixture dir"); + let path = dir.join("desktop-state.json"); + let expected = test_stored_config("recovered"); + write_atomic_file( + &desktop_state_backup_path(&path), + &serde_json::to_vec_pretty(&expected).unwrap(), + ) + .expect("write valid backup"); + std::fs::write(&path, b"{corrupt-primary").expect("write corrupt primary"); + let activity = ActivityLog::default(); + + let recovered = load_config(&path, &activity).expect("recover from backup"); + assert_eq!(recovered, expected); + assert!(matches!( + read_stored_config(&path).expect("read restored primary"), + StoredConfigFile::Valid { .. } + )); + assert!(activity.snapshot().iter().any(|entry| { + entry.event_kind == ActivityEventKind::StateRecovered && entry.source == "desktop_state" + })); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn corrupt_primary_and_backup_returns_explicit_error() { + let dir = unique_state_dir("both-corrupt"); + std::fs::create_dir_all(&dir).expect("create state fixture dir"); + let path = dir.join("desktop-state.json"); + std::fs::write(&path, b"{corrupt-primary").expect("write corrupt primary"); + std::fs::write(desktop_state_backup_path(&path), b"{corrupt-backup") + .expect("write corrupt backup"); + + let error = load_config(&path, &ActivityLog::default()) + .expect_err("both corrupt copies must fail closed"); + assert_eq!(error.code, "desktop_state_corrupt"); + assert_eq!( + error + .details + .as_ref() + .and_then(|details| details.get("category")) + .and_then(Value::as_str), + Some("state_corrupt") + ); + let _ = std::fs::remove_dir_all(dir); + } + #[test] fn stored_runtime_contains_paths_not_credentials() { let runtime = StoredRuntime { @@ -1782,10 +2346,10 @@ mod tests { .unwrap_or_default() .as_nanos() )); - let state = Arc::new(AppState::new( - data_dir.clone(), - data_dir.join("test-resources"), - )); + let state = Arc::new( + AppState::new(data_dir.clone(), data_dir.join("test-resources")) + .expect("create Desktop test state"), + ); let (started_tx, started_rx) = tokio::sync::oneshot::channel(); let (release_tx, release_rx) = tokio::sync::oneshot::channel(); let operation_state = Arc::clone(&state); @@ -1931,12 +2495,12 @@ mod tests { let long_marker = data_dir.join("long-lived-pids.txt"); let one_shot_marker = data_dir.join("one-shot-pids.txt"); std::fs::create_dir_all(&data_dir).expect("create shutdown fixture dir"); - let state = Arc::new(AppState::new( - data_dir.clone(), - data_dir.join("test-resources"), - )); + let state = Arc::new( + AppState::new(data_dir.clone(), data_dir.join("test-resources")) + .expect("create Desktop shutdown test state"), + ); - let mut long_command = tokio::process::Command::new("/bin/sh"); + let mut long_command = std::process::Command::new("/bin/sh"); long_command.args([ "-c", "sleep 8 & descendant=$!; printf '%s %s\\n' \"$$\" \"$descendant\" > \"$1\"; wait \"$descendant\"", @@ -2035,7 +2599,8 @@ mod tests { std::process::id() )); let _ = std::fs::remove_dir_all(&data_dir); - let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")); + let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")) + .expect("create local dogfood state"); let cancellation = CancellationContext::never(); let setup = core.configure_local_setup(&project, &cancellation).await; let snapshot = match setup { @@ -2120,7 +2685,8 @@ mod tests { std::process::id() )); let _ = std::fs::remove_dir_all(&data_dir); - let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")); + let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")) + .expect("create Quick Share dogfood state"); let cancellation = CancellationContext::never(); let started = core .start_quick_share(&project, "none", &cancellation) @@ -2171,7 +2737,8 @@ mod tests { let _ = std::fs::remove_dir_all(&host_data); let _ = std::fs::remove_dir_all(&client_data); - let mut host = DesktopCore::new(host_data.clone(), host_data.join("test-resources")); + let mut host = DesktopCore::new(host_data.clone(), host_data.join("test-resources")) + .expect("create remote dogfood host state"); let cancellation = CancellationContext::never(); let host_runtime = host_data.join("runtime"); let env_file = host_runtime.join("webcodex.env"); @@ -2194,11 +2761,19 @@ mod tests { .await .expect("start remote dogfood Server"); - let mut client = DesktopCore::new(client_data.clone(), client_data.join("test-resources")); + let mut client = DesktopCore::new(client_data.clone(), client_data.join("test-resources")) + .expect("create remote dogfood client state"); let result: DesktopResult<(DesktopStateSnapshot, DesktopStateSnapshot, bool, bool)> = async { - host.wait_for_server(&server_url, Some(&env_file), None, &cancellation) - .await?; + host.wait_for_server( + &server_url, + Some(&env_file), + None, + &cancellation, + Deadline::after(SERVER_READY_TIMEOUT), + true, + ) + .await?; let pairing_code = host .adapter .create_local_pairing(&server_url, &env_file, &cancellation) @@ -2316,7 +2891,8 @@ mod tests { .unwrap_or_default() .as_nanos() )); - let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")); + let mut core = DesktopCore::new(data_dir.clone(), data_dir.join("test-resources")) + .expect("create tunnel failure state"); let topology = RuntimeTopology { experience: Experience::Full, server: ServerTopology::Local, @@ -2340,7 +2916,7 @@ mod tests { ready_for_chatgpt: true, }); - let mut command = tokio::process::Command::new("cmd.exe"); + let mut command = std::process::Command::new("cmd.exe"); command.args(["/D", "/C", "exit", "/B", "23"]); core.supervisor .lock() diff --git a/apps/desktop/src-tauri/src/webcodex/adapter.rs b/apps/desktop/src-tauri/src/webcodex/adapter.rs index 7cc89126a..9a36e9f9c 100644 --- a/apps/desktop/src-tauri/src/webcodex/adapter.rs +++ b/apps/desktop/src-tauri/src/webcodex/adapter.rs @@ -1,13 +1,14 @@ -use super::cli::{run_json, ResolvedBinaries}; +use super::cli::{run_json, run_json_until, ResolvedBinaries}; use super::models::{ LoginOutput, OpsProjectsOutput, PairingCreateOutput, RunnerStatusOutput, ServerStatusOutput, }; +use crate::deadline::Deadline; use crate::error::{DesktopError, DesktopResult}; use crate::models::ProjectSelection; use crate::operation::CancellationContext; use crate::platform; use std::path::{Path, PathBuf}; -use tokio::process::Command; +use std::process::Command; use url::Url; #[derive(Debug, Clone, PartialEq, Eq)] @@ -46,6 +47,24 @@ impl WebCodexAdapter { Ok(self.binaries.as_ref().expect("resolved above")) } + pub async fn ensure_binaries_until( + &mut self, + cancellation: &CancellationContext, + deadline: Deadline, + ) -> DesktopResult<&ResolvedBinaries> { + if self.binaries.is_none() { + self.binaries = Some( + ResolvedBinaries::resolve_until( + self.bundled_runtime_dir.as_deref(), + cancellation, + deadline, + ) + .await?, + ); + } + Ok(self.binaries.as_ref().expect("resolved above")) + } + pub fn binaries(&self) -> DesktopResult<&ResolvedBinaries> { self.binaries.as_ref().ok_or_else(|| { DesktopError::new( @@ -113,7 +132,44 @@ impl WebCodexAdapter { token_file: Option<&Path>, cancellation: &CancellationContext, ) -> DesktopResult { - let webcodex = self.ensure_binaries(cancellation).await?.webcodex.clone(); + self.server_status_with_deadline(server_url, env_file, token_file, cancellation, None) + .await + } + + pub async fn server_status_until( + &mut self, + server_url: Option<&str>, + env_file: Option<&Path>, + token_file: Option<&Path>, + cancellation: &CancellationContext, + deadline: Deadline, + ) -> DesktopResult { + self.server_status_with_deadline( + server_url, + env_file, + token_file, + cancellation, + Some(deadline), + ) + .await + } + + async fn server_status_with_deadline( + &mut self, + server_url: Option<&str>, + env_file: Option<&Path>, + token_file: Option<&Path>, + cancellation: &CancellationContext, + deadline: Option, + ) -> DesktopResult { + let webcodex = match deadline { + Some(deadline) => self + .ensure_binaries_until(cancellation, deadline) + .await? + .webcodex + .clone(), + None => self.ensure_binaries(cancellation).await?.webcodex.clone(), + }; let mut args = vec!["server".into(), "status".into()]; if let Some(url) = server_url { args.extend(["--url".into(), url.into()]); @@ -125,8 +181,12 @@ impl WebCodexAdapter { args.extend(["--token-file".into(), path.to_string_lossy().to_string()]); } args.push("--json".into()); - let output: ServerStatusOutput = - run_json(&webcodex, &args, None, false, cancellation).await?; + let output: ServerStatusOutput = match deadline { + Some(deadline) => { + run_json_until(&webcodex, &args, None, false, cancellation, deadline).await? + } + None => run_json(&webcodex, &args, None, false, cancellation).await?, + }; if output.probe_url.trim().is_empty() { return Err(invalid_contract("server status")); } @@ -147,6 +207,7 @@ impl WebCodexAdapter { pub fn local_server_command(&self, env_file: &Path) -> DesktopResult { let binaries = self.binaries()?; let mut command = Command::new(&binaries.server); + command.arg("--stop-on-stdin-eof"); command.env("WEBCODEX_ENV_FILE", env_file); remove_tunnel_credentials(&mut command); Ok(command) @@ -155,7 +216,10 @@ impl WebCodexAdapter { pub fn local_runner_command(&self, config: &Path) -> DesktopResult { let binaries = self.binaries()?; let mut command = Command::new(&binaries.runner); - command.arg("--config").arg(config); + command + .arg("--config") + .arg(config) + .arg("--stop-on-stdin-eof"); remove_tunnel_credentials(&mut command); Ok(command) } @@ -304,25 +368,51 @@ impl WebCodexAdapter { identity: &ProjectRuntimeIdentity, cancellation: &CancellationContext, ) -> DesktopResult { - let webcodex = self.ensure_binaries(cancellation).await?.webcodex.clone(); - let output: RunnerStatusOutput = run_json( - &webcodex, - &[ - "runner".into(), - "status".into(), - "--config".into(), - identity.runner_config.to_string_lossy().to_string(), - "--server-url".into(), - identity.server_url.clone(), - "--user-token-file".into(), - identity.user_token_file.to_string_lossy().to_string(), - "--json".into(), - ], - None, - false, - cancellation, - ) - .await?; + self.runner_ready_with_deadline(identity, cancellation, None) + .await + } + + pub async fn runner_ready_until( + &mut self, + identity: &ProjectRuntimeIdentity, + cancellation: &CancellationContext, + deadline: Deadline, + ) -> DesktopResult { + self.runner_ready_with_deadline(identity, cancellation, Some(deadline)) + .await + } + + async fn runner_ready_with_deadline( + &mut self, + identity: &ProjectRuntimeIdentity, + cancellation: &CancellationContext, + deadline: Option, + ) -> DesktopResult { + let webcodex = match deadline { + Some(deadline) => self + .ensure_binaries_until(cancellation, deadline) + .await? + .webcodex + .clone(), + None => self.ensure_binaries(cancellation).await?.webcodex.clone(), + }; + let args = [ + "runner".into(), + "status".into(), + "--config".into(), + identity.runner_config.to_string_lossy().to_string(), + "--server-url".into(), + identity.server_url.clone(), + "--user-token-file".into(), + identity.user_token_file.to_string_lossy().to_string(), + "--json".into(), + ]; + let output: RunnerStatusOutput = match deadline { + Some(deadline) => { + run_json_until(&webcodex, &args, None, false, cancellation, deadline).await? + } + None => run_json(&webcodex, &args, None, false, cancellation).await?, + }; if output.config.path.trim().is_empty() || output.config.client_id.trim().is_empty() || output.config.server_url.trim().is_empty() @@ -346,23 +436,49 @@ impl WebCodexAdapter { identity: &ProjectRuntimeIdentity, cancellation: &CancellationContext, ) -> DesktopResult { - let webcodex = self.ensure_binaries(cancellation).await?.webcodex.clone(); - let output: OpsProjectsOutput = run_json( - &webcodex, - &[ - "ops".into(), - "projects".into(), - "--server-url".into(), - identity.server_url.clone(), - "--token-file".into(), - identity.user_token_file.to_string_lossy().to_string(), - "--json".into(), - ], - None, - false, - cancellation, - ) - .await?; + self.project_ready_with_deadline(identity, cancellation, None) + .await + } + + pub async fn project_ready_until( + &mut self, + identity: &ProjectRuntimeIdentity, + cancellation: &CancellationContext, + deadline: Deadline, + ) -> DesktopResult { + self.project_ready_with_deadline(identity, cancellation, Some(deadline)) + .await + } + + async fn project_ready_with_deadline( + &mut self, + identity: &ProjectRuntimeIdentity, + cancellation: &CancellationContext, + deadline: Option, + ) -> DesktopResult { + let webcodex = match deadline { + Some(deadline) => self + .ensure_binaries_until(cancellation, deadline) + .await? + .webcodex + .clone(), + None => self.ensure_binaries(cancellation).await?.webcodex.clone(), + }; + let args = [ + "ops".into(), + "projects".into(), + "--server-url".into(), + identity.server_url.clone(), + "--token-file".into(), + identity.user_token_file.to_string_lossy().to_string(), + "--json".into(), + ]; + let output: OpsProjectsOutput = match deadline { + Some(deadline) => { + run_json_until(&webcodex, &args, None, false, cancellation, deadline).await? + } + None => run_json(&webcodex, &args, None, false, cancellation).await?, + }; Ok(output .summary .projects @@ -549,7 +665,7 @@ mod tests { let local = adapter .quick_share_command(Path::new("repo"), "none") .unwrap(); - let local_env: Vec<_> = local.as_std().get_envs().collect(); + let local_env: Vec<_> = local.get_envs().collect(); for key in [ "CONTROL_PLANE_API_KEY", "CONTROL_PLANE_TUNNEL_ID", @@ -564,7 +680,7 @@ mod tests { let openai = adapter .quick_share_command(Path::new("repo"), "openai") .unwrap(); - let openai_env: Vec<_> = openai.as_std().get_envs().collect(); + let openai_env: Vec<_> = openai.get_envs().collect(); for key in ["OPENAI_ADMIN_KEY", "OPENAI_API_KEY"] { assert!(openai_env .iter() @@ -596,7 +712,6 @@ mod tests { .regular_tunnel_command(Path::new("server.env"), Path::new("user-token")) .unwrap(); let args: Vec<_> = command - .as_std() .get_args() .map(|value| value.to_string_lossy().to_string()) .collect(); @@ -615,7 +730,7 @@ mod tests { "--stop-on-stdin-eof", ] ); - let env: Vec<_> = command.as_std().get_envs().collect(); + let env: Vec<_> = command.get_envs().collect(); for key in ["OPENAI_ADMIN_KEY", "OPENAI_API_KEY"] { assert!(env .iter() diff --git a/apps/desktop/src-tauri/src/webcodex/cli.rs b/apps/desktop/src-tauri/src/webcodex/cli.rs index 1a0a62257..1b749e335 100644 --- a/apps/desktop/src-tauri/src/webcodex/cli.rs +++ b/apps/desktop/src-tauri/src/webcodex/cli.rs @@ -1,3 +1,4 @@ +use crate::deadline::Deadline; use crate::error::{DesktopError, DesktopResult}; use crate::models::BinaryInfo; use crate::operation::{cancelled_error, CancellationContext}; @@ -50,8 +51,24 @@ impl ResolvedBinaries { pub async fn resolve( bundled_runtime_dir: Option<&Path>, cancellation: &CancellationContext, + ) -> DesktopResult { + Self::resolve_until( + bundled_runtime_dir, + cancellation, + Deadline::after(CLI_TIMEOUT), + ) + .await + } + + pub async fn resolve_until( + bundled_runtime_dir: Option<&Path>, + cancellation: &CancellationContext, + deadline: Deadline, ) -> DesktopResult { cancellation.check()?; + if deadline.is_elapsed() { + return Err(timeout_error()); + } let (directory, source) = if let Some(directory) = bundled_runtime_dir.filter(|path| path.is_dir()) { (directory.to_path_buf(), ResolvedBinarySource::Bundled) @@ -131,9 +148,9 @@ impl ResolvedBinaries { } } - let cli_version = binary_version(&webcodex, cancellation).await?; - let server_version = binary_version(&server, cancellation).await?; - let runner_version = binary_version(&runner, cancellation).await?; + let cli_version = binary_version(&webcodex, cancellation, deadline).await?; + let server_version = binary_version(&server, cancellation, deadline).await?; + let runner_version = binary_version(&runner, cancellation, deadline).await?; if cli_version.version != server_version.version || cli_version.version != runner_version.version || cli_version.git_commit != server_version.git_commit @@ -182,8 +199,17 @@ struct VersionLine { async fn binary_version( path: &Path, cancellation: &CancellationContext, + deadline: Deadline, ) -> DesktopResult { - let output = run_bounded(path, &["--version".to_string()], None, false, cancellation).await?; + let output = run_bounded_until( + path, + &["--version".to_string()], + None, + false, + cancellation, + deadline, + ) + .await?; if output.exit_code != Some(0) { return Err(DesktopError::new( "binary_probe_failed", @@ -225,7 +251,34 @@ pub async fn run_json( secret_output: bool, cancellation: &CancellationContext, ) -> DesktopResult { - let output = run_bounded(executable, args, stdin, secret_output, cancellation).await?; + run_json_until( + executable, + args, + stdin, + secret_output, + cancellation, + Deadline::after(CLI_TIMEOUT), + ) + .await +} + +pub async fn run_json_until( + executable: &Path, + args: &[String], + stdin: Option<&[u8]>, + secret_output: bool, + cancellation: &CancellationContext, + deadline: Deadline, +) -> DesktopResult { + let output = run_bounded_until( + executable, + args, + stdin, + secret_output, + cancellation, + deadline, + ) + .await?; if output.exit_code != Some(0) { return Err(DesktopError::new( "webcodex_command_failed", @@ -251,34 +304,38 @@ struct BoundedOutput { stderr: Vec, } -async fn run_bounded( +#[cfg(test)] +async fn run_bounded_with_timeout( executable: &Path, args: &[String], stdin_payload: Option<&[u8]>, - secret_output: bool, + _secret_output: bool, cancellation: &CancellationContext, + timeout: Duration, ) -> DesktopResult { - run_bounded_with_timeout( + run_bounded_until( executable, args, stdin_payload, - secret_output, + _secret_output, cancellation, - CLI_TIMEOUT, + Deadline::after(timeout), ) .await } -async fn run_bounded_with_timeout( +async fn run_bounded_until( executable: &Path, args: &[String], stdin_payload: Option<&[u8]>, _secret_output: bool, cancellation: &CancellationContext, - timeout: Duration, + deadline: Deadline, ) -> DesktopResult { - let deadline = Instant::now() + timeout; cancellation.check()?; + if deadline.is_elapsed() { + return Err(timeout_error()); + } if stdin_payload.is_some_and(|payload| payload.len() > CLI_INPUT_BYTES) { return Err(DesktopError::new( "webcodex_command_input_failed", @@ -352,7 +409,7 @@ async fn run_bounded_with_timeout( stdin_task = Some(tokio::task::spawn_blocking(move || { write_and_close_stdin(stdin, payload) })); - match await_stdin_writer(&mut stdin_task, deadline, cancellation).await { + match await_stdin_writer(&mut stdin_task, deadline.instant(), cancellation).await { Ok(Ok(())) => {} Ok(Err(_)) => { cleanup_command( @@ -394,7 +451,7 @@ async fn run_bounded_with_timeout( } } - let status = match wait_for_direct_child(&mut child, deadline, cancellation).await { + let status = match wait_for_direct_child(&mut child, deadline.instant(), cancellation).await { Ok(Ok(status)) => status, Ok(Err(_)) => { cleanup_command( @@ -435,7 +492,7 @@ async fn run_bounded_with_timeout( } }; - let stdout = match await_reader(&mut stdout_task, deadline, cancellation).await { + let stdout = match await_reader(&mut stdout_task, deadline.instant(), cancellation).await { Ok(output) => output, Err(interruption) => { cleanup_command( @@ -449,7 +506,7 @@ async fn run_bounded_with_timeout( return Err(interruption.into_error()); } }; - let stderr = match await_reader(&mut stderr_task, deadline, cancellation).await { + let stderr = match await_reader(&mut stderr_task, deadline.instant(), cancellation).await { Ok(output) => output, Err(interruption) => { cleanup_command( @@ -464,7 +521,7 @@ async fn run_bounded_with_timeout( } }; - match wait_for_tree_exit(&child, deadline, cancellation).await { + match wait_for_tree_exit(&child, deadline.instant(), cancellation).await { Ok(Ok(())) => {} Ok(Err(_)) => { cleanup_command( @@ -653,8 +710,12 @@ async fn await_reader( Ok(result.unwrap_or_default()) } -async fn cleanup_child_only(child: &mut ManagedChild, operation_deadline: Instant) { - terminate_managed_tree(child, cleanup_deadline(operation_deadline)).await; +async fn cleanup_child_only(child: &mut ManagedChild, operation_deadline: Deadline) { + terminate_managed_tree( + child, + operation_deadline.cleanup_deadline(CLI_CLEANUP_SLACK), + ) + .await; } async fn cleanup_command( @@ -662,9 +723,9 @@ async fn cleanup_command( stdin_task: &mut Option>>, stdout_task: &mut Option>>, stderr_task: &mut Option>>, - operation_deadline: Instant, + operation_deadline: Deadline, ) { - let deadline = cleanup_deadline(operation_deadline); + let deadline = operation_deadline.cleanup_deadline(CLI_CLEANUP_SLACK); terminate_managed_tree(child, deadline).await; finish_task(stdin_task, deadline).await; finish_task(stdout_task, deadline).await; @@ -711,15 +772,6 @@ async fn finish_task(task: &mut Option>, deadline: Instant) { } } -fn cleanup_deadline(operation_deadline: Instant) -> Instant { - let now = Instant::now(); - if operation_deadline > now { - std::cmp::min(operation_deadline, now + CLI_CLEANUP_SLACK) - } else { - now + CLI_CLEANUP_SLACK - } -} - fn timeout_error() -> DesktopError { DesktopError::new( "webcodex_command_timeout", @@ -801,7 +853,7 @@ mod tests { assert_eq!(ResolvedBinarySource::Bundled.label(), "Bundled"); } - #[cfg(target_os = "macos")] + #[cfg(unix)] fn process_exists(pid: u32) -> bool { let Ok(pid) = i32::try_from(pid) else { return false; @@ -816,7 +868,7 @@ mod tests { ) } - #[cfg(target_os = "macos")] + #[cfg(unix)] fn unique_marker(name: &str) -> PathBuf { std::env::temp_dir().join(format!( "webcodex-cli-{name}-{}-{}", @@ -828,7 +880,7 @@ mod tests { )) } - #[cfg(target_os = "macos")] + #[cfg(unix)] fn read_fixture_pids(marker: &Path) -> Vec { std::fs::read_to_string(marker) .expect("fixture must publish owned pids") @@ -837,7 +889,7 @@ mod tests { .collect() } - #[cfg(target_os = "macos")] + #[cfg(unix)] #[tokio::test] async fn blocked_stdin_uses_one_total_deadline_and_reclaims_owned_tree() { let marker = unique_marker("blocked-stdin"); @@ -871,7 +923,7 @@ mod tests { let _ = std::fs::remove_file(marker); } - #[cfg(target_os = "macos")] + #[cfg(unix)] #[tokio::test] async fn inherited_output_pipe_cannot_extend_the_command_lifetime() { let marker = unique_marker("inherited-pipe"); @@ -904,7 +956,7 @@ mod tests { let _ = std::fs::remove_file(marker); } - #[cfg(target_os = "macos")] + #[cfg(unix)] #[tokio::test] async fn explicit_cancel_interrupts_blocked_stdin_and_keeps_cancel_classification() { let marker = unique_marker("cancelled-stdin"); @@ -949,6 +1001,101 @@ mod tests { let _ = std::fs::remove_file(marker); } + #[cfg(unix)] + #[tokio::test] + async fn absolute_deadline_is_not_reset_by_nested_cli_work() { + let cancellation = CancellationContext::never(); + let outer_started = Instant::now(); + let deadline = Deadline::after(Duration::from_millis(350)); + tokio::time::sleep(Duration::from_millis(150)).await; + let error = run_bounded_until( + Path::new("/bin/sh"), + &["-c".to_string(), "sleep 8".to_string()], + None, + false, + &cancellation, + deadline, + ) + .await + .unwrap_err(); + assert_eq!(error.code, "webcodex_command_timeout"); + assert!( + outer_started.elapsed() < Duration::from_secs(3), + "nested CLI work must consume the original deadline plus only bounded cleanup slack" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn elapsed_deadline_does_not_start_a_new_cli_command() { + let marker = unique_marker("expired-before-spawn"); + let cancellation = CancellationContext::never(); + let deadline = Deadline::after(Duration::from_millis(20)); + tokio::time::sleep(Duration::from_millis(30)).await; + let args = vec![ + "-c".to_string(), + "printf started > \"$1\"".to_string(), + "webcodex-expired-deadline".to_string(), + marker.to_string_lossy().to_string(), + ]; + let error = run_bounded_until( + Path::new("/bin/sh"), + &args, + None, + false, + &cancellation, + deadline, + ) + .await + .unwrap_err(); + assert_eq!(error.code, "webcodex_command_timeout"); + assert!(!marker.exists(), "expired deadline must prevent spawn"); + } + + #[cfg(unix)] + #[tokio::test] + async fn cancellation_during_output_drain_reclaims_the_owned_tree() { + let marker = unique_marker("cancel-output-drain"); + let args = vec![ + "-c".to_string(), + "sleep 8 & descendant=$!; printf '%s\\n' \"$descendant\" > \"$1\"; exit 0".to_string(), + "webcodex-cancel-output-drain".to_string(), + marker.to_string_lossy().to_string(), + ]; + let operation = CancellationSignal::new(); + let cancellation = CancellationContext::new(operation.clone(), CancellationSignal::new()); + let command = tokio::spawn(async move { + run_bounded_with_timeout( + Path::new("/bin/sh"), + &args, + None, + false, + &cancellation, + Duration::from_secs(8), + ) + .await + }); + let marker_deadline = Instant::now() + Duration::from_secs(2); + while !marker.is_file() { + assert!(Instant::now() < marker_deadline, "fixture did not start"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + operation.cancel(); + let error = tokio::time::timeout(Duration::from_secs(4), command) + .await + .expect("cancelled drain must return promptly") + .expect("fixture task") + .unwrap_err(); + assert_eq!(error.code, "desktop_operation_cancelled"); + for pid in read_fixture_pids(&marker) { + assert!( + !process_exists(pid), + "pipe-holding PID {pid} survived output-drain cancellation" + ); + } + let _ = std::fs::remove_file(marker); + } + #[cfg(target_os = "windows")] fn windows_process_exists(pid: u32) -> bool { let filter = format!("PID eq {pid}"); diff --git a/crates/webcodex-process/src/bin/process_tree_helper.rs b/crates/webcodex-process/src/bin/process_tree_helper.rs index f3e43c796..fa7f38bde 100644 --- a/crates/webcodex-process/src/bin/process_tree_helper.rs +++ b/crates/webcodex-process/src/bin/process_tree_helper.rs @@ -13,8 +13,11 @@ //! * `grandchild ` — sleep `delay`, write our own PID //! to `marker`, sleep until `total`, then exit 0. //! * `hold-stdout` — write `PING` and keep the stdout write end open forever. +//! * `spawn-parent-lease-child ` — spawn a child with piped stdin and +//! exit immediately; used to model an owner process disappearing. +//! * `parent-lease-child ` — wait for stdin EOF, write `marker`, exit. -use std::io::Write; +use std::io::{Read, Write}; use std::process::{Command, Stdio}; fn main() { @@ -54,6 +57,31 @@ fn main() { std::thread::sleep(std::time::Duration::from_secs(3600)); } } + "spawn-parent-lease-child" => { + let marker = args.get(2).expect("marker path"); + let self_exe = std::env::current_exe().expect("current_exe"); + let mut cmd = Command::new(self_exe); + cmd.args(["parent-lease-child", marker]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + #[allow(clippy::zombie_processes)] + let child = cmd.spawn().expect("spawn parent-lease child"); + println!("LEASE_CHILD_PID={}", child.id()); + std::io::stdout().flush().expect("flush stdout"); + // process::exit intentionally skips Rust destructors. The kernel + // still closes the parent's pipe write handle, exactly as it does + // when the owning Desktop process disappears unexpectedly. + std::process::exit(0); + } + "parent-lease-child" => { + let marker = args.get(2).expect("marker path"); + let mut bytes = Vec::new(); + std::io::stdin() + .read_to_end(&mut bytes) + .expect("read parent lease stdin"); + std::fs::write(marker, b"parent_eof").expect("write parent EOF marker"); + } other => { eprintln!("process_tree_helper: unknown mode: {other}"); std::process::exit(2); diff --git a/crates/webcodex-process/tests/managed_child.rs b/crates/webcodex-process/tests/managed_child.rs index 598fad734..d1f95d785 100644 --- a/crates/webcodex-process/tests/managed_child.rs +++ b/crates/webcodex-process/tests/managed_child.rs @@ -157,6 +157,23 @@ fn wait_until_file(path: &Path, timeout: Duration) -> bool { } } +#[test] +fn inherited_stdin_lease_detects_parent_process_disappearance() { + let marker = unique_temp_path("parent-lease-eof"); + let mut parent = Command::new(helper()); + parent + .args(["spawn-parent-lease-child", marker.to_str().unwrap()]) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let mut parent = parent.spawn().expect("spawn parent-lease owner"); + let status = parent.wait().expect("wait parent-lease owner"); + assert!(status.success(), "fixture parent should exit cleanly"); + assert!( + wait_until_file(&marker, Duration::from_secs(5)), + "child did not observe stdin EOF after its exact owner process disappeared" + ); +} + /// Spawn the helper in `mode`. When `capture_stdout` is set, returns a /// [`LineReader`] fed from the child's piped stdout. fn spawn_helper( @@ -606,6 +623,32 @@ fn graceful_request_repeated_and_already_exited_do_not_panic() { let _ = result; } +/// Once the owned Unix generation is authoritatively known empty, its numeric +/// process-group id is no longer valid kill authority. This is the stale-PID / +/// PID-reuse fence Desktop relies on by retaining the ManagedChild generation +/// rather than remembering and later targeting a pid/pgid integer. +#[cfg(unix)] +#[test] +fn confirmed_generation_never_reuses_numeric_pgid_as_kill_authority() { + let (mut managed, _) = spawn_helper("sleep", &["0", "0"], false); + let stale_numeric_identity = managed.id(); + let _ = managed.wait().expect("wait direct child"); + assert!(managed + .wait_tree_exit(Duration::from_secs(10)) + .expect("confirm whole-tree exit")); + + assert_eq!( + managed + .request_terminate_tree() + .expect("post-exit graceful request"), + GracefulTermination::AlreadyExited, + "confirmed generation {stale_numeric_identity} must not probe or signal its old numeric pgid" + ); + managed + .terminate_tree() + .expect("post-exit force cleanup is idempotent and must not retarget the numeric pgid"); +} + /// ManagedChild must preserve the platform's normal `Command::spawn` behavior /// for an executable text file without a shebang. Linux reports ENOEXEC while /// macOS's standard-library spawn path executes it through the platform shell; diff --git a/crates/webcodex-runner/src/main.rs b/crates/webcodex-runner/src/main.rs index 2a1a89727..f86590141 100644 --- a/crates/webcodex-runner/src/main.rs +++ b/crates/webcodex-runner/src/main.rs @@ -581,6 +581,7 @@ enum RunnerCliAction { Run { config_path: PathBuf, once: bool, + stop_on_stdin_eof: bool, }, Exit { code: i32, @@ -590,13 +591,14 @@ enum RunnerCliAction { } fn usage() -> &'static str { - "Usage: webcodex-runner [--config PATH] [--once]\n\n\ + "Usage: webcodex-runner [--config PATH] [--once] [--stop-on-stdin-eof]\n\n\ Options:\n\ -h, --help Print help and exit\n\ -V, --version Print version and exit\n\ -c, --config PATH Runner config path for normal runtime\n\ --profile NAME Client config profile for default config path\n\ - --once Complete one successful poll, then exit (polling transport)\n\n\ + --once Complete one successful poll, then exit (polling transport)\n\ + --stop-on-stdin-eof Stop when the invoking parent closes stdin\n\n\ With --profile, the default config path is derived under\n\ /etc/webcodex/clients/ for root or\n\ ~/.config/webcodex/clients/ for non-root users. Explicit\n\ @@ -657,6 +659,7 @@ where let mut config_path: Option = None; let mut profile: Option = None; let mut once = false; + let mut stop_on_stdin_eof = false; let mut args = args.into_iter(); while let Some(arg) = args.next() { match arg.as_str() { @@ -675,6 +678,7 @@ where }); } "--once" => once = true, + "--stop-on-stdin-eof" => stop_on_stdin_eof = true, "--config" | "-c" => { let Some(path) = args.next() else { return Err("--config requires a path".to_string()); @@ -713,7 +717,11 @@ where .unwrap_or_else(default_config_path)? } }; - Ok(RunnerCliAction::Run { config_path, once }) + Ok(RunnerCliAction::Run { + config_path, + once, + stop_on_stdin_eof, + }) } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -5753,8 +5761,12 @@ fn main() { std::process::exit(2); } }; - let (config_path, once) = match action { - RunnerCliAction::Run { config_path, once } => (config_path, once), + let (config_path, once, stop_on_stdin_eof) = match action { + RunnerCliAction::Run { + config_path, + once, + stop_on_stdin_eof, + } => (config_path, once, stop_on_stdin_eof), RunnerCliAction::Exit { code, stdout, @@ -5781,7 +5793,7 @@ fn main() { "webcodex-runner warning: agent token is empty; connecting without Authorization; the server must be started with --open" ); } - if let Err(e) = run_runner(cfg, config_path, once) { + if let Err(e) = run_runner(cfg, config_path, once, stop_on_stdin_eof) { eprintln!("webcodex-runner failed: {}", e); std::process::exit(1); } diff --git a/crates/webcodex-runner/src/main_tests/runner_config.rs b/crates/webcodex-runner/src/main_tests/runner_config.rs index bdd4e9e2a..0c64d2215 100644 --- a/crates/webcodex-runner/src/main_tests/runner_config.rs +++ b/crates/webcodex-runner/src/main_tests/runner_config.rs @@ -419,6 +419,22 @@ fn runner_cli_legacy_runtime_args_are_preserved() { RunnerCliAction::Run { config_path: PathBuf::from("/tmp/agent.toml"), once: true, + stop_on_stdin_eof: false, + } + ); +} + +#[test] +fn runner_parent_liveness_is_explicit_opt_in() { + let _guard = test_env_lock(); + let action = + parse_runner_args(["--config", "/tmp/runner.toml", "--stop-on-stdin-eof"]).unwrap(); + assert_eq!( + action, + RunnerCliAction::Run { + config_path: PathBuf::from("/tmp/runner.toml"), + once: false, + stop_on_stdin_eof: true, } ); } @@ -434,6 +450,7 @@ fn runner_cli_config_env_prefers_runner_name_and_keeps_legacy_alias_fail_closed( RunnerCliAction::Run { config_path: PathBuf::from("/tmp/runner.toml"), once: false, + stop_on_stdin_eof: false, } ); drop(_env); @@ -446,6 +463,7 @@ fn runner_cli_config_env_prefers_runner_name_and_keeps_legacy_alias_fail_closed( RunnerCliAction::Run { config_path: PathBuf::from("/tmp/agent.toml"), once: false, + stop_on_stdin_eof: false, } ); drop(_legacy); @@ -461,6 +479,7 @@ fn runner_cli_config_env_prefers_runner_name_and_keeps_legacy_alias_fail_closed( RunnerCliAction::Run { config_path: PathBuf::from("/tmp/explicit.toml"), once: false, + stop_on_stdin_eof: false, }, "an explicit --config path must not be blocked by conflicting default-path env aliases" ); @@ -469,6 +488,7 @@ fn runner_cli_config_env_prefers_runner_name_and_keeps_legacy_alias_fail_closed( RunnerCliAction::Run { config_path: client_profile_runner_config("special").unwrap(), once: false, + stop_on_stdin_eof: false, }, "an explicit profile must not be blocked by conflicting default-path env aliases" ); @@ -510,6 +530,7 @@ fn runner_cli_profile_derives_default_config_path() { RunnerCliAction::Run { config_path: client_profile_runner_config("special").unwrap(), once: false, + stop_on_stdin_eof: false, } ); } @@ -524,6 +545,7 @@ fn runner_cli_explicit_config_overrides_profile() { RunnerCliAction::Run { config_path: PathBuf::from("/tmp/agent.toml"), once: false, + stop_on_stdin_eof: false, } ); } diff --git a/crates/webcodex-runner/src/webcodex_runner/transport.rs b/crates/webcodex-runner/src/webcodex_runner/transport.rs index 124575fab..aa39c853d 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport.rs @@ -477,6 +477,33 @@ fn install_shutdown_listener( .map_err(|_| "failed to start process shutdown signal listener".to_string()) } +fn install_parent_liveness_listener(runtime: RunnerRuntimeState) -> Result<(), String> { + use std::io::Read; + + let listener = std::thread::Builder::new() + .name("webcodex-runner-parent-lease".to_string()) + .spawn(move || { + let mut stdin = std::io::stdin(); + let mut buffer = [0_u8; 64]; + loop { + match stdin.read(&mut buffer) { + Ok(0) | Err(_) => { + runtime.request_shutdown_signal(); + return; + } + Ok(_) => {} + } + } + }) + .map_err(|_| "failed to start parent-liveness listener".to_string())?; + // This reader is intentionally detached. A blocking stdin read cannot be + // cancelled portably; joining it during an ordinary signal-driven shutdown + // would hang until the parent closed the lease. Process exit reclaims the + // detached thread, while EOF still triggers exact-generation shutdown. + drop(listener); + Ok(()) +} + fn send_polling_offline_best_effort(client: &Client, cfg: &RunnerConfig, runner_instance_id: &str) { let url = format!( "{}{}", @@ -1440,6 +1467,7 @@ pub(crate) fn run_runner( cfg: RunnerConfig, config_path: PathBuf, once: bool, + stop_on_stdin_eof: bool, ) -> Result<(), String> { // Generate the per-process agent instance identity once. It is stable for // the whole process lifetime, including across WebSocket reconnects, so the @@ -1507,6 +1535,15 @@ pub(crate) fn run_runner( tracing::error!(error = %error, "detached Job state root is unavailable"); } } + if stop_on_stdin_eof { + if let Err(error) = install_parent_liveness_listener(runtime.clone()) { + #[cfg(windows)] + if let Some(diagnostics) = exit_diagnostics.as_ref() { + diagnostics.mark_terminal(false, "parent_liveness_listener_install_failed", None); + } + return Err(error); + } + } let shutdown_listener = match install_shutdown_listener(runtime.clone()) { Ok(listener) => listener, Err(error) => { diff --git a/src/bin/webcodex-server.rs b/src/bin/webcodex-server.rs index dd5e283f1..fad2155ed 100644 --- a/src/bin/webcodex-server.rs +++ b/src/bin/webcodex-server.rs @@ -16,9 +16,10 @@ fn build_server_runtime() -> std::io::Result { fn main() -> Result<(), Box> { match server_binary_action(std::env::args().skip(1)) { - ServerBinaryAction::Run => { + ServerBinaryAction::Run { stop_on_stdin_eof } => { webcodex::prepare_server_process_environment().map_err(std::io::Error::other)?; - build_server_runtime()?.block_on(webcodex::run_server()) + build_server_runtime()? + .block_on(webcodex::run_server_with_parent_liveness(stop_on_stdin_eof)) } ServerBinaryAction::Exit { code, diff --git a/src/lib.rs b/src/lib.rs index 84c72c6a3..6a1ab6910 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -94,7 +94,9 @@ pub use startup::{ #[derive(Debug, Clone, PartialEq, Eq)] pub enum ServerBinaryAction { - Run, + Run { + stop_on_stdin_eof: bool, + }, Exit { code: i32, stdout: String, @@ -112,10 +114,15 @@ where .map(|arg| arg.as_ref().to_string()) .collect(); match args.as_slice() { - [] => ServerBinaryAction::Run, + [] => ServerBinaryAction::Run { + stop_on_stdin_eof: false, + }, + [arg] if arg == "--stop-on-stdin-eof" => ServerBinaryAction::Run { + stop_on_stdin_eof: true, + }, [arg] if matches!(arg.as_str(), "--help" | "-h") => ServerBinaryAction::Exit { code: 0, - stdout: "Usage: webcodex-server [OPTIONS]\n\nRun the WebCodex server runtime.\n\nOptions:\n -h, --help Print help and exit\n -V, --version Print version and exit\n".to_string(), + stdout: "Usage: webcodex-server [OPTIONS]\n\nRun the WebCodex server runtime.\n\nOptions:\n --stop-on-stdin-eof Stop when the invoking parent closes stdin\n -h, --help Print help and exit\n -V, --version Print version and exit\n".to_string(), stderr: String::new(), }, [arg] if matches!(arg.as_str(), "--version" | "-V") => ServerBinaryAction::Exit { @@ -176,6 +183,13 @@ pub fn prepare_server_process_environment() -> Result<(), String> { } pub async fn run_server() -> Result<(), Box> { + run_server_with_parent_liveness(false).await +} + +#[doc(hidden)] +pub async fn run_server_with_parent_liveness( + stop_on_stdin_eof: bool, +) -> Result<(), Box> { let env_loads = match PREPARED_SERVER_ENV_LOADS.get() { Some(prepared) => prepared.clone(), None => load_startup_env_files().map_err(std::io::Error::other)?, @@ -804,6 +818,7 @@ only for local/trusted-network demos." router, shutdown_coordinator, std::time::Duration::from_secs(SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_SECS), + stop_on_stdin_eof, ) .await?; Ok(()) @@ -813,6 +828,22 @@ only for local/trusted-network demos." mod tests { use super::*; + #[test] + fn server_parent_liveness_is_explicit_opt_in() { + assert_eq!( + server_binary_action(std::iter::empty::<&str>()), + ServerBinaryAction::Run { + stop_on_stdin_eof: false, + } + ); + assert_eq!( + server_binary_action(["--stop-on-stdin-eof"]), + ServerBinaryAction::Run { + stop_on_stdin_eof: true, + } + ); + } + #[test] fn test_parse_env_file_line_basic() { let parsed = parse_env_file_line("WEBCODEX_ADDR=127.0.0.1:8080") diff --git a/src/server_shutdown.rs b/src/server_shutdown.rs index 468b50a0e..100573c66 100644 --- a/src/server_shutdown.rs +++ b/src/server_shutdown.rs @@ -11,6 +11,7 @@ use std::time::{Duration, Instant}; pub(crate) enum ShutdownReason { Sigint, Sigterm, + ParentEof, } impl ShutdownReason { @@ -18,6 +19,7 @@ impl ShutdownReason { match self { Self::Sigint => "SIGINT", Self::Sigterm => "SIGTERM", + Self::ParentEof => "parent_stdin_eof", } } } @@ -164,20 +166,51 @@ pub(crate) async fn serve_until_termination( service: S, coordinator: Arc, graceful_timeout: Duration, + stop_on_stdin_eof: bool, ) -> io::Result<()> where A: Acceptor + Send, S: Into + Send, { let mut signals = TerminationSignals::new()?; - serve_with_signal( - server, - service, - coordinator, - signals.recv(), - graceful_timeout, - ) - .await + let parent_eof = if stop_on_stdin_eof { + Some(parent_eof_signal()?) + } else { + None + }; + let signal = async move { + if let Some(parent_eof) = parent_eof { + tokio::select! { + reason = signals.recv() => reason, + _ = parent_eof => ShutdownReason::ParentEof, + } + } else { + signals.recv().await + } + }; + serve_with_signal(server, service, coordinator, signal, graceful_timeout).await +} + +fn parent_eof_signal() -> io::Result> { + use std::io::Read; + + let (tx, rx) = tokio::sync::oneshot::channel(); + std::thread::Builder::new() + .name("webcodex-server-parent-lease".to_string()) + .spawn(move || { + let mut stdin = std::io::stdin(); + let mut buffer = [0_u8; 64]; + loop { + match stdin.read(&mut buffer) { + Ok(0) | Err(_) => { + let _ = tx.send(()); + return; + } + Ok(_) => {} + } + } + })?; + Ok(rx) } async fn serve_with_signal( From 253be5e83d4a64420c906ddb17a5e1de818e9007 Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Sat, 5 Sep 2026 15:45:55 +0800 Subject: [PATCH 2/2] Surface Desktop machine event overflow --- apps/desktop/src-tauri/src/state.rs | 154 ++++++++++++++++++---------- 1 file changed, 98 insertions(+), 56 deletions(-) diff --git a/apps/desktop/src-tauri/src/state.rs b/apps/desktop/src-tauri/src/state.rs index cf9c86da0..2357bcdae 100644 --- a/apps/desktop/src-tauri/src/state.rs +++ b/apps/desktop/src-tauri/src/state.rs @@ -1101,13 +1101,15 @@ impl DesktopCore { self.publish_snapshot(); let event_wait = async { while let Some(value) = events.recv().await { - if value.get("event").and_then(Value::as_str) == Some("ready") { - return Some(value); + match value.get("event").and_then(Value::as_str) { + Some("ready") => return Ok(Some(value)), + Some("machine_event_overflow") => return Err(value), + _ => {} } } - None + Ok(None) }; - let event_value = tokio::select! { + let event_result = tokio::select! { biased; _ = cancellation.cancelled() => { self.stop_process_until( @@ -1117,25 +1119,36 @@ impl DesktopCore { return Err(cancelled_error()); } result = tokio::time::timeout_at(deadline.instant(), event_wait) => { - result.ok().flatten() + result } }; - let Some(event_value) = event_value else { - let logs = self.process_logs(ProcessKind::QuickShare).await; - self.stop_process_until( - ProcessKind::QuickShare, - Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), - ) - .await; - return Err(DesktopError::new( - "quick_share_not_ready", - "Quick Share did not reach verified readiness", - "Check Activity and Tunnel prerequisites, then retry.", - ) - .with_details(serde_json::json!({ - "category": "readiness_timeout", - "diagnostic_lines": logs, - }))); + let event_value = match event_result { + Ok(Ok(Some(value))) => value, + Ok(Err(overflow)) => { + self.stop_process_until( + ProcessKind::QuickShare, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; + return Err(machine_event_overflow_error(&overflow)); + } + Ok(Ok(None)) | Err(_) => { + let logs = self.process_logs(ProcessKind::QuickShare).await; + self.stop_process_until( + ProcessKind::QuickShare, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; + return Err(DesktopError::new( + "quick_share_not_ready", + "Quick Share did not reach verified readiness", + "Check Activity and Tunnel prerequisites, then retry.", + ) + .with_details(serde_json::json!({ + "category": "readiness_timeout", + "diagnostic_lines": logs, + }))); + } }; let event: QuickShareReadyEvent = match serde_json::from_value(event_value) { Ok(event) => event, @@ -1348,13 +1361,15 @@ impl DesktopCore { self.publish_snapshot(); let event_wait = async { while let Some(value) = events.recv().await { - if value.get("event").and_then(Value::as_str) == Some("ready") { - return Some(value); + match value.get("event").and_then(Value::as_str) { + Some("ready") => return Ok(Some(value)), + Some("machine_event_overflow") => return Err(value), + _ => {} } } - None + Ok(None) }; - let event_value = tokio::select! { + let event_result = tokio::select! { biased; _ = cancellation.cancelled() => { self.stop_process_until( @@ -1364,39 +1379,51 @@ impl DesktopCore { return Err(cancelled_error()); } result = tokio::time::timeout_at(deadline.instant(), event_wait) => { - result.ok().flatten() + result } }; - let Some(event_value) = event_value else { - let logs = self.process_logs(ProcessKind::RegularTunnel).await; - self.stop_process_until( - ProcessKind::RegularTunnel, - Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), - ) - .await; - self.snapshot.regular_tunnel = Some(RegularTunnelState { - provider: "openai".to_string(), - status: RegularTunnelStatus::Error, - clipboard_state: "unavailable".to_string(), - clipboard_contains: "tunnel_id".to_string(), - ready_for_chatgpt: false, - }); - self.snapshot.readiness = aggregate_readiness( - ServerReadiness::Ready, - RunnerReadiness::Ready, - ExposureReadiness::Error, - ProjectReadiness::Ready, - ); - apply_regular_tunnel_next_action(&mut self.snapshot, &ExposureReadiness::Error); - return Err(DesktopError::new( - "tunnel_unavailable", - "OpenAI Secure Tunnel did not reach verified readiness", - "Check Activity and the canonical Tunnel prerequisites, then retry.", - ) - .with_details(serde_json::json!({ - "category": "readiness_timeout", - "diagnostic_lines": logs, - }))); + let event_value = match event_result { + Ok(Ok(Some(value))) => value, + Ok(Err(overflow)) => { + self.stop_process_until( + ProcessKind::RegularTunnel, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; + self.snapshot.regular_tunnel = None; + return Err(machine_event_overflow_error(&overflow)); + } + Ok(Ok(None)) | Err(_) => { + let logs = self.process_logs(ProcessKind::RegularTunnel).await; + self.stop_process_until( + ProcessKind::RegularTunnel, + Deadline::at(deadline.cleanup_deadline(READINESS_CLEANUP_SLACK)), + ) + .await; + self.snapshot.regular_tunnel = Some(RegularTunnelState { + provider: "openai".to_string(), + status: RegularTunnelStatus::Error, + clipboard_state: "unavailable".to_string(), + clipboard_contains: "tunnel_id".to_string(), + ready_for_chatgpt: false, + }); + self.snapshot.readiness = aggregate_readiness( + ServerReadiness::Ready, + RunnerReadiness::Ready, + ExposureReadiness::Error, + ProjectReadiness::Ready, + ); + apply_regular_tunnel_next_action(&mut self.snapshot, &ExposureReadiness::Error); + return Err(DesktopError::new( + "tunnel_unavailable", + "OpenAI Secure Tunnel did not reach verified readiness", + "Check Activity and the canonical Tunnel prerequisites, then retry.", + ) + .with_details(serde_json::json!({ + "category": "readiness_timeout", + "diagnostic_lines": logs, + }))); + } }; let event: RegularTunnelReadyEvent = match serde_json::from_value(event_value) { Ok(event) => event, @@ -1793,6 +1820,21 @@ fn readiness_timeout_error( .with_details(serde_json::json!({ "category": "readiness_timeout" })) } +fn machine_event_overflow_error(event: &Value) -> DesktopError { + DesktopError::new( + "machine_event_overflow", + "Desktop could not retain every critical machine-readiness event", + "Retry the operation and inspect Activity if the child keeps emitting excessive machine events.", + ) + .with_details(serde_json::json!({ + "category": "machine_event_overflow", + "dropped_critical": event + .get("dropped_critical") + .and_then(Value::as_u64) + .unwrap_or(1), + })) +} + fn project_snapshot(config: &StoredDesktopConfig) -> Option { let mut project = config.project.clone()?; if identity_from_config(config).is_none() {