diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 20dd36b6bd..5194ab6059 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3230,6 +3230,7 @@ dependencies = [ "codex-protocol", "codex-utils-absolute-path", "codex-utils-path-uri", + "codex-utils-pty", "futures", "gix", "once_cell", diff --git a/codex-rs/git-utils/Cargo.toml b/codex-rs/git-utils/Cargo.toml index 18ba488252..a0bd787285 100644 --- a/codex-rs/git-utils/Cargo.toml +++ b/codex-rs/git-utils/Cargo.toml @@ -15,6 +15,7 @@ codex-file-system = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-path-uri = { workspace = true } +codex-utils-pty = { workspace = true } futures = { workspace = true, features = ["alloc"] } gix = { workspace = true } once_cell = { workspace = true } diff --git a/codex-rs/git-utils/src/git_process.rs b/codex-rs/git-utils/src/git_process.rs new file mode 100644 index 0000000000..f9b6566232 --- /dev/null +++ b/codex-rs/git-utils/src/git_process.rs @@ -0,0 +1,104 @@ +use std::process::Output; +use std::process::Stdio; +use std::time::Duration; + +#[cfg(windows)] +use codex_utils_pty::JobObject; +#[cfg(unix)] +use codex_utils_pty::process_group::kill_process_group; +use tokio::process::Child; +use tokio::process::Command; +use tokio::time::timeout; + +struct KillGitProcessTreeOnDrop { + #[cfg(unix)] + process_id: u32, + #[cfg(windows)] + job: Option, + #[cfg(unix)] + armed: bool, +} + +#[cfg(unix)] +impl Drop for KillGitProcessTreeOnDrop { + fn drop(&mut self) { + if self.armed { + let _ = kill_process_group(self.process_id); + } + } +} + +fn spawn_git_command(command: &mut Command) -> Option<(Child, KillGitProcessTreeOnDrop)> { + #[cfg(unix)] + command.process_group(0); + command.kill_on_drop(true); + + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + #[cfg(windows)] + let (child, job) = match JobObject::create() + .and_then(|job| job.spawn_contained(command).map(|child| (child, job))) + { + Ok((child, job)) => (child, Some(job)), + Err(_) => { + // A failed contained spawn leaves CREATE_SUSPENDED on the command. + command.creation_flags(0); + (command.spawn().ok()?, None) + } + }; + #[cfg(not(windows))] + let child = command.spawn().ok()?; + + let process_tree = KillGitProcessTreeOnDrop { + #[cfg(unix)] + process_id: child.id()?, + #[cfg(windows)] + job, + #[cfg(unix)] + armed: true, + }; + + Some((child, process_tree)) +} + +async fn wait_for_git_command_with_timeout_output( + child: Child, + process_tree: KillGitProcessTreeOnDrop, + timeout_duration: Duration, +) -> Option { + #[cfg(unix)] + let mut process_tree = process_tree; + + let result = timeout(timeout_duration, child.wait_with_output()).await; + + match result { + Ok(Ok(output)) => { + #[cfg(windows)] + if let Some(job) = &process_tree.job { + job.preserve_descendants().ok()?; + } + + #[cfg(unix)] + { + process_tree.armed = false; + } + Some(output) + } + _ => None, + } +} + +pub(crate) async fn run_git_command_with_timeout_output( + command: &mut Command, + timeout_duration: Duration, +) -> Option { + let (child, process_tree) = spawn_git_command(command)?; + wait_for_git_command_with_timeout_output(child, process_tree, timeout_duration).await +} + +#[cfg(test)] +#[path = "git_process_tests.rs"] +mod tests; diff --git a/codex-rs/git-utils/src/git_process_tests.rs b/codex-rs/git-utils/src/git_process_tests.rs new file mode 100644 index 0000000000..1640d62668 --- /dev/null +++ b/codex-rs/git-utils/src/git_process_tests.rs @@ -0,0 +1,134 @@ +use super::spawn_git_command; +use super::wait_for_git_command_with_timeout_output; +use pretty_assertions::assert_eq; +#[cfg(windows)] +use std::process::Stdio; +use std::time::Duration; +use tokio::process::Command; + +#[derive(Clone, Copy)] +enum GitWrapperLifetime { + WaitForChild, + ExitBeforeTimeout, +} + +async fn assert_timed_out_git_wrapper_does_not_leave_child_process_running( + wrapper_lifetime: GitWrapperLifetime, +) { + let temp_dir = tempfile::tempdir().expect("create temp dir"); + let child_pid_file = temp_dir.path().join("child.pid"); + let child_ready_file = temp_dir.path().join("child-ready"); + let release_child_file = temp_dir.path().join("release-child"); + let child_survived_file = temp_dir.path().join("child-survived"); + let release_wrapper_file = temp_dir.path().join("release-wrapper"); + #[cfg(unix)] + let mut command = { + let mut command = Command::new("/bin/sh"); + let wrapper_command = match wrapper_lifetime { + GitWrapperLifetime::WaitForChild => { + r#"( : > "$CHILD_READY_FILE"; while [ ! -f "$RELEASE_CHILD_FILE" ]; do sleep 0.01; done; sleep 1; : > "$CHILD_SURVIVED_FILE"; sleep 60 ) & child_pid=$!; printf '%s\n' "$child_pid" > "$CHILD_PID_FILE"; wait "$child_pid""# + } + GitWrapperLifetime::ExitBeforeTimeout => { + r#"( : > "$CHILD_READY_FILE"; while [ ! -f "$RELEASE_CHILD_FILE" ]; do sleep 0.01; done; sleep 1; : > "$CHILD_SURVIVED_FILE"; sleep 60 ) & child_pid=$!; printf '%s\n' "$child_pid" > "$CHILD_PID_FILE"; while [ ! -f "$RELEASE_WRAPPER_FILE" ]; do sleep 0.01; done"# + } + }; + command.args(["-c", wrapper_command]); + command + }; + #[cfg(windows)] + let mut command = { + let mut command = Command::new("powershell.exe"); + let child_command = "Set-Content -LiteralPath $env:CHILD_READY_FILE -Value ready; while (-not (Test-Path $env:RELEASE_CHILD_FILE)) { Start-Sleep -Milliseconds 25 }; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:CHILD_SURVIVED_FILE -Value survived; Start-Sleep -Seconds 60"; + let wrapper_command = match wrapper_lifetime { + GitWrapperLifetime::WaitForChild => format!( + "$child = Start-Process -FilePath powershell.exe -ArgumentList @('-NoProfile', '-NonInteractive', '-Command', '{child_command}') -PassThru -NoNewWindow; [System.IO.File]::WriteAllText($env:CHILD_PID_FILE, [string]$child.Id); Wait-Process -Id $child.Id" + ), + GitWrapperLifetime::ExitBeforeTimeout => format!( + "$child = Start-Process -FilePath powershell.exe -ArgumentList @('-NoProfile', '-NonInteractive', '-Command', '{child_command}') -PassThru -NoNewWindow; [System.IO.File]::WriteAllText($env:CHILD_PID_FILE, [string]$child.Id); while (-not (Test-Path $env:RELEASE_WRAPPER_FILE)) {{ Start-Sleep -Milliseconds 25 }}" + ), + }; + command + .args(["-NoProfile", "-NonInteractive", "-Command"]) + .arg(wrapper_command); + command + }; + command + .env("CHILD_PID_FILE", &child_pid_file) + .env("CHILD_READY_FILE", &child_ready_file) + .env("RELEASE_CHILD_FILE", &release_child_file) + .env("CHILD_SURVIVED_FILE", &child_survived_file) + .env("RELEASE_WRAPPER_FILE", &release_wrapper_file); + + let (mut wrapper, process_tree) = spawn_git_command(&mut command).expect("spawn Git wrapper"); + let child_pid = tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Ok(child_pid) = std::fs::read_to_string(&child_pid_file) + && !child_pid.trim().is_empty() + && child_ready_file.exists() + { + break child_pid.trim().to_string(); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("wait for Git wrapper child readiness"); + + if matches!(wrapper_lifetime, GitWrapperLifetime::ExitBeforeTimeout) { + std::fs::write(&release_wrapper_file, "release").expect("release Git wrapper"); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if wrapper + .try_wait() + .expect("check Git wrapper state") + .is_some() + { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("wait for Git wrapper exit"); + } + + let output = + wait_for_git_command_with_timeout_output(wrapper, process_tree, Duration::from_millis(100)) + .await; + assert_eq!(output, None); + + std::fs::write(&release_child_file, "release").expect("release Git wrapper child"); + tokio::time::sleep(Duration::from_secs(3)).await; + if !child_survived_file.exists() { + return; + } + + #[cfg(unix)] + let _ = std::process::Command::new("kill") + .args(["-KILL", &child_pid]) + .status(); + #[cfg(windows)] + let _ = std::process::Command::new("taskkill") + .args(["/PID", &child_pid, "/T", "/F"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + panic!("Git wrapper child process {child_pid} survived timeout cleanup"); +} + +#[tokio::test] +async fn timed_out_git_wrapper_does_not_leave_child_process_running() { + assert_timed_out_git_wrapper_does_not_leave_child_process_running( + GitWrapperLifetime::WaitForChild, + ) + .await; +} + +#[tokio::test] +async fn timed_out_exited_git_wrapper_does_not_leave_child_process_running() { + assert_timed_out_git_wrapper_does_not_leave_child_process_running( + GitWrapperLifetime::ExitBeforeTimeout, + ) + .await; +} diff --git a/codex-rs/git-utils/src/info.rs b/codex-rs/git-utils/src/info.rs index ff013525d4..fc45572703 100644 --- a/codex-rs/git-utils/src/info.rs +++ b/codex-rs/git-utils/src/info.rs @@ -3,7 +3,6 @@ use std::collections::HashSet; use std::ffi::OsStr; use std::path::Path; use std::path::PathBuf; -use std::process::Stdio; use codex_file_system::ExecutorFileSystem; use codex_file_system::FindUpErrorPolicy; @@ -16,10 +15,10 @@ use serde::Deserialize; use serde::Serialize; use tokio::process::Command; use tokio::time::Duration as TokioDuration; -use tokio::time::timeout; use ts_rs::TS; use crate::GitSha; +use crate::git_process::run_git_command_with_timeout_output; /// Return `true` if the project folder specified by the `Config` is inside a /// Git repository. @@ -401,13 +400,9 @@ impl crate::FsmonitorProbeRunner for LocalFsmonitorProbeRunner<'_> { // Both probes are fast, bounded metadata queries that do not inspect the // worktree or index, so do not reduce the requested command's timeout. let mut command = Command::new(self.git); - command - .args(args) - .current_dir(self.cwd) - .stdin(Stdio::null()) - .kill_on_drop(true); - match timeout(GIT_COMMAND_TIMEOUT, command.output()).await { - Ok(Ok(output)) if output.status.success() => Some(output.stdout), + command.args(args).current_dir(self.cwd); + match run_git_command_with_timeout_output(&mut command, GIT_COMMAND_TIMEOUT).await { + Some(output) if output.status.success() => Some(output.stdout), _ => None, } } @@ -432,15 +427,8 @@ async fn run_git_command_with_timeout_from( .args(["-c", &format!("core.hooksPath={DISABLED_HOOKS_PATH}")]) .args(["-c", fsmonitor.git_config_arg()]) .args(args) - .current_dir(cwd) - .stdin(Stdio::null()) - .kill_on_drop(true); - let result = timeout(GIT_COMMAND_TIMEOUT, command.output()).await; - - match result { - Ok(Ok(output)) => Some(output), - _ => None, // Timeout or error - } + .current_dir(cwd); + run_git_command_with_timeout_output(&mut command, GIT_COMMAND_TIMEOUT).await } async fn get_git_remotes(cwd: &Path) -> Option> { @@ -905,6 +893,7 @@ mod tests { use pretty_assertions::assert_eq; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; + use std::process::Stdio; #[tokio::test] async fn git_metadata_commands_do_not_inherit_stdin() { diff --git a/codex-rs/git-utils/src/lib.rs b/codex-rs/git-utils/src/lib.rs index c91688470f..5cbcb89e1c 100644 --- a/codex-rs/git-utils/src/lib.rs +++ b/codex-rs/git-utils/src/lib.rs @@ -3,6 +3,7 @@ mod baseline; mod branch; mod errors; mod fsmonitor; +mod git_process; mod info; mod operations; mod platform; diff --git a/codex-rs/utils/pty/Cargo.toml b/codex-rs/utils/pty/Cargo.toml index d03962edc7..54f90f1ae5 100644 --- a/codex-rs/utils/pty/Cargo.toml +++ b/codex-rs/utils/pty/Cargo.toml @@ -22,6 +22,7 @@ log = { workspace = true } shared_library = "0.1.9" winapi = { version = "0.3.9", features = [ "handleapi", + "jobapi", "jobapi2", "minwinbase", "processthreadsapi", diff --git a/codex-rs/utils/pty/src/win/job.rs b/codex-rs/utils/pty/src/win/job.rs index 8073360573..f22152eb78 100644 --- a/codex-rs/utils/pty/src/win/job.rs +++ b/codex-rs/utils/pty/src/win/job.rs @@ -4,15 +4,26 @@ use std::os::windows::io::AsRawHandle; use std::os::windows::io::FromRawHandle; use std::os::windows::io::RawHandle; use std::sync::Mutex; +use tokio::process::Child; +use tokio::process::Command; +use winapi::shared::ntdef::NT_SUCCESS; +use winapi::shared::ntdef::NTSTATUS; use winapi::um::jobapi2::AssignProcessToJobObject; use winapi::um::jobapi2::CreateJobObjectW; use winapi::um::jobapi2::SetInformationJobObject; use winapi::um::jobapi2::TerminateJobObject; +use winapi::um::winbase::CREATE_SUSPENDED; +use winapi::um::winnt::HANDLE; use winapi::um::winnt::JOB_OBJECT_LIMIT_BREAKAWAY_OK; use winapi::um::winnt::JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; use winapi::um::winnt::JOBOBJECT_EXTENDED_LIMIT_INFORMATION; use winapi::um::winnt::JobObjectExtendedLimitInformation; +#[link(name = "ntdll")] +unsafe extern "system" { + fn NtResumeProcess(process_handle: HANDLE) -> NTSTATUS; +} + /// Owns a Windows Job Object used to terminate a spawned process tree. #[derive(Debug)] pub struct JobObject { @@ -74,6 +85,25 @@ impl JobObject { } } + /// Starts a process only after assigning it to this Job Object. + pub fn spawn_contained(&self, command: &mut Command) -> io::Result { + command.creation_flags(CREATE_SUSPENDED).kill_on_drop(true); + let child = command.spawn()?; + let process_handle = child + .raw_handle() + .ok_or_else(|| io::Error::other("missing child process handle"))?; + self.assign_process(process_handle)?; + + let status = unsafe { NtResumeProcess(process_handle.cast()) }; + if !NT_SUCCESS(status) { + return Err(io::Error::other(format!( + "failed to resume contained process: NTSTATUS {status:#x}" + ))); + } + + Ok(child) + } + /// Allows contained descendants to keep running after the root exits normally. /// /// This disables both explicit job termination and kill-on-close for this diff --git a/codex-rs/utils/pty/src/windows_tests.rs b/codex-rs/utils/pty/src/windows_tests.rs index 0dc095752c..91322be1cf 100644 --- a/codex-rs/utils/pty/src/windows_tests.rs +++ b/codex-rs/utils/pty/src/windows_tests.rs @@ -6,8 +6,18 @@ use crate::TerminalSize; use crate::spawn_pipe_process_no_stdin; use crate::spawn_pty_process; use std::collections::HashMap; +use std::os::windows::io::AsRawHandle; +use std::os::windows::io::FromRawHandle; +use std::os::windows::io::OwnedHandle; use std::path::Path; +use std::process::Stdio; use std::time::Duration; +use tokio::io::AsyncBufReadExt; +use tokio::io::BufReader; +use tokio::process::Command; +use winapi::um::jobapi::IsProcessInJob; +use winapi::um::processthreadsapi::OpenProcess; +use winapi::um::winnt::PROCESS_QUERY_LIMITED_INFORMATION; const READY_MARKER: &str = "__CODEX_CHILD_READY__"; const VALUE_MARKER: &str = "__CODEX_CHILD_VALUE__"; @@ -183,6 +193,54 @@ async fn normal_exit_preserves_descendants_for_pipe_and_conpty() -> anyhow::Resu assert_normal_exit_preserves_descendant("ConPTY", &python, &env).await } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn contained_spawn_owns_immediate_descendant() -> anyhow::Result<()> { + let Some(python) = find_python() else { + eprintln!("python not found; skipping Windows contained-spawn test"); + return Ok(()); + }; + + let mut command = Command::new(&python); + command + .args([ + "-u", + "-c", + "import subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); print(child.pid,flush=True); child.wait()", + ]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + + let job = crate::JobObject::create()?; + let mut root = job.spawn_contained(&mut command)?; + let stdout = root + .stdout + .take() + .ok_or_else(|| anyhow::anyhow!("missing contained process stdout"))?; + let mut stdout = BufReader::new(stdout); + let mut child_pid = String::new(); + tokio::time::timeout(Duration::from_secs(10), stdout.read_line(&mut child_pid)).await??; + let child_pid: u32 = child_pid.trim().parse()?; + + let process = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, child_pid) }; + anyhow::ensure!(!process.is_null(), "failed to open immediate child process"); + let process = unsafe { OwnedHandle::from_raw_handle(process.cast()) }; + let mut in_job = 0; + let checked = unsafe { + IsProcessInJob( + process.as_raw_handle().cast(), + job.as_raw_handle().cast(), + &mut in_job, + ) + }; + anyhow::ensure!(checked != 0, "failed to inspect child Job Object"); + anyhow::ensure!(in_job != 0, "immediate child escaped its Job Object"); + + job.terminate()?; + tokio::time::timeout(Duration::from_secs(10), root.wait()).await??; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conpty_delivers_input_to_foreground_children() -> anyhow::Result<()> { let Some(python) = find_python() else {