Expand Linux spawn-helper lifecycle test coverage (#47613)

## What changed

Add real-child tests for parent-death signaling, large environment transfers, explicit file descriptor inheritance, and fallback when the helper exits before acknowledgement.

Introduce test-only pause points during environment transfer, reporting, and exec to verify that cancelling startup kills and reaps the helper. Cover cancellation after runtime shutdown and verify that the fallback reaper can reap an exited child while an earlier child remains alive.

Probe an actual helper launch before asserting native-helper behavior, and skip fixtures when their required procfs or pidfd support is unavailable.

GitOrigin-RevId: 5136dc242a91efd0596171266965f29903c6c970

codex-rs/utils/pty/src/spawn_helper_main.rs

@@ -62,6 +62,8 @@ pub(super) fn dispatch(mut args: impl Iterator<Item = std::ffi::OsString>) -> !
         // Allocation is safe in this fresh, single-threaded image. CLOEXEC
         // leaves the report socket open until the target actually execs.
         crate::pty::close_inherited_fds_except(&inherited_fds);
+        #[cfg(test)]
+        crate::spawn_helper_tests::pause_handshake("environment")?;
         let mut size = [0; 4];
         control.read_exact(&mut size)?;
         let size = u32::from_le_bytes(size) as usize;
@@ -83,8 +85,12 @@ pub(super) fn dispatch(mut args: impl Iterator<Item = std::ffi::OsString>) -> !
                 std::ffi::OsStr::from_bytes(&entry[separator + 1..]),
             );
         }
+        #[cfg(test)]
+        crate::spawn_helper_tests::pause_handshake("report")?;
         control.write_all(&[REPORT_PREFIX])?;
         reported = true;
+        #[cfg(test)]
+        crate::spawn_helper_tests::pause_handshake("exec")?;
         Err(command.exec())
     })();
     if !reported {

codex-rs/utils/pty/src/spawn_helper_tests.rs

@@ -1,14 +1,24 @@
-//! Verify Linux helper dispatch, launch errors, and target environment isolation.
+//! Exercise the exec-helper boundary with real children, including descriptors,
+//! startup errors, cancellation cleanup, and concurrent exit notification.
+//! Procfs-dependent fixtures are skipped when procfs is unavailable.
 
 use std::collections::HashMap;
 use std::io;
+use std::os::fd::AsFd;
+use std::os::fd::AsRawFd;
+use std::os::fd::FromRawFd;
+use std::os::fd::OwnedFd;
 use std::path::Path;
+use std::time::Duration;
 
 use pretty_assertions::assert_eq;
 
 use crate::spawn_pipe_process;
 use crate::spawn_pipe_process_no_stdin;
 
+const PARENT_FIXTURE: &str = "--codex-spawn-test-parent";
+const CHILD_FIXTURE: &str = "--codex-spawn-test-child";
+
 #[ctor::ctor]
 fn initialize_spawn_helper() {
     if let Ok(arguments) = std::fs::read("/proc/self/cmdline") {
@@ -31,6 +41,148 @@ fn initialize_spawn_helper() {
     } else {
         crate::init_spawn_helper(std::env::args_os());
     }
+    let Ok(arguments) = process_arguments() else {
+        return;
+    };
+    let mut arguments = arguments.into_iter().skip(1);
+    let argument = arguments.next();
+    if argument.as_deref() == Some(std::ffi::OsStr::new(CHILD_FIXTURE)) {
+        let Some(cleanup_fd) = arguments
+            .next()
+            .and_then(|argument| argument.to_str()?.parse().ok())
+        else {
+            std::process::exit(1);
+        };
+        // SAFETY: The parent passed this owned read end as an inherited descriptor.
+        let mut cleanup = unsafe { std::fs::File::from_raw_fd(cleanup_fd) };
+        // Preserve a final empty argument through musl's constructor argv fallback.
+        assert_eq!(arguments.collect::<Vec<_>>(), [std::ffi::OsString::new()]);
+        let mut signal = 0;
+        assert_eq!(
+            unsafe { libc::prctl(libc::PR_GET_PDEATHSIG, &mut signal) },
+            0
+        );
+        assert_eq!(signal, libc::SIGTERM);
+        println!("{}", std::process::id());
+        // Only the test owns the write end. EOF provides cleanup if the
+        // parent-death signal fails, without a timer that could mask that failure.
+        match std::io::copy(&mut cleanup, &mut std::io::sink()) {
+            Ok(_) => std::process::exit(0),
+            Err(_) => std::process::exit(1),
+        }
+    }
+    if argument.as_deref() == Some(std::ffi::OsStr::new(PARENT_FIXTURE)) {
+        let result = (|| -> anyhow::Result<()> {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()?;
+            runtime.block_on(async {
+                let cleanup = std::io::stdin().as_fd().try_clone_to_owned()?;
+                // The shared launcher only preserves explicitly inheritable fds.
+                assert_eq!(
+                    unsafe { libc::fcntl(cleanup.as_raw_fd(), libc::F_SETFD, 0) },
+                    0
+                );
+                let mut child = spawn_pipe_process_no_stdin(
+                    "/proc/self/exe",
+                    &[
+                        CHILD_FIXTURE.to_string(),
+                        cleanup.as_raw_fd().to_string(),
+                        String::new(),
+                    ],
+                    Path::new("/"),
+                    &HashMap::new(),
+                    &None,
+                    &[cleanup.as_raw_fd()],
+                )
+                .await?;
+                let ready = child
+                    .stdout_rx
+                    .recv()
+                    .await
+                    .ok_or_else(|| io::Error::other("fixture child did not start"))?;
+                std::io::Write::write_all(&mut std::io::stdout(), &ready)?;
+                // Kill this parent without destructors so session cleanup
+                // cannot mask the kernel's parent-death signal behavior.
+                std::future::pending::<()>().await;
+                anyhow::Ok(())
+            })
+        })();
+        if let Err(error) = result {
+            eprintln!("spawn fixture failed: {error}");
+            std::process::exit(1);
+        }
+        std::process::exit(0);
+    }
+}
+
+/// Pin a process identity so exit observation and cleanup cannot follow PID reuse.
+fn open_pidfd(pid: libc::pid_t) -> io::Result<OwnedFd> {
+    // SAFETY: pidfd_open takes a PID and flags, and returns a new owned descriptor.
+    let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0u32) };
+    if fd == -1 {
+        return Err(io::Error::last_os_error());
+    }
+    // SAFETY: The successful syscall returned a fresh descriptor owned by this call.
+    Ok(unsafe { OwnedFd::from_raw_fd(fd as _) })
+}
+
+#[tokio::test]
+async fn helper_target_dies_when_parent_is_killed() -> anyhow::Result<()> {
+    use tokio::io::AsyncBufReadExt;
+    if !Path::new("/proc/self/exe").exists() || process_arguments().is_err() {
+        eprintln!("skipping parent-death test: procfs is unavailable");
+        return Ok(());
+    }
+    // Check support and sandbox permissions before creating fixtures.
+    match open_pidfd(std::process::id() as _) {
+        Ok(_) => {}
+        Err(error)
+            if matches!(
+                error.raw_os_error(),
+                Some(libc::ENOSYS | libc::EPERM | libc::EACCES)
+            ) =>
+        {
+            eprintln!("skipping parent-death test: pidfd_open is unavailable");
+            return Ok(());
+        }
+        Err(error) => return Err(error.into()),
+    }
+    let mut parent = tokio::process::Command::new(std::env::current_exe()?)
+        .arg(PARENT_FIXTURE)
+        .stdin(std::process::Stdio::piped())
+        .stdout(std::process::Stdio::piped())
+        .kill_on_drop(true)
+        .spawn()?;
+    // Keep the cleanup pipe open independently of the parent's lifetime.
+    let cleanup = parent.stdin.take().expect("parent fixture stdin is piped");
+    let mut output = tokio::io::BufReader::new(parent.stdout.take().unwrap()).lines();
+    let pid = tokio::time::timeout(Duration::from_secs(5), output.next_line())
+        .await??
+        .expect("child reports its PID after verifying PDEATHSIG")
+        .parse::<libc::pid_t>()?;
+    // The fixture child waits on the cleanup pipe while its parent is alive.
+    let child = tokio::io::unix::AsyncFd::new(open_pidfd(pid)?)?;
+    parent.kill().await?;
+    let stopped = tokio::time::timeout(Duration::from_secs(5), child.readable()).await;
+    drop(cleanup);
+    if stopped.is_err() {
+        drop(tokio::time::timeout(Duration::from_secs(5), child.readable()).await??);
+    }
+    drop(stopped??);
+    Ok(())
+}
+
+/// Probe an actual helper launch before requiring the native handshake.
+async fn native_helper_available() -> anyhow::Result<bool> {
+    if !Path::new("/proc/self/exe").exists() || process_arguments().is_err() {
+        return Ok(false);
+    }
+    let command = crate::Command::new("/bin/true");
+    let Some(child) = crate::spawn_helper::spawn(&command).await? else {
+        return Ok(false);
+    };
+    Ok(child.wait_with_output().await?.status.success())
 }
 
 #[tokio::test]
@@ -52,17 +204,7 @@ async fn helper_preserves_cwd_env_arg0_and_streams() -> anyhow::Result<()> {
         &Some("custom-arg0".into()),
         &[],
     ).await?;
-    #[cfg(target_env = "gnu")]
-    let native_available = unsafe {
-        !libc::dlsym(
-            libc::RTLD_DEFAULT,
-            c"posix_spawn_file_actions_addchdir_np".as_ptr(),
-        )
-        .is_null()
-    };
-    #[cfg(not(target_env = "gnu"))]
-    let native_available = true;
-    if native_available {
+    if native_helper_available().await? {
         assert_eq!(FORKS.with(std::cell::Cell::get), 0);
     }
     child
@@ -128,6 +270,286 @@ async fn target_loader_environment_does_not_disable_helper_dispatch() -> anyhow:
     Ok(())
 }
 
+#[tokio::test]
+async fn large_environment_can_cross_the_control_socket() -> anyhow::Result<()> {
+    if !Path::new("/proc/self/exe").exists() || process_arguments().is_err() {
+        eprintln!("skipping control-socket test: procfs is unavailable");
+        return Ok(());
+    }
+    // Stay below Linux's minimum ARG_MAX; cancellation tests force socket backpressure.
+    let mut command = crate::Command::new("/bin/sh");
+    command
+        .args([
+            "-c",
+            "test ${#LARGE_0} -eq 8192 && test ${#LARGE_7} -eq 8192",
+        ])
+        .current_dir("/")
+        .envs((0..8).map(|index| (format!("LARGE_{index}"), "x".repeat(8_192))));
+    let child = tokio::time::timeout(Duration::from_secs(5), crate::spawn_helper::spawn(&command))
+        .await??
+        .expect("native helper must transfer the environment");
+    assert!(child.wait_with_output().await?.status.success());
+    Ok(())
+}
+
+#[tokio::test]
+async fn helper_preserves_inheritable_fds_and_closes_unrequested_fds() -> anyhow::Result<()> {
+    let file = std::fs::File::open("/dev/null")?;
+    if std::fs::metadata(format!("/proc/self/fd/{}", file.as_raw_fd())).is_err() {
+        eprintln!("skipping descriptor inspection test: procfs is unavailable");
+        return Ok(());
+    }
+    let descriptors = [file.try_clone()?, file.try_clone()?];
+    let preserved = descriptors[0].as_raw_fd();
+    let unwanted = descriptors[1].as_raw_fd();
+    assert_eq!(unsafe { libc::fcntl(unwanted, libc::F_SETFD, 0) }, 0);
+    assert_eq!(unsafe { libc::fcntl(preserved, libc::F_SETFD, 0) }, 0);
+    let script = format!("test -e /proc/self/fd/{preserved} && test ! -e /proc/self/fd/{unwanted}");
+    let mut command = crate::Command::new("/bin/sh");
+    command
+        .args(["-c", &script])
+        .current_dir("/")
+        .descriptor_policy(crate::DescriptorPolicy::Explicit)
+        .preserve_fds(&[preserved]);
+    let child = crate::spawn_helper::spawn(&command)
+        .await?
+        .expect("native helper must preserve requested descriptors");
+    assert!(child.wait_with_output().await?.status.success());
+    assert_eq!(unsafe { libc::fcntl(preserved, libc::F_GETFD) }, 0);
+    assert_eq!(unsafe { libc::fcntl(unwanted, libc::F_GETFD) }, 0);
+    Ok(())
+}
+
+// Test-only helper synchronization travels as target arguments and inherited
+// descriptors. Production dispatch has no environment switches or pause points.
+pub(crate) fn pause_handshake(phase: &str) -> io::Result<()> {
+    let arguments = process_arguments()?;
+    let marker = format!("--codex-spawn-test-pause={phase}");
+    let Some(index) = arguments
+        .iter()
+        .position(|argument| argument == std::ffi::OsStr::new(&marker))
+    else {
+        return Ok(());
+    };
+    let fd = arguments
+        .get(index + 1)
+        .and_then(|argument| argument.to_str())
+        .and_then(|argument| argument.parse::<i32>().ok())
+        .ok_or_else(|| io::Error::other("invalid handshake notification fd"))?;
+    let pid = std::process::id().to_le_bytes();
+    assert_eq!(
+        unsafe { libc::write(fd, pid.as_ptr().cast(), pid.len()) },
+        4
+    );
+    loop {
+        unsafe { libc::pause() };
+    }
+}
+
+// Keep the inherited notification socket alive inside the future, so the
+// cancellation tests exercise the same fd ownership as a real spawn caller.
+fn paused_spawn(
+    phase: &str,
+) -> io::Result<(
+    impl std::future::Future<Output = anyhow::Result<crate::SpawnedProcess>>,
+    tokio::net::UnixStream,
+)> {
+    let (notify, helper_notify) = std::os::unix::net::UnixStream::pair()?;
+    notify.set_nonblocking(true)?;
+    let notify = tokio::net::UnixStream::from_std(notify)?;
+    assert_eq!(
+        unsafe { libc::fcntl(helper_notify.as_raw_fd(), libc::F_SETFD, 0) },
+        0
+    );
+    let arguments = vec![
+        "-c".into(),
+        "printf fallback".into(),
+        "sh".into(),
+        format!("--codex-spawn-test-pause={phase}"),
+        helper_notify.as_raw_fd().to_string(),
+    ];
+    let env = if phase == "environment" {
+        // Exceed socket capacity to test cancellation of an environment write.
+        HashMap::from([("LARGE".into(), "x".repeat(2 * 1024 * 1024))])
+    } else {
+        HashMap::new()
+    };
+    let spawn = async move {
+        spawn_pipe_process_no_stdin(
+            "/bin/sh",
+            &arguments,
+            Path::new("/"),
+            &env,
+            &None,
+            &[helper_notify.as_raw_fd()],
+        )
+        .await
+    };
+    Ok((spawn, notify))
+}
+
+async fn paused_pid(
+    spawn: std::pin::Pin<
+        &mut impl std::future::Future<Output = anyhow::Result<crate::SpawnedProcess>>,
+    >,
+    notify: &mut tokio::net::UnixStream,
+) -> anyhow::Result<libc::pid_t> {
+    use tokio::io::AsyncReadExt;
+    let mut pid = [0; 4];
+    tokio::time::timeout(Duration::from_secs(5), async {
+        tokio::select! {
+            biased;
+            result = spawn => panic!("helper returned before cancellation: {result:?}"),
+            result = notify.read_exact(&mut pid) => result,
+        }
+    })
+    .await??;
+    Ok(u32::from_le_bytes(pid) as libc::pid_t)
+}
+
+fn is_reaped(pid: libc::pid_t) -> bool {
+    let mut info = std::mem::MaybeUninit::<libc::siginfo_t>::zeroed();
+    // SAFETY: Observe our child without reaping it or probing an unrelated reused PID.
+    if unsafe {
+        libc::waitid(
+            libc::P_PID,
+            pid as libc::id_t,
+            info.as_mut_ptr(),
+            libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
+        )
+    } == 0
+    {
+        return false;
+    }
+    assert_eq!(
+        io::Error::last_os_error().raw_os_error(),
+        Some(libc::ECHILD)
+    );
+    true
+}
+
+fn wait_for_fallback_reaping(pid: libc::pid_t) -> anyhow::Result<()> {
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    while !is_reaped(pid) {
+        anyhow::ensure!(std::time::Instant::now() < deadline, "child was not reaped");
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn cancelling_real_helper_handshakes_kills_and_reaps() -> anyhow::Result<()> {
+    if !native_helper_available().await? {
+        eprintln!("skipping helper handshake test: native spawning is unavailable");
+        return Ok(());
+    }
+    for phase in ["environment", "report", "exec"] {
+        let (spawn, mut notify) = paused_spawn(phase)?;
+        let mut spawn = Box::pin(spawn);
+        let pid = paused_pid(spawn.as_mut(), &mut notify).await?;
+        drop(spawn);
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while !is_reaped(pid) {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await?;
+    }
+    Ok(())
+}
+
+#[test]
+fn runtime_shutdown_retains_reaping_ownership_without_head_of_line_blocking() -> anyhow::Result<()>
+{
+    let mut children = Vec::new();
+    for _ in 0..2 {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?;
+        let child = runtime.block_on(async {
+            let mut child = spawn_pipe_process_no_stdin(
+                "/bin/sh",
+                &["-c".into(), "echo $$; exec sleep 60".into()],
+                Path::new("/"),
+                &HashMap::new(),
+                &None,
+                &[],
+            )
+            .await?;
+            let output = tokio::time::timeout(Duration::from_secs(5), child.stdout_rx.recv())
+                .await?
+                .expect("target reports pid");
+            let pid = std::str::from_utf8(&output)?
+                .trim()
+                .parse::<libc::pid_t>()?;
+            anyhow::Ok((child, pid))
+        })?;
+        // Shutdown completes the first child's transfer to the shared reaper
+        // before the second can arrive, while both children remain alive.
+        drop(runtime);
+        children.push(child);
+    }
+    for index in [1, 0] {
+        children[index].0.session.terminate();
+        wait_for_fallback_reaping(children[index].1)?;
+        if index == 1 {
+            assert_eq!(unsafe { libc::kill(children[0].1, 0) }, 0);
+        }
+    }
+    Ok(())
+}
+
+#[test]
+fn cancelling_startup_after_runtime_shutdown_reaps_helper() -> anyhow::Result<()> {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+    if !runtime.block_on(native_helper_available())? {
+        eprintln!("skipping helper handshake test: native spawning is unavailable");
+        return Ok(());
+    }
+    let (spawn, mut notify) = {
+        let _entered = runtime.enter();
+        paused_spawn("environment")?
+    };
+    let mut spawn = Box::pin(spawn);
+    let pid = runtime.block_on(paused_pid(spawn.as_mut(), &mut notify))?;
+    drop(runtime);
+    drop(spawn);
+    wait_for_fallback_reaping(pid)
+}
+
+/// Read argv before Rust's musl startup has initialized std::env::args_os.
+fn process_arguments() -> io::Result<Vec<std::ffi::OsString>> {
+    use std::os::unix::ffi::OsStringExt;
+    let command_line = std::fs::read("/proc/self/cmdline")?;
+    Ok(command_line
+        .strip_suffix(&[0])
+        .unwrap_or(&command_line)
+        .split(|byte| *byte == 0)
+        .map(|arg| std::ffi::OsString::from_vec(arg.to_vec()))
+        .collect())
+}
+
+#[tokio::test]
+async fn helper_exit_before_acknowledgement_falls_back() -> anyhow::Result<()> {
+    if !native_helper_available().await? {
+        eprintln!("skipping helper handshake test: native spawning is unavailable");
+        return Ok(());
+    }
+    let (spawn, mut notify) = paused_spawn("report")?;
+    let mut spawn = Box::pin(spawn);
+    let pid = paused_pid(spawn.as_mut(), &mut notify).await?;
+    assert_eq!(unsafe { libc::kill(pid, libc::SIGKILL) }, 0);
+    let mut child = tokio::time::timeout(Duration::from_secs(5), spawn).await??;
+    let mut stdout = Vec::new();
+    while let Some(chunk) = child.stdout_rx.recv().await {
+        stdout.extend(chunk);
+    }
+    assert_eq!((child.exit_rx.await?, stdout), (0, b"fallback".to_vec()));
+    wait_for_fallback_reaping(pid)
+}
+
 #[test]
 fn helper_dispatch_before_rust_initializes_argv() -> anyhow::Result<()> {
     let output = std::process::Command::new(std::env::current_exe()?)