diff --git a/codex-rs/app-server-daemon/src/backend/pid_start.rs b/codex-rs/app-server-daemon/src/backend/pid_start.rs index 212f90e9aa..1c286a8b14 100644 --- a/codex-rs/app-server-daemon/src/backend/pid_start.rs +++ b/codex-rs/app-server-daemon/src/backend/pid_start.rs @@ -235,7 +235,11 @@ impl PidBackend { } } - let child = match command.spawn() { + #[cfg(windows)] + let child = super::super::windows::spawn_without_inheriting_stdio(&mut command); + #[cfg(not(windows))] + let child = command.spawn().map_err(anyhow::Error::from); + let child = match child { Ok(child) => child, Err(err) => { if replacement.is_none() { diff --git a/codex-rs/app-server-daemon/src/backend/windows.rs b/codex-rs/app-server-daemon/src/backend/windows.rs index 57514b73cd..3b8b91ae47 100644 --- a/codex-rs/app-server-daemon/src/backend/windows.rs +++ b/codex-rs/app-server-daemon/src/backend/windows.rs @@ -2,6 +2,7 @@ //! so PID reuse can never redirect forced termination to a different process. //! Managed servers must not elevate ordinary clients sharing the account's socket. //! Installer jobs contain extraction processes when an update is cancelled. +//! Detached launches stop the launcher's original stdio handles from propagating. use std::io; use std::os::windows::io::AsRawHandle; @@ -14,9 +15,13 @@ use std::process::Stdio; use anyhow::Context; use anyhow::Result; +use windows_sys::Win32::Foundation::ERROR_INVALID_HANDLE; use windows_sys::Win32::Foundation::ERROR_INVALID_PARAMETER; use windows_sys::Win32::Foundation::ERROR_LOCK_VIOLATION; use windows_sys::Win32::Foundation::FILETIME; +use windows_sys::Win32::Foundation::HANDLE_FLAG_INHERIT; +use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; +use windows_sys::Win32::Foundation::SetHandleInformation; use windows_sys::Win32::Foundation::WAIT_OBJECT_0; use windows_sys::Win32::Foundation::WAIT_TIMEOUT; use windows_sys::Win32::Security::GetTokenInformation; @@ -48,6 +53,38 @@ use windows_sys::Win32::System::Threading::PROCESS_TERMINATE; use windows_sys::Win32::System::Threading::TerminateProcess; use windows_sys::Win32::System::Threading::WaitForSingleObject; +pub(super) fn spawn_without_inheriting_stdio( + command: &mut tokio::process::Command, +) -> Result { + // The daemon must not inherit the launcher's output pipes: callers wait for + // them to close after the launcher exits. Leave the flags cleared so concurrent + // launches cannot inherit them either; Rust duplicates the child's chosen stdio. + for (name, handle) in [ + ("stdin", io::stdin().as_raw_handle()), + ("stdout", io::stdout().as_raw_handle()), + ("stderr", io::stderr().as_raw_handle()), + ] { + if handle.is_null() || handle as isize == INVALID_HANDLE_VALUE { + continue; + } + // SAFETY: these are borrowed standard handles; changing the inherit + // flag neither closes them nor changes their read/write access. + if unsafe { + SetHandleInformation(handle as _, HANDLE_FLAG_INHERIT, /*dwflags*/ 0) + } == 0 + { + let error = io::Error::last_os_error(); + // An already-closed handle cannot be inherited. + if error.raw_os_error() == Some(ERROR_INVALID_HANDLE as i32) { + continue; + } + return Err(error) + .with_context(|| format!("failed to clear launcher {name} inheritance")); + } + } + Ok(command.spawn()?) +} + pub(crate) fn ensure_not_elevated() -> Result<()> { let mut token = 0; if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut token) } == 0 { diff --git a/codex-rs/app-server-daemon/src/backend/windows_tests.rs b/codex-rs/app-server-daemon/src/backend/windows_tests.rs index 87104adddb..ed9044bd18 100644 --- a/codex-rs/app-server-daemon/src/backend/windows_tests.rs +++ b/codex-rs/app-server-daemon/src/backend/windows_tests.rs @@ -4,11 +4,133 @@ use std::os::windows::io::AsRawHandle; use std::os::windows::io::FromRawHandle; use std::os::windows::io::OwnedHandle; use std::os::windows::process::CommandExt; +use std::process::Stdio; +use std::time::Duration; +use tokio::io::AsyncReadExt; use windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED; +use windows_sys::Win32::Foundation::HANDLE_FLAG_INHERIT; +use windows_sys::Win32::Foundation::SetHandleInformation; use windows_sys::Win32::System::JobObjects::IsProcessInJob; use windows_sys::Win32::System::Threading::CREATE_NO_WINDOW; +use windows_sys::Win32::System::Threading::DETACHED_PROCESS; use windows_sys::Win32::System::Threading::TerminateProcess; +#[tokio::test] +async fn captured_stdio_closes_while_child_is_alive() { + const TEST: &str = "backend::windows::tests::captured_stdio_closes_while_child_is_alive"; + const ROLE: &str = "CODEX_TEST_STDIO_ROLE"; + const HOME: &str = "CODEX_TEST_STDIO_HOME"; + let executable = std::env::current_exe().expect("test executable"); + + if let Ok(role) = std::env::var(ROLE) { + let home = std::path::PathBuf::from(std::env::var_os(HOME).expect("test home")); + if role == "child" { + eprintln!("child stderr"); + std::fs::write(home.join("ready"), []).unwrap(); + tokio::time::sleep(Duration::from_secs(30)).await; + return; + } + + let handles = [ + std::io::stdout().as_raw_handle() as _, + std::io::stderr().as_raw_handle() as _, + ]; + for handle in handles { + assert_ne!( + unsafe { SetHandleInformation(handle, HANDLE_FLAG_INHERIT, HANDLE_FLAG_INHERIT) }, + 0 + ); + } + let mut command = tokio::process::Command::new(executable); + command + .args(["--exact", TEST, "--nocapture"]) + .env(ROLE, "child") + // The CI job may forbid breakaway; it does not affect stdio inheritance. + .creation_flags(DETACHED_PROCESS) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(home.join("child.log")).unwrap()); + let child = super::spawn_without_inheriting_stdio(&mut command).unwrap(); + let pid = child.id().unwrap(); + let started = Process::open(pid).unwrap().unwrap().start_time().unwrap(); + std::fs::write( + home.join("pid"), + serde_json::to_vec(&(pid, started)).unwrap(), + ) + .unwrap(); + println!("launcher stdout"); + eprintln!("launcher stderr"); + return; + } + + let home = tempfile::tempdir().unwrap(); + let mut launcher = tokio::process::Command::new(executable) + .args(["--exact", TEST, "--nocapture"]) + .env(ROLE, "launcher") + .env(HOME, home.path()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .unwrap(); + let status = tokio::time::timeout(Duration::from_secs(10), launcher.wait()) + .await + .expect("launcher exit") + .unwrap(); + let (pid, started): (u32, String) = + serde_json::from_slice(&std::fs::read(home.path().join("pid")).unwrap()).unwrap(); + let child = Process::open(pid).unwrap().expect("live child"); + assert_eq!( + child.start_time().unwrap(), + started, + "fixture PID was reused" + ); + let mut stdout = String::new(); + let mut stderr = String::new(); + let captured = tokio::time::timeout(Duration::from_secs(5), async { + let ready = home.path().join("ready"); + while !ready.exists() { + tokio::time::sleep(Duration::from_millis(25)).await; + } + tokio::try_join!( + launcher + .stdout + .as_mut() + .unwrap() + .read_to_string(&mut stdout), + launcher + .stderr + .as_mut() + .unwrap() + .read_to_string(&mut stderr), + ) + }) + .await; + let still_running = child.is_running(); + child.terminate().expect("cleanup child"); + tokio::time::timeout(Duration::from_secs(5), async { + while child.is_running().unwrap() { + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("child exit after cleanup"); + + captured.expect("launcher output must close").unwrap(); + assert!(status.success(), "{stderr}"); + assert!( + still_running.unwrap(), + "child exited before its launcher output closed" + ); + assert!(stdout.contains("launcher stdout")); + assert!(stderr.contains("launcher stderr")); + assert!( + std::fs::read_to_string(home.path().join("child.log")) + .unwrap() + .contains("child stderr") + ); +} + #[test] fn detached_launch_preflight_rejects_restrictive_job() { const CHILD: &str = "CODEX_TEST_RESTRICTIVE_LAUNCH_JOB"; diff --git a/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs b/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs index f5ef343ae4..d366b75f41 100644 --- a/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs +++ b/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs @@ -14,6 +14,9 @@ use test_case::test_case; const SERVER_NAME: &str = "optional_startup"; const TOOL_NAMESPACE: &str = "mcp__optional_startup"; const TOOL_NAME: &str = "calendar_create_event"; +// Allow the whole turn to finish on slow runners without reaching the server's own timeout. +const TURN_TIMEOUT: Duration = Duration::from_secs(5); +const PENDING_SERVER_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone, Copy)] enum StartupGraceScenario { @@ -46,9 +49,9 @@ async fn optional_mcp_startup_grace_controls_initial_turn_tool_catalog( | StartupGraceScenario::DisabledGraceRespectsStartupTimeout => Duration::ZERO, }; let startup_timeout = match scenario { + StartupGraceScenario::ShortGraceOmitsPending => PENDING_SERVER_TIMEOUT, StartupGraceScenario::DisabledGraceRespectsStartupTimeout => Duration::from_millis(250), - StartupGraceScenario::ShortGraceOmitsPending - | StartupGraceScenario::CustomGraceIncludesReady + StartupGraceScenario::CustomGraceIncludesReady | StartupGraceScenario::DisabledGraceWaitsForStartup => Duration::from_secs(1), }; let responses_server = responses::start_mock_server().await; @@ -101,7 +104,7 @@ async fn optional_mcp_startup_grace_controls_initial_turn_tool_catalog( let mut turn = Box::pin(fixture.submit_turn("show optional MCP tools")); match scenario { StartupGraceScenario::ShortGraceOmitsPending => { - tokio::time::timeout(Duration::from_millis(500), &mut turn) + tokio::time::timeout(TURN_TIMEOUT, &mut turn) .await .context("the configured grace should omit the pending server")??; release_startup @@ -206,7 +209,7 @@ async fn running_thread_uses_refreshed_optional_mcp_startup_grace( "url": server_url, "http_headers": { "Authorization": "Bearer synthetic-test-token" }, "enabled_tools": [TOOL_NAME], - "startup_timeout_sec": 5, + "startup_timeout_sec": PENDING_SERVER_TIMEOUT.as_secs(), })) .expect("synthetic optional MCP server configuration"), ); @@ -226,8 +229,10 @@ async fn running_thread_uses_refreshed_optional_mcp_startup_grace( .await .context("optional MCP initialization should begin before the initial turn")?; + // Allow for turn setup on remote workers while still finishing before the + // pending MCP startup timeout. The server remains gated throughout. tokio::time::timeout( - Duration::from_millis(500), + TURN_TIMEOUT, fixture.submit_turn("show initial optional MCP tools"), ) .await