From 6db53df37f4e87cbf4a01888168c11c4d356f199 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Fri, 7 Aug 2026 20:00:24 +0000 Subject: [PATCH] Preserve child waiters during process termination (#37498) ## Why Aborting the child waiter during termination can leave an exited PTY child unreaped and prevent the session from recording its exit status. ## What changed - Detach the child waiter instead of aborting it when terminating or dropping a `ProcessHandle`, while continuing to abort the I/O helper tasks. - Cover explicit termination and handle drops for pipe processes, plus queued PTY waiters and PTY process-group termination on Unix. GitOrigin-RevId: 313787f4edec80b30cf5a78f0cc6733f1520d9a4 --- codex-rs/utils/pty/src/process.rs | 5 +- codex-rs/utils/pty/src/tests.rs | 142 +++++++++++++++++++++++++++++- 2 files changed, 144 insertions(+), 3 deletions(-) diff --git a/codex-rs/utils/pty/src/process.rs b/codex-rs/utils/pty/src/process.rs index c4067eedb7..5593450b89 100644 --- a/codex-rs/utils/pty/src/process.rs +++ b/codex-rs/utils/pty/src/process.rs @@ -242,7 +242,7 @@ impl ProcessHandle { result } - /// Attempts to kill the child and abort helper tasks. + /// Attempts to kill the child and abort I/O helper tasks. pub fn terminate(&self) { self.request_terminate(); @@ -264,7 +264,8 @@ impl ProcessHandle { if let Ok(mut h) = self.wait_handle.lock() && let Some(handle) = h.take() { - handle.abort(); + // Even a queued PTY waiter must run to reap the terminated child. + drop(handle); } } } diff --git a/codex-rs/utils/pty/src/tests.rs b/codex-rs/utils/pty/src/tests.rs index 98e255c613..8a7ddbebfa 100644 --- a/codex-rs/utils/pty/src/tests.rs +++ b/codex-rs/utils/pty/src/tests.rs @@ -827,6 +827,140 @@ async fn pipe_terminate_aborts_detached_readers() -> anyhow::Result<()> { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pipe_terminate_reaps_child() -> anyhow::Result<()> { + let env_map: HashMap = std::env::vars().collect(); + let command = if cfg!(windows) { + "ping -n 60 127.0.0.1 > NUL" + } else { + "sleep 60" + }; + let (program, args) = shell_command(command); + let spawned = spawn_pipe_process(&program, &args, Path::new("."), &env_map, &None, &[]).await?; + let (session, _output_rx, exit_rx) = combine_spawned_output(spawned); + + session.terminate(); + + let exit_code = tokio::time::timeout(tokio::time::Duration::from_secs(5), exit_rx) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for terminated child to be reaped"))? + .map_err(|_| anyhow::anyhow!("child waiter was aborted before reaping"))?; + assert_eq!(session.exit_code(), Some(exit_code)); + assert!(session.has_exited()); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pipe_drop_reaps_child() -> anyhow::Result<()> { + let env_map: HashMap = std::env::vars().collect(); + let command = if cfg!(windows) { + "ping -n 60 127.0.0.1 > NUL" + } else { + "sleep 60" + }; + let (program, args) = shell_command(command); + let spawned = spawn_pipe_process(&program, &args, Path::new("."), &env_map, &None, &[]).await?; + let (session, _output_rx, exit_rx) = combine_spawned_output(spawned); + + drop(session); + + tokio::time::timeout(tokio::time::Duration::from_secs(5), exit_rx) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for dropped child to be reaped"))? + .map_err(|_| anyhow::anyhow!("child waiter was aborted before reaping"))?; + + Ok(()) +} + +#[cfg(unix)] +#[test] +fn pty_terminate_reaps_child_when_waiter_is_queued() -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build()?; + let (blocker_started_tx, blocker_started_rx) = std::sync::mpsc::channel(); + let (release_blocker_tx, release_blocker_rx) = std::sync::mpsc::channel(); + // Keep the PTY's blocking child waiter queued until after termination. + runtime.spawn_blocking(move || { + let _ = blocker_started_tx.send(()); + let _ = release_blocker_rx.recv_timeout(std::time::Duration::from_secs(5)); + }); + blocker_started_rx.recv_timeout(std::time::Duration::from_secs(2))?; + + let pid_file = std::env::temp_dir().join(format!( + "codex-pty-reap-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_nanos() + )); + runtime.block_on(async { + let mut env_map: HashMap = std::env::vars().collect(); + env_map.insert( + "CODEX_PTY_TEST_PID_FILE".to_string(), + pid_file.display().to_string(), + ); + let (program, args) = + shell_command("printf '%s' \"$$\" > \"$CODEX_PTY_TEST_PID_FILE\"; sleep 60"); + let spawned = spawn_pty_process( + &program, + &args, + Path::new("."), + &env_map, + &None, + TerminalSize::default(), + &[], + ) + .await?; + let (session, _output_rx, exit_rx) = combine_spawned_output(spawned); + + let child_pid = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + if let Ok(pid) = std::fs::read_to_string(&pid_file) + && let Ok(pid) = pid.parse::() + { + return pid; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for the PTY child PID"))?; + std::fs::remove_file(&pid_file)?; + + session.terminate(); + drop(session); + release_blocker_tx.send(())?; + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), exit_rx) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for the PTY child waiter"))?; + + // A returned PID proves an exited child was still an unreaped zombie. + let wait_result = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let result = + unsafe { libc::waitpid(child_pid, std::ptr::null_mut(), libc::WNOHANG) }; + if result != 0 { + return result; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for the PTY child to exit"))?; + let wait_error = std::io::Error::last_os_error(); + assert_eq!( + wait_result, -1, + "PTY child {child_pid} remained an unreaped zombie" + ); + assert_eq!(wait_error.raw_os_error(), Some(libc::ECHILD)); + + Ok(()) + }) +} + #[cfg(unix)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn pty_terminate_kills_background_children_in_same_process_group() -> anyhow::Result<()> { @@ -844,7 +978,7 @@ async fn pty_terminate_kills_background_children_in_same_process_group() -> anyh &[], ) .await?; - let (session, mut output_rx, _exit_rx) = combine_spawned_output(spawned); + let (session, mut output_rx, exit_rx) = combine_spawned_output(spawned); let bg_pid = match wait_for_marker_pid(&mut output_rx, marker, /*timeout_ms*/ 2_000).await { Ok(pid) => pid, @@ -860,6 +994,12 @@ async fn pty_terminate_kills_background_children_in_same_process_group() -> anyh session.terminate(); + tokio::time::timeout(tokio::time::Duration::from_secs(3), exit_rx) + .await + .map_err(|_| anyhow::anyhow!("timed out waiting for terminated PTY child to be reaped"))? + .map_err(|_| anyhow::anyhow!("PTY child waiter was aborted before reaping"))?; + assert!(session.has_exited()); + let exited = wait_for_process_exit(bg_pid, /*timeout_ms*/ 3_000).await?; if !exited { let _ = unsafe { libc::kill(bg_pid, libc::SIGKILL) };