diff --git a/codex-rs/app-server-daemon/src/backend/pid.rs b/codex-rs/app-server-daemon/src/backend/pid.rs index 2deb9f37b8..8e79a240e8 100644 --- a/codex-rs/app-server-daemon/src/backend/pid.rs +++ b/codex-rs/app-server-daemon/src/backend/pid.rs @@ -117,7 +117,7 @@ impl PidBackend { #[cfg(any(unix, windows))] pub(crate) async fn start(&self) -> Result> { - self.start_inner().await + self.start_inner(/*replacement*/ None).await } #[cfg(not(any(unix, windows)))] @@ -347,7 +347,7 @@ impl PidBackend { } } - #[cfg(unix)] + #[cfg(not(windows))] fn force_terminate_process(&self, pid: u32) -> Result<()> { match self.command_kind { PidCommandKind::AppServer { .. } => force_terminate_process(pid), @@ -670,6 +670,10 @@ fn force_terminate_process(pid: u32) -> Result<()> { #[cfg(windows)] use force_terminate_process as terminate_process; +#[cfg(windows)] +#[path = "pid_windows.rs"] +mod windows; + #[cfg(any(unix, windows))] #[path = "pid_start.rs"] mod start; diff --git a/codex-rs/app-server-daemon/src/backend/pid_start.rs b/codex-rs/app-server-daemon/src/backend/pid_start.rs index c007a1c4c2..0f0181b178 100644 --- a/codex-rs/app-server-daemon/src/backend/pid_start.rs +++ b/codex-rs/app-server-daemon/src/backend/pid_start.rs @@ -1,18 +1,21 @@ //! Detached process launch and PID publication. Hold the reservation lock until -//! the record is published. +//! the record is published, and on Windows until an updater acknowledges startup. use super::PidBackend; +#[cfg(windows)] +use super::PidCommandKind; use super::PidFileState; use super::PidRecord; use super::read_process_start_time; use anyhow::Context; use anyhow::Result; +use anyhow::bail; use std::process::Stdio; use tokio::fs; use tokio::process::Command; impl PidBackend { - pub(super) async fn start_inner(&self) -> Result> { + pub(super) async fn start_inner(&self, replacement: Option) -> Result> { if let Some(parent) = self.pid_file.parent() { codex_uds::prepare_private_socket_directory(parent) .await @@ -34,6 +37,16 @@ impl PidBackend { match self.read_pid_file_state_with_lock_held().await? { PidFileState::Missing => continue, PidFileState::Running(record) => { + if replacement.as_ref() == Some(&record) { + break; + } + if replacement.is_some() { + if self.record_is_active(&record).await? { + bail!("updater ownership changed before handoff"); + } + // A failed rollback may leave the dead successor's record. + break; + } if self.record_is_active(&record).await? { return Ok(None); } @@ -63,7 +76,9 @@ impl PidBackend { let stderr_log = match self.open_stderr_log().await { Ok(stderr_log) => stderr_log, Err(err) => { - let _ = fs::remove_file(&self.pid_file).await; + if replacement.is_none() { + let _ = fs::remove_file(&self.pid_file).await; + } return Err(err); } }; @@ -95,6 +110,13 @@ impl PidBackend { // Never retry inside the parent's Job Object: that would report a // successful launch that dies when the terminal/SSH session closes. command.creation_flags(DETACHED_PROCESS | CREATE_BREAKAWAY_FROM_JOB); + if matches!(self.command_kind, PidCommandKind::UpdateLoop) { + match fs::remove_file(self.pid_file.with_extension("ready")).await { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => return Err(err).context("failed to clear updater readiness"), + } + } let shutdown_file = self.pid_file.with_extension("shutdown"); match fs::remove_file(&shutdown_file).await { Ok(()) => {} @@ -110,7 +132,9 @@ impl PidBackend { let child = match command.spawn() { Ok(child) => child, Err(err) => { - let _ = fs::remove_file(&self.pid_file).await; + if replacement.is_none() { + let _ = fs::remove_file(&self.pid_file).await; + } return Err(err).with_context(|| { let job_hint = if cfg!(windows) { " (the Windows host job must allow breakaway)" @@ -145,7 +169,9 @@ impl PidBackend { let mut context = format!("failed to record pid-managed app-server process {pid} startup"); super::super::append_stderr_log_tail_context(&self.pid_file, &mut context).await; - let _ = fs::remove_file(&self.pid_file).await; + if replacement.is_none() { + let _ = fs::remove_file(&self.pid_file).await; + } return Err(err).context(context); } }; @@ -153,7 +179,9 @@ impl PidBackend { let temp_pid_file = self.pid_file.with_extension("pid.tmp"); if let Err(err) = fs::write(&temp_pid_file, &contents).await { let _ = self.terminate_process(pid); - let _ = fs::remove_file(&self.pid_file).await; + if replacement.is_none() { + let _ = fs::remove_file(&self.pid_file).await; + } return Err(err).with_context(|| { format!("failed to write pid temp file {}", temp_pid_file.display()) }); @@ -161,11 +189,18 @@ impl PidBackend { if let Err(err) = fs::rename(&temp_pid_file, &self.pid_file).await { let _ = self.terminate_process(pid); let _ = fs::remove_file(&temp_pid_file).await; - let _ = fs::remove_file(&self.pid_file).await; + if replacement.is_none() { + let _ = fs::remove_file(&self.pid_file).await; + } return Err(err).with_context(|| { format!("failed to publish pid file {}", self.pid_file.display()) }); } + #[cfg(windows)] + if matches!(self.command_kind, PidCommandKind::UpdateLoop) { + self.finish_updater_start(&record, replacement.as_ref()) + .await?; + } drop(reservation_lock); Ok(Some(pid)) } diff --git a/codex-rs/app-server-daemon/src/backend/pid_tests.rs b/codex-rs/app-server-daemon/src/backend/pid_tests.rs index 46cc07d808..ce94d75705 100644 --- a/codex-rs/app-server-daemon/src/backend/pid_tests.rs +++ b/codex-rs/app-server-daemon/src/backend/pid_tests.rs @@ -284,6 +284,169 @@ async fn stale_creation_time_never_stops_reused_pid() { assert!(!backend.pid_file.with_extension("shutdown").exists()); } +#[cfg(windows)] +#[tokio::test] +async fn failed_updater_handoff_preserves_predecessor_record() { + let temp = TempDir::new().expect("temp"); + let state_dir = temp.path().join("state"); + codex_uds::prepare_private_socket_directory(&state_dir) + .await + .expect("private state directory"); + let backend = PidBackend::new_update_loop( + temp.path().join("missing-codex.exe"), + state_dir.join("updater.pid"), + ); + let record = PidRecord { + pid: std::process::id(), + process_start_time: super::read_process_start_time(std::process::id()) + .await + .unwrap(), + }; + for record in [ + record.clone(), + PidRecord { + process_start_time: "stale".into(), + ..record + }, + ] { + tokio::fs::write(&backend.pid_file, serde_json::to_vec(&record).unwrap()) + .await + .unwrap(); + let error = backend.replace_current_updater().await.unwrap_err(); + // Windows may reject breakaway before reporting the missing executable. + // Both must reach process I/O rather than rejecting stale ownership. + assert!(error.root_cause().is::(), "{error:#}"); + assert_eq!( + backend.read_pid_file_state().await.unwrap(), + PidFileState::Running(record) + ); + } +} + +#[cfg(windows)] +#[tokio::test] +async fn updater_readiness_and_post_publication_failure_preserve_ownership() { + use futures::FutureExt; + use windows_sys::Win32::Security::ImpersonateAnonymousToken; + use windows_sys::Win32::Security::RevertToSelf; + use windows_sys::Win32::System::Threading::GetCurrentThread; + + let temp = TempDir::new().expect("temp"); + let backend = PidBackend::new_update_loop( + temp.path().join("codex.exe"), + temp.path().join("updater.pid"), + ); + let _lock = backend + .acquire_reservation_lock() + .await + .expect("reservation lock"); + let mut child = tokio::process::Command::new("powershell.exe") + .args([ + "-NoProfile", + "-NonInteractive", + "-Command", + "Start-Sleep 60", + ]) + .creation_flags(windows_sys::Win32::System::Threading::CREATE_NO_WINDOW) + .kill_on_drop(true) + .spawn() + .expect("successor fixture"); + let pid = child.id().expect("pid"); + let successor = PidRecord { + pid, + process_start_time: super::read_process_start_time(pid) + .await + .expect("creation time"), + }; + let predecessor = PidRecord { + pid: std::process::id(), + process_start_time: super::read_process_start_time(std::process::id()) + .await + .expect("creation time"), + }; + tokio::fs::write(&backend.pid_file, serde_json::to_vec(&successor).unwrap()) + .await + .unwrap(); + let ready = backend.pid_file.with_extension("ready"); + tokio::fs::write(&ready, b"").await.unwrap(); + backend + .finish_updater_start(&successor, Some(&predecessor)) + .await + .expect("ready successor"); + assert!(!ready.exists()); + assert_eq!( + backend.read_pid_file_state().await.unwrap(), + PidFileState::Running(successor.clone()) + ); + + // If cleanup cannot even query the successor, leave its ownership intact. + std::thread::scope(|scope| { + scope + .spawn(|| { + assert_ne!(unsafe { ImpersonateAnonymousToken(GetCurrentThread()) }, 0); + let result = backend + .finish_updater_start(&successor, Some(&predecessor)) + .now_or_never(); + let reverted = unsafe { RevertToSelf() }; + assert_ne!(reverted, 0); + let error = result + .expect("denied cleanup must not suspend") + .unwrap_err(); + assert_eq!( + error + .downcast_ref::() + .unwrap() + .raw_os_error(), + Some(windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED as i32) + ); + }) + .join() + .expect("anonymous cleanup check"); + }); + assert_eq!( + backend.read_pid_file_state().await.unwrap(), + PidFileState::Running(successor.clone()) + ); + + let reused_pid = PidRecord { + process_start_time: "stale".into(), + ..successor.clone() + }; + assert!( + backend + .finish_updater_start(&reused_pid, Some(&predecessor)) + .await + .is_err() + ); + assert_eq!( + backend.read_pid_file_state().await.unwrap(), + PidFileState::Running(predecessor.clone()) + ); + assert!( + child.try_wait().unwrap().is_none(), + "PID reuse must not terminate another process" + ); + tokio::fs::write(&backend.pid_file, serde_json::to_vec(&successor).unwrap()) + .await + .unwrap(); + // An invalid readiness path triggers cleanup while the successor is alive. + tokio::fs::create_dir(&ready).await.unwrap(); + assert!( + backend + .finish_updater_start(&successor, Some(&predecessor)) + .await + .is_err() + ); + assert!( + child.try_wait().unwrap().is_some(), + "successor must exit before rollback" + ); + assert_eq!( + backend.read_pid_file_state().await.unwrap(), + PidFileState::Running(predecessor) + ); +} + #[cfg(windows)] #[test] fn inaccessible_reused_pid_is_stale_without_hiding_process_open_errors() { diff --git a/codex-rs/app-server-daemon/src/backend/pid_windows.rs b/codex-rs/app-server-daemon/src/backend/pid_windows.rs new file mode 100644 index 0000000000..4bbf6da561 --- /dev/null +++ b/codex-rs/app-server-daemon/src/backend/pid_windows.rs @@ -0,0 +1,98 @@ +//! Windows updater publication handshake. The predecessor holds the operation +//! and reservation locks until its successor initializes and acknowledges readiness. + +use super::super::windows::Process; +use super::PidBackend; +use super::PidFileState; +use super::PidRecord; +use super::START_TIMEOUT; +use super::STOP_POLL_INTERVAL; +use super::process_matches_record; +use super::read_process_start_time; +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use tokio::fs; +use tokio::time::sleep; + +impl PidBackend { + pub(crate) async fn wait_for_ownership(&self) -> Result<()> { + let deadline = tokio::time::Instant::now() + START_TIMEOUT; + loop { + if let PidFileState::Running(record) = self.read_pid_file_state().await? + && record.pid == std::process::id() + && self.record_is_active(&record).await? + { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + bail!("updater was not published by its launcher"); + } + sleep(STOP_POLL_INTERVAL).await; + } + } + + pub(crate) async fn mark_ready(&self) -> Result<()> { + fs::write(self.pid_file.with_extension("ready"), b"") + .await + .context("failed to acknowledge updater startup") + } + + pub(crate) async fn replace_current_updater(&self) -> Result<()> { + let record = PidRecord { + pid: std::process::id(), + process_start_time: read_process_start_time(std::process::id()).await?, + }; + self.start_inner(Some(record)).await?; + Ok(()) + } + + pub(super) async fn finish_updater_start( + &self, + record: &PidRecord, + replacement: Option<&PidRecord>, + ) -> Result<()> { + let temp_pid_file = self.pid_file.with_extension("pid.tmp"); + let ready = self.pid_file.with_extension("ready"); + let deadline = tokio::time::Instant::now() + START_TIMEOUT; + let startup = async { + loop { + if !process_matches_record(record).await? { + bail!("updater exited before becoming ready"); + } + match fs::remove_file(&ready).await { + Ok(()) => return Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => return Err(err.into()), + } + if tokio::time::Instant::now() >= deadline { + bail!("updater did not become ready"); + } + sleep(STOP_POLL_INTERVAL).await; + } + } + .await; + if let Err(err) = startup { + if let Some(process) = Process::open(record.pid)? + && process.start_time()? == record.process_start_time + { + process.terminate()?; + let deadline = tokio::time::Instant::now() + START_TIMEOUT; + while process.is_running()? { + if tokio::time::Instant::now() >= deadline { + bail!("failed updater did not exit; ownership was not restored"); + } + sleep(STOP_POLL_INTERVAL).await; + } + } + if let Some(previous) = replacement { + fs::write(&temp_pid_file, serde_json::to_vec(previous)?).await?; + fs::rename(&temp_pid_file, &self.pid_file).await?; + } else { + let _ = fs::remove_file(&self.pid_file).await; + } + return Err(err); + } + Ok(()) + } +} diff --git a/codex-rs/app-server-daemon/src/backend/windows.rs b/codex-rs/app-server-daemon/src/backend/windows.rs index 178d8c5f58..f93e4bb20d 100644 --- a/codex-rs/app-server-daemon/src/backend/windows.rs +++ b/codex-rs/app-server-daemon/src/backend/windows.rs @@ -16,7 +16,15 @@ use windows_sys::Win32::Foundation::WAIT_TIMEOUT; use windows_sys::Win32::Storage::FileSystem::LOCKFILE_EXCLUSIVE_LOCK; use windows_sys::Win32::Storage::FileSystem::LOCKFILE_FAIL_IMMEDIATELY; use windows_sys::Win32::Storage::FileSystem::LockFileEx; +use windows_sys::Win32::System::JobObjects::AssignProcessToJobObject; +use windows_sys::Win32::System::JobObjects::CreateJobObjectW; use windows_sys::Win32::System::JobObjects::IsProcessInJob; +use windows_sys::Win32::System::JobObjects::JOB_OBJECT_LIMIT_BREAKAWAY_OK; +use windows_sys::Win32::System::JobObjects::JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; +use windows_sys::Win32::System::JobObjects::JOBOBJECT_EXTENDED_LIMIT_INFORMATION; +use windows_sys::Win32::System::JobObjects::JobObjectExtendedLimitInformation; +use windows_sys::Win32::System::JobObjects::SetInformationJobObject; +use windows_sys::Win32::System::Threading::GetCurrentProcess; use windows_sys::Win32::System::Threading::GetProcessId; use windows_sys::Win32::System::Threading::GetProcessTimes; use windows_sys::Win32::System::Threading::OpenProcess; @@ -156,6 +164,33 @@ pub(crate) fn try_lock_file(file: &tokio::fs::File) -> Result { Err(err).context("failed to lock daemon state") } +// Keep installer descendants bounded by the updater's lifetime, while allowing +// app-server launches and successor updaters to break away from this job. +pub(crate) fn updater_job() -> Result { + let job = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) }; + if job == 0 { + return Err(io::Error::last_os_error()).context("failed to create updater job"); + } + let owned = unsafe { OwnedHandle::from_raw_handle(job as _) }; + let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() }; + limits.BasicLimitInformation.LimitFlags = + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE | JOB_OBJECT_LIMIT_BREAKAWAY_OK; + if unsafe { + SetInformationJobObject( + job, + JobObjectExtendedLimitInformation, + (&limits as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(), + std::mem::size_of_val(&limits) as u32, + ) + } == 0 + || unsafe { AssignProcessToJobObject(job, GetCurrentProcess()) } == 0 + { + return Err(io::Error::last_os_error()) + .context("failed to contain updater installer processes"); + } + Ok(owned) +} + #[cfg(test)] #[path = "windows_tests.rs"] mod tests; diff --git a/codex-rs/app-server-daemon/src/lib.rs b/codex-rs/app-server-daemon/src/lib.rs index 0bd4a97408..93d5e8b067 100644 --- a/codex-rs/app-server-daemon/src/lib.rs +++ b/codex-rs/app-server-daemon/src/lib.rs @@ -23,7 +23,7 @@ use codex_app_server_protocol::RemoteControlPairingStartResponse; use codex_app_server_transport::app_server_control_socket_path; use codex_utils_home_dir::find_codex_home; use managed_install::managed_codex_bin; -#[cfg(unix)] +#[cfg(any(unix, windows))] use managed_install::managed_codex_version; use serde::Serialize; use settings::DaemonSettings; @@ -160,7 +160,7 @@ pub struct RemoteControlOutput { pub app_server_version: Option, } -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RestartIfRunningOutcome { Busy, @@ -170,21 +170,21 @@ pub(crate) enum RestartIfRunningOutcome { Restarted, } -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RestartMode { IfVersionChanged, Always, } -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum UpdaterRefreshMode { None, ReexecIfManagedBinaryChanged, } -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum RestartDecision { NotReady, @@ -363,7 +363,7 @@ impl Daemon { .await) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) async fn try_restart_if_running( &self, mode: RestartMode, @@ -402,9 +402,18 @@ impl Daemon { RestartIfRunningOutcome::NotRunning }; + #[cfg(unix)] if should_reexec_updater(updater_refresh_mode, outcome) { crate::update_loop::reexec_managed_updater(managed_codex_bin)?; } + #[cfg(windows)] + if should_reexec_updater(updater_refresh_mode, outcome) { + backend::pid_update_loop_backend( + self.backend_paths_with_bin(&settings, managed_codex_bin), + ) + .replace_current_updater() + .await?; + } Ok(outcome) } @@ -683,12 +692,12 @@ impl Daemon { )) } - #[cfg(unix)] + #[cfg(any(unix, windows))] async fn managed_codex_version_best_effort(&self) -> Option { managed_codex_version(&self.managed_codex_bin).await.ok() } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] async fn managed_codex_version_best_effort(&self) -> Option { None } @@ -810,7 +819,7 @@ fn already_remote_control_status(mode: RemoteControlMode) -> RemoteControlStatus } } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn restart_decision( mode: RestartMode, info: Option<&client::ProbeInfo>, @@ -827,7 +836,7 @@ fn restart_decision( } } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn should_reexec_updater( updater_refresh_mode: UpdaterRefreshMode, outcome: RestartIfRunningOutcome, @@ -857,7 +866,7 @@ fn try_lock_file(_file: &tokio::fs::File) -> Result { Ok(true) } -#[cfg(all(test, unix))] +#[cfg(all(test, any(unix, windows)))] mod tests { use pretty_assertions::assert_eq; use tempfile::TempDir; diff --git a/codex-rs/app-server-daemon/src/managed_install.rs b/codex-rs/app-server-daemon/src/managed_install.rs index 10f2c9c61c..c0a19ac8a0 100644 --- a/codex-rs/app-server-daemon/src/managed_install.rs +++ b/codex-rs/app-server-daemon/src/managed_install.rs @@ -1,21 +1,14 @@ -//! Resolves packaged and legacy standalone install layouts. +//! Resolves both package and legacy standalone layouts and compares installed executables. use std::path::Path; use std::path::PathBuf; -#[cfg(unix)] use anyhow::Context; -#[cfg(unix)] use anyhow::Result; -#[cfg(unix)] use anyhow::anyhow; -#[cfg(unix)] use sha2::Digest; -#[cfg(unix)] use sha2::Sha256; -#[cfg(unix)] use tokio::fs; -#[cfg(unix)] use tokio::process::Command; /// Returns the packaged executable when present, otherwise an existing legacy executable. @@ -37,7 +30,6 @@ pub(crate) fn managed_codex_bin(codex_home: &Path) -> PathBuf { } } -#[cfg(unix)] pub(crate) async fn resolved_managed_codex_bin(codex_bin: &Path) -> Result { fs::canonicalize(codex_bin).await.with_context(|| { format!( @@ -47,18 +39,16 @@ pub(crate) async fn resolved_managed_codex_bin(codex_bin: &Path) -> Result Result { - let output = Command::new(codex_bin) - .arg("--version") - .output() - .await - .with_context(|| { - format!( - "failed to invoke managed Codex binary {}", - codex_bin.display() - ) - })?; + let mut command = Command::new(codex_bin); + #[cfg(windows)] + command.creation_flags(windows_sys::Win32::System::Threading::CREATE_NO_WINDOW); + let output = command.arg("--version").output().await.with_context(|| { + format!( + "failed to invoke managed Codex binary {}", + codex_bin.display() + ) + })?; if !output.status.success() { return Err(anyhow!( "managed Codex binary {} exited with status {}", @@ -76,13 +66,11 @@ pub(crate) async fn managed_codex_version(codex_bin: &Path) -> Result { parse_codex_version(&stdout) } -#[cfg(unix)] #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct ExecutableIdentity { digest: [u8; 32], } -#[cfg(unix)] pub(crate) async fn executable_identity(executable: &Path) -> Result { let bytes = fs::read(executable) .await @@ -90,7 +78,6 @@ pub(crate) async fn executable_identity(executable: &Path) -> Result ExecutableIdentity { ExecutableIdentity { digest: Sha256::digest(bytes).into(), @@ -101,7 +88,6 @@ fn managed_codex_file_name() -> &'static str { if cfg!(windows) { "codex.exe" } else { "codex" } } -#[cfg(unix)] fn parse_codex_version(output: &str) -> Result { let version = output .split_whitespace() @@ -111,7 +97,7 @@ fn parse_codex_version(output: &str) -> Result { Ok(version.to_string()) } -#[cfg(all(test, unix))] +#[cfg(test)] #[path = "managed_install_tests.rs"] mod tests; diff --git a/codex-rs/app-server-daemon/src/update_loop.rs b/codex-rs/app-server-daemon/src/update_loop.rs index 76fd94837d..06c9b01529 100644 --- a/codex-rs/app-server-daemon/src/update_loop.rs +++ b/codex-rs/app-server-daemon/src/update_loop.rs @@ -1,27 +1,19 @@ -#[cfg(unix)] -use std::process::Command as StdCommand; -#[cfg(unix)] -use std::process::Stdio; -#[cfg(unix)] -use std::time::Duration; +//! Installs updates, validates the server restart, then transfers updater ownership. #[cfg(unix)] +use std::process::Command as StdCommand; +use std::process::Stdio; +use std::time::Duration; + use anyhow::Context; use anyhow::Result; -#[cfg(not(unix))] -use anyhow::bail; -#[cfg(unix)] use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; -#[cfg(unix)] use codex_http_client::RouteAwareClientPool; -#[cfg(unix)] use futures::FutureExt; #[cfg(unix)] use std::os::unix::process::CommandExt; -#[cfg(unix)] use tokio::io::AsyncWriteExt; -#[cfg(unix)] use tokio::process::Command; #[cfg(unix)] use tokio::signal::unix::Signal; @@ -29,38 +21,44 @@ use tokio::signal::unix::Signal; use tokio::signal::unix::SignalKind; #[cfg(unix)] use tokio::signal::unix::signal; -#[cfg(unix)] use tokio::time::sleep; -#[cfg(unix)] use crate::Daemon; -#[cfg(unix)] use crate::RestartIfRunningOutcome; -#[cfg(unix)] use crate::RestartMode; -#[cfg(unix)] use crate::UpdaterRefreshMode; -#[cfg(unix)] use crate::managed_install::ExecutableIdentity; -#[cfg(unix)] use crate::managed_install::executable_identity; -#[cfg(unix)] use crate::managed_install::resolved_managed_codex_bin; -#[cfg(unix)] const INITIAL_UPDATE_DELAY: Duration = Duration::from_secs(5 * 60); -#[cfg(unix)] const RESTART_RETRY_INTERVAL: Duration = Duration::from_millis(50); -#[cfg(unix)] const UPDATE_INTERVAL: Duration = Duration::from_secs(60 * 60); #[cfg(unix)] const INSTALL_URL: &str = "https://chatgpt.com/codex/install.sh"; +#[cfg(windows)] +const INSTALL_URL: &str = "https://chatgpt.com/codex/install.ps1"; -#[cfg(unix)] pub(crate) async fn run(http_client_factory: HttpClientFactory) -> Result<()> { + #[cfg(unix)] let mut terminate = signal(SignalKind::terminate()).context("failed to install updater shutdown handler")?; + #[cfg(windows)] + let updater = { + let daemon = Daemon::from_environment()?; + crate::backend::pid_update_loop_backend( + daemon.backend_paths(&daemon.load_settings().await?), + ) + }; + #[cfg(windows)] + updater.wait_for_ownership().await?; + #[cfg(windows)] + let mut terminate = Signal; + #[cfg(windows)] + let _installer_job = crate::backend::windows::updater_job()?; let running_updater_identity = current_updater_identity().await?; + #[cfg(windows)] + updater.mark_ready().await?; let http = RouteAwareClientPool::new_without_request_logging( http_client_factory, ClientRouteClass::Other, @@ -69,6 +67,10 @@ pub(crate) async fn run(http_client_factory: HttpClientFactory) -> Result<()> { return Ok(()); } loop { + // Failed successor cleanup leaves its PID published. The predecessor + // must stop instead of installing again without ownership. + #[cfg(windows)] + updater.wait_for_ownership().await?; match update_once(&http, &running_updater_identity, &mut terminate).await { Ok(UpdateLoopControl::Continue) | Err(_) => {} Ok(UpdateLoopControl::Stop) => return Ok(()), @@ -79,12 +81,6 @@ pub(crate) async fn run(http_client_factory: HttpClientFactory) -> Result<()> { } } -#[cfg(not(unix))] -pub(crate) async fn run(_http_client_factory: HttpClientFactory) -> Result<()> { - bail!("pid-managed updater loop is unsupported on this platform") -} - -#[cfg(unix)] async fn sleep_or_terminate(duration: Duration, terminate: &mut Signal) -> bool { tokio::select! { _ = sleep(duration) => false, @@ -92,19 +88,23 @@ async fn sleep_or_terminate(duration: Duration, terminate: &mut Signal) -> bool } } -#[cfg(unix)] enum UpdateLoopControl { Continue, Stop, } -#[cfg(unix)] async fn update_once( http: &RouteAwareClientPool, running_updater_identity: &ExecutableIdentity, terminate: &mut Signal, ) -> Result { + #[cfg(unix)] install_latest_standalone(http).await?; + #[cfg(windows)] + tokio::select! { + result = install_latest_standalone(http) => result?, + _ = terminate.recv() => return Ok(UpdateLoopControl::Stop), + } let daemon = Daemon::from_environment()?; let managed_codex_bin = resolved_managed_codex_bin(&daemon.managed_codex_bin).await?; @@ -125,19 +125,26 @@ async fn update_once( return Ok(UpdateLoopControl::Stop); } } - _ => return Ok(UpdateLoopControl::Continue), + RestartIfRunningOutcome::Restarted => { + #[cfg(windows)] + if updater_refresh_mode == UpdaterRefreshMode::ReexecIfManagedBinaryChanged { + return Ok(UpdateLoopControl::Stop); + } + return Ok(UpdateLoopControl::Continue); + } + RestartIfRunningOutcome::NotRunning + | RestartIfRunningOutcome::NotReady + | RestartIfRunningOutcome::AlreadyCurrent => return Ok(UpdateLoopControl::Continue), } } } -#[cfg(unix)] async fn current_updater_identity() -> Result { let current_exe = std::env::current_exe().context("failed to resolve current updater executable")?; executable_identity(¤t_exe).await } -#[cfg(unix)] fn update_modes_for_identities( running_updater_identity: &ExecutableIdentity, managed_identity: &ExecutableIdentity, @@ -165,12 +172,25 @@ pub(crate) fn reexec_managed_updater(managed_codex_bin: &std::path::Path) -> Res }) } -#[cfg(unix)] -async fn install_latest_standalone(http: &RouteAwareClientPool) -> Result<()> { +async fn install_latest_standalone(http: &impl InstallerHttp) -> Result<()> { let script = fetch_installer_script(http).await?; - let mut child = Command::new("/bin/sh") - .arg("-s") + #[cfg(unix)] + let mut command = { + let mut command = Command::new("/bin/sh"); + command.arg("-s"); + command + }; + #[cfg(windows)] + let mut command = { + let mut command = Command::new("powershell.exe"); + command.creation_flags(windows_sys::Win32::System::Threading::CREATE_NO_WINDOW); + command.args(["-NoLogo", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-Command", "try { Invoke-Expression ([Console]::In.ReadToEnd()) } catch { Write-Error $_; exit 1 }"]) + .env("CODEX_NON_INTERACTIVE", "1") + .kill_on_drop(true); + command + }; + let mut child = command .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(Stdio::null()) @@ -197,7 +217,6 @@ async fn install_latest_standalone(http: &RouteAwareClientPool) -> Result<()> { } } -#[cfg(unix)] async fn fetch_installer_script(http: &impl InstallerHttp) -> Result> { match http.get(INSTALL_URL).await? { InstallerResponse::Success(body) => Ok(body), @@ -207,14 +226,12 @@ async fn fetch_installer_script(http: &impl InstallerHttp) -> Result> { } } -#[cfg(unix)] #[derive(Clone, Debug, PartialEq, Eq)] enum InstallerResponse { Success(Vec), Unsuccessful { status: u16 }, } -#[cfg(unix)] /// HTTP boundary used to download the standalone installer. /// /// Implementations must issue a GET for the supplied URL, return exact response bytes for a @@ -226,7 +243,6 @@ trait InstallerHttp: Send + Sync { ) -> impl std::future::Future> + Send + 'a; } -#[cfg(unix)] impl InstallerHttp for RouteAwareClientPool { async fn get(&self, url: &str) -> Result { let response = RouteAwareClientPool::get(self, url) @@ -247,6 +263,18 @@ impl InstallerHttp for RouteAwareClientPool { } } -#[cfg(all(test, unix))] +#[cfg(test)] #[path = "update_loop_tests.rs"] mod tests; + +#[cfg(windows)] +struct Signal; + +#[cfg(windows)] +impl Signal { + async fn recv(&mut self) -> Option<()> { + // An unreadable control path must stop the updater rather than disable shutdown. + let _ = codex_app_server_transport::daemon_shutdown_signal().await; + Some(()) + } +} diff --git a/codex-rs/app-server-daemon/src/update_loop_tests.rs b/codex-rs/app-server-daemon/src/update_loop_tests.rs index 8c94f91202..bb2a87f028 100644 --- a/codex-rs/app-server-daemon/src/update_loop_tests.rs +++ b/codex-rs/app-server-daemon/src/update_loop_tests.rs @@ -92,3 +92,24 @@ impl InstallerHttp for FakeInstallerHttp { Ok(self.response.clone()) } } + +#[cfg(windows)] +#[tokio::test] +async fn powershell_installer_is_noninteractive_and_reports_script_failure() { + let valid = FakeInstallerHttp::new(InstallerResponse::Success( + br#" +function Test-Installer { + if ($env:CODEX_NON_INTERACTIVE -ne '1') { throw 'interactive installer' } +} +Test-Installer +"# + .to_vec(), + )); + super::install_latest_standalone(&valid) + .await + .expect("installer succeeds"); + let failing = FakeInstallerHttp::new(InstallerResponse::Success( + b"throw 'installer failed'".to_vec(), + )); + assert!(super::install_latest_standalone(&failing).await.is_err()); +}