Release empty PID probe locks explicitly on Unix (#57533)

## Why

A concurrent fork can inherit a PID reservation probe's file descriptor and keep its lock held after the probe closes it, making a stale reservation appear active.

## What changed

Add a drop guard that explicitly unlocks empty PID reservation probes on Unix, including when inspection fails or is cancelled.

## Testing

Extend the duplicated-handle regression test to cover empty PID files, read errors, and cancellation alongside missing PID files. Verify that cancellation occurs while the lock is held and that subsequent status checks do not report an active reservation.

GitOrigin-RevId: 793569c9bcaca02900662a66c33aa48f2bebfd5e

codex-rs/app-server-daemon/src/backend/pid.rs

@@ -1,5 +1,5 @@
 //! PID reservations serialize detached launches; process identities protect stale-record cleanup.
-//! Missing-PID probes explicitly unlock so concurrent forks cannot retain their transient reservations.
+//! PID probes explicitly unlock so concurrent forks cannot retain their transient reservations.
 
 #[path = "pid_identity.rs"]
 mod identity;
@@ -750,9 +750,36 @@ async fn inspect_empty_pid_reservation(
             });
         }
     };
+    inspect_empty_pid_reservation_with_file(pid_path, file).await
+}
+
+#[cfg(unix)]
+struct EmptyPidProbeLock(fs::File);
+
+#[cfg(unix)]
+impl Drop for EmptyPidProbeLock {
+    fn drop(&mut self) {
+        use std::os::fd::AsRawFd;
+
+        // Closing alone leaves the lock held by any concurrent fork. Release it
+        // on every exit, including errors and cancellation while inspecting the PID.
+        if unsafe { libc::flock(self.0.as_raw_fd(), libc::LOCK_UN) } != 0 {
+            let error = std::io::Error::last_os_error();
+            tracing::warn!(%error, "failed to unlock empty pid reservation probe");
+        }
+    }
+}
+
+#[cfg(any(unix, windows))]
+async fn inspect_empty_pid_reservation_with_file(
+    pid_path: &Path,
+    file: fs::File,
+) -> Result<EmptyPidReservation> {
     if !try_lock_file(&file)? {
         return Ok(EmptyPidReservation::Active);
     }
+    #[cfg(unix)]
+    let _probe_lock = EmptyPidProbeLock(file);
 
     let contents = match fs::read_to_string(pid_path).await {
         Ok(contents) => contents,

codex-rs/app-server-daemon/src/backend/pid_tests.rs

@@ -101,35 +101,112 @@ async fn unlocked_empty_pid_file_is_treated_as_stale_reservation() {
 }
 
 #[cfg(unix)]
-#[tokio::test]
-async fn reservation_probe_releases_lock_with_duplicated_handle() {
-    let temp_dir = TempDir::new().expect("temp dir");
-    let backend = PidBackend::new(
-        temp_dir.path().join("codex"),
-        temp_dir.path().join("app-server.pid"),
-        /*remote_control_enabled*/ false,
-    );
-    let file = std::fs::OpenOptions::new()
-        .create(true)
-        .truncate(false)
-        .write(true)
-        .open(&backend.lock_file)
-        .expect("open pid lock file");
-    // Share the open file description as fork would, without inheriting
-    // unrelated descriptors from other tests.
-    let inherited = file.try_clone().expect("clone pid lock file");
+#[test]
+fn reservation_probe_releases_lock_with_duplicated_handle() {
+    #[derive(Debug, Clone, Copy)]
+    enum Probe {
+        Missing,
+        Empty,
+        ReadError,
+        Cancelled,
+    }
 
-    let probe = super::probe_reservation_lock(tokio::fs::File::from_std(file));
-    let running = backend.is_starting_or_running().await;
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .max_blocking_threads(/*val*/ 1)
+        .build()
+        .expect("runtime");
+    runtime.block_on(async {
+        for case in [
+            Probe::Missing,
+            Probe::Empty,
+            Probe::ReadError,
+            Probe::Cancelled,
+        ] {
+            let temp_dir = TempDir::new().expect("temp dir");
+            let backend = PidBackend::new(
+                temp_dir.path().join("codex"),
+                temp_dir.path().join("app-server.pid"),
+                /*remote_control_enabled*/ false,
+            );
+            match case {
+                Probe::Missing => {}
+                Probe::Empty | Probe::Cancelled => {
+                    std::fs::write(&backend.pid_file, "").expect("empty pid file");
+                }
+                Probe::ReadError => {
+                    std::fs::create_dir(&backend.pid_file).expect("unreadable pid file");
+                }
+            }
+            let file = std::fs::OpenOptions::new()
+                .create(true)
+                .truncate(false)
+                .write(true)
+                .open(&backend.lock_file)
+                .expect("open pid lock file");
+            // Share the open file description as fork would, without inheriting
+            // unrelated descriptors from other tests.
+            let inherited = file.try_clone().expect("clone pid lock file");
 
-    drop(inherited);
-    assert_eq!(
-        (
-            probe.expect("probe reservation"),
-            running.expect("read daemon status"),
-        ),
-        (false, false),
-    );
+            let file = tokio::fs::File::from_std(file);
+            let mut cancellation = None;
+            let probe = match case {
+                Probe::Missing => super::probe_reservation_lock(file),
+                Probe::Empty | Probe::ReadError => {
+                    super::inspect_empty_pid_reservation_with_file(&backend.pid_file, file)
+                        .await
+                        .map(|state| state != super::EmptyPidReservation::Stale)
+                }
+                Probe::Cancelled => {
+                    // Occupy the sole blocking worker so the PID read cannot finish
+                    // before cancellation. No timing assumptions or sleeps are needed.
+                    let (started_tx, started_rx) = std::sync::mpsc::channel();
+                    let (release_tx, release_rx) = std::sync::mpsc::channel();
+                    let blocker = tokio::task::spawn_blocking(move || {
+                        started_tx.send(()).expect("worker started");
+                        release_rx.recv().expect("release worker");
+                    });
+                    started_rx.recv().expect("wait for worker");
+                    let mut inspection = Box::pin(super::inspect_empty_pid_reservation_with_file(
+                        &backend.pid_file,
+                        file,
+                    ));
+                    let pending = futures::poll!(&mut inspection).is_pending();
+                    let contender = tokio::fs::File::from_std(
+                        std::fs::OpenOptions::new()
+                            .write(true)
+                            .open(&backend.lock_file)
+                            .expect("open competing reservation"),
+                    );
+                    cancellation = Some((pending, try_lock_file(&contender)));
+                    drop(contender);
+                    drop(inspection);
+                    release_tx.send(()).expect("unblock PID read");
+                    blocker.await.expect("blocking worker");
+                    Ok(false)
+                }
+            };
+            if matches!(case, Probe::ReadError) {
+                std::fs::remove_dir(&backend.pid_file).expect("remove unreadable pid file");
+            }
+            let running = backend.is_starting_or_running().await;
+
+            drop(inherited);
+            match case {
+                Probe::ReadError => assert!(probe.is_err(), "{case:?}"),
+                Probe::Missing | Probe::Empty | Probe::Cancelled => {
+                    assert!(!probe.expect("probe reservation"), "{case:?}");
+                }
+            }
+            if let Some((pending, acquired)) = cancellation {
+                assert_eq!(
+                    (pending, acquired.expect("competing reservation")),
+                    (true, false)
+                );
+            }
+            assert!(!running.expect("read daemon status"), "{case:?}");
+        }
+    });
 }
 
 #[tokio::test]