Support managed daemon updates on Windows (#42392)

## What changed

- Run the daemon update loop on Windows using the non-interactive PowerShell installer.
- Restart the managed app server after updates and hand updater ownership to a changed managed executable through a readiness handshake.
- Contain installer descendants in a Windows job while allowing app-server and successor updater processes to break away.
- Preserve or restore PID ownership when successor startup fails, without terminating a process that reused the PID.

## Testing

- Cover non-interactive installer success and script failure on Windows.
- Cover updater handoff, readiness, rollback, inaccessible processes, and PID reuse.

GitOrigin-RevId: 5178c01124b4dd86b507b53f5093ec9a9d48e12c

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<Option<u32>> {
-        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;

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<Option<u32>> {
+    pub(super) async fn start_inner(&self, replacement: Option<PidRecord>) -> Result<Option<u32>> {
         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))
     }

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::<std::io::Error>(), "{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::<std::io::Error>()
+                        .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() {

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(())
+    }
+}

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<bool> {
     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<OwnedHandle> {
+    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;

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<String>,
 }
 
-#[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<String> {
         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<String> {
         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<bool> {
     Ok(true)
 }
 
-#[cfg(all(test, unix))]
+#[cfg(all(test, any(unix, windows)))]
 mod tests {
     use pretty_assertions::assert_eq;
     use tempfile::TempDir;

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<PathBuf> {
     fs::canonicalize(codex_bin).await.with_context(|| {
         format!(
@@ -47,18 +39,16 @@ pub(crate) async fn resolved_managed_codex_bin(codex_bin: &Path) -> Result<PathB
     })
 }
 
-#[cfg(unix)]
 pub(crate) async fn managed_codex_version(codex_bin: &Path) -> Result<String> {
-    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<String> {
     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<ExecutableIdentity> {
     let bytes = fs::read(executable)
         .await
@@ -90,7 +78,6 @@ pub(crate) async fn executable_identity(executable: &Path) -> Result<ExecutableI
     Ok(executable_identity_from_bytes(&bytes))
 }
 
-#[cfg(unix)]
 pub(crate) fn executable_identity_from_bytes(bytes: &[u8]) -> 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<String> {
     let version = output
         .split_whitespace()
@@ -111,7 +97,7 @@ fn parse_codex_version(output: &str) -> Result<String> {
     Ok(version.to_string())
 }
 
-#[cfg(all(test, unix))]
+#[cfg(test)]
 #[path = "managed_install_tests.rs"]
 mod tests;
 

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<UpdateLoopControl> {
+    #[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<ExecutableIdentity> {
     let current_exe =
         std::env::current_exe().context("failed to resolve current updater executable")?;
     executable_identity(&current_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<Vec<u8>> {
     match http.get(INSTALL_URL).await? {
         InstallerResponse::Success(body) => Ok(body),
@@ -207,14 +226,12 @@ async fn fetch_installer_script(http: &impl InstallerHttp) -> Result<Vec<u8>> {
     }
 }
 
-#[cfg(unix)]
 #[derive(Clone, Debug, PartialEq, Eq)]
 enum InstallerResponse {
     Success(Vec<u8>),
     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<Output = Result<InstallerResponse>> + Send + 'a;
 }
 
-#[cfg(unix)]
 impl InstallerHttp for RouteAwareClientPool {
     async fn get(&self, url: &str) -> Result<InstallerResponse> {
         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(())
+    }
+}

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());
+}