diff --git a/codex-rs/app-server-transport/src/daemon_shutdown.rs b/codex-rs/app-server-transport/src/daemon_shutdown.rs new file mode 100644 index 0000000000..7c7f23e69b --- /dev/null +++ b/codex-rs/app-server-transport/src/daemon_shutdown.rs @@ -0,0 +1,54 @@ +//! Detached Windows processes cannot receive console signals. The daemon passes +//! a request path inside its user-only state directory. Only the addressed PID +//! consumes a request, triggering the SIGTERM drain without changing the wire API. + +use std::io; +use std::io::Read; +use std::path::Path; +#[cfg(windows)] +use std::path::PathBuf; +#[cfg(windows)] +use std::time::Duration; + +#[cfg(windows)] +pub const DAEMON_SHUTDOWN_FILE_ENV: &str = "CODEX_DAEMON_SHUTDOWN_FILE"; + +/// Waits for and consumes one daemon shutdown request. Without a managed launch +/// environment this stays pending, leaving normal console shutdown unchanged. +#[cfg(windows)] +pub async fn daemon_shutdown_signal() -> io::Result<()> { + let Some(path) = std::env::var_os(DAEMON_SHUTDOWN_FILE_ENV).map(PathBuf::from) else { + return std::future::pending().await; + }; + loop { + if take_shutdown_request(&path, std::process::id())? { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +fn take_shutdown_request(path: &Path, pid: u32) -> io::Result { + let file = match std::fs::File::open(path) { + Ok(file) => file, + Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(false), + Err(err) => return Err(err), + }; + // Descendant app-servers may inherit the control path. Only the intended + // process can consume a request. Bound reads to a u32 PID plus one extra byte. + let mut contents = Vec::new(); + file.take(/*limit*/ 11).read_to_end(&mut contents)?; + if contents != pid.to_string().as_bytes() { + return Ok(false); + } + // Synchronous consumption cannot lose a request to select cancellation. + match std::fs::remove_file(path) { + Ok(()) => Ok(true), + Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false), + Err(err) => Err(err), + } +} + +#[cfg(test)] +#[path = "daemon_shutdown_tests.rs"] +mod tests; diff --git a/codex-rs/app-server-transport/src/daemon_shutdown_tests.rs b/codex-rs/app-server-transport/src/daemon_shutdown_tests.rs new file mode 100644 index 0000000000..7454dffcd1 --- /dev/null +++ b/codex-rs/app-server-transport/src/daemon_shutdown_tests.rs @@ -0,0 +1,22 @@ +use super::take_shutdown_request; + +#[test] +fn inherited_control_path_cannot_consume_another_process_request() { + let directory = tempfile::tempdir().expect("directory"); + let request = directory.path().join("shutdown"); + std::fs::write(&request, "1234").expect("request"); + assert!(!take_shutdown_request(&request, /*pid*/ 5678).expect("descendant probe")); + assert!(take_shutdown_request(&request, /*pid*/ 1234).expect("parent probe")); + assert!(!take_shutdown_request(&request, /*pid*/ 1234).expect("already consumed")); +} + +#[test] +fn incomplete_or_invalid_requests_are_not_consumed() { + let directory = tempfile::tempdir().expect("directory"); + let request = directory.path().join("shutdown"); + for contents in [b"".as_slice(), b"123", b"1234junk", &[0xff; 20]] { + std::fs::write(&request, contents).expect("request"); + assert!(!take_shutdown_request(&request, /*pid*/ 1234).expect("probe")); + assert!(request.exists()); + } +} diff --git a/codex-rs/app-server-transport/src/lib.rs b/codex-rs/app-server-transport/src/lib.rs index 9fb49f591d..9387ba73bf 100644 --- a/codex-rs/app-server-transport/src/lib.rs +++ b/codex-rs/app-server-transport/src/lib.rs @@ -1,3 +1,9 @@ +#[cfg(any(windows, test))] +mod daemon_shutdown; +#[cfg(windows)] +pub use daemon_shutdown::DAEMON_SHUTDOWN_FILE_ENV; +#[cfg(windows)] +pub use daemon_shutdown::daemon_shutdown_signal; mod outgoing_message; mod transport; diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 0b4fe17493..7edccfbf66 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -217,11 +217,28 @@ async fn shutdown_signal() -> IoResult { } } - #[cfg(not(unix))] + #[cfg(windows)] { - tokio::signal::ctrl_c() - .await - .map(|_| ShutdownSignal::Forceable) + let console_signal = async { + if tokio::signal::ctrl_c().await.is_err() { + // A detached daemon has no console; keep its local control path active. + std::future::pending::<()>().await; + } + }; + let daemon_signal = async { + let result = codex_app_server_transport::daemon_shutdown_signal().await; + // The processor retries listener errors; the updater instead needs + // immediate errors from its nonblocking shutdown probe. + if result.is_err() { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + result + }; + tokio::select! { + _ = console_signal => Ok(ShutdownSignal::Forceable), + result = daemon_signal => + result.map(|_| ShutdownSignal::Forceable), + } } } diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index 1e4a7a23fd..4b79d667fc 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -508,6 +508,12 @@ pub(super) async fn spawn_websocket_server_with_args( .stderr(Stdio::piped()) .env("CODEX_HOME", codex_home) .env("RUST_LOG", "warn"); + #[cfg(windows)] + cmd.env( + codex_app_server_transport::DAEMON_SHUTDOWN_FILE_ENV, + codex_home.join("daemon.shutdown"), + ); + let mut process = cmd .kill_on_drop(true) .spawn() diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs index 0c8a59a9c8..9faa21ec65 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs @@ -1,3 +1,6 @@ +//! Signal-driven shutdown tests; Windows uses the detached daemon control request +//! for SIGTERM-equivalent graceful and forced shutdown coverage. + use super::connection_handling_websocket::DEFAULT_READ_TIMEOUT; use super::connection_handling_websocket::WsClient; use super::connection_handling_websocket::connect_websocket; @@ -19,6 +22,8 @@ use codex_app_server_protocol::UserInput as V2UserInput; use core_test_support::responses; use futures::SinkExt; use futures::StreamExt; +use pretty_assertions::assert_eq; +#[cfg(unix)] use std::process::Command as StdCommand; use tempfile::TempDir; use tokio::process::Child; @@ -31,6 +36,7 @@ use wiremock::Mock; use wiremock::matchers::method; use wiremock::matchers::path_regex; +#[cfg(unix)] #[tokio::test] async fn websocket_transport_ctrl_c_waits_for_running_turn_before_exit() -> Result<()> { let GracefulCtrlCFixture { @@ -56,6 +62,7 @@ async fn websocket_transport_ctrl_c_waits_for_running_turn_before_exit() -> Resu Ok(()) } +#[cfg(unix)] #[tokio::test] async fn websocket_transport_second_ctrl_c_forces_exit_while_turn_running() -> Result<()> { let GracefulCtrlCFixture { @@ -91,7 +98,7 @@ async fn websocket_transport_sigterm_waits_for_running_turn_before_exit() -> Res mut ws, } = start_ctrl_c_restart_fixture(Duration::from_secs(3)).await?; - send_sigterm(&process)?; + send_sigterm(&process, _codex_home.path())?; assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?; let status = wait_for_process_exit_within( @@ -116,10 +123,10 @@ async fn websocket_transport_second_sigterm_forces_exit_while_turn_running() -> mut ws, } = start_ctrl_c_restart_fixture(Duration::from_secs(3)).await?; - send_sigterm(&process)?; + send_sigterm(&process, _codex_home.path())?; assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?; - send_sigterm(&process)?; + send_sigterm(&process, _codex_home.path())?; let status = wait_for_process_exit_within( &mut process, Duration::from_secs(2), @@ -133,6 +140,7 @@ async fn websocket_transport_second_sigterm_forces_exit_while_turn_running() -> Ok(()) } +#[cfg(unix)] #[tokio::test] async fn websocket_transport_repeated_sighup_keeps_waiting_for_running_turn() -> Result<()> { let GracefulCtrlCFixture { @@ -257,18 +265,22 @@ async fn wait_for_responses_post(server: &wiremock::MockServer, wait_for: Durati } } +#[cfg(unix)] fn send_sigint(process: &Child) -> Result<()> { send_signal(process, "-INT") } -fn send_sigterm(process: &Child) -> Result<()> { +#[cfg(unix)] +fn send_sigterm(process: &Child, _home: &std::path::Path) -> Result<()> { send_signal(process, "-TERM") } +#[cfg(unix)] fn send_sighup(process: &Child) -> Result<()> { send_signal(process, "-HUP") } +#[cfg(unix)] fn send_signal(process: &Child, signal: &str) -> Result<()> { let pid = process .id() @@ -325,3 +337,12 @@ async fn expect_websocket_disconnect(stream: &mut WsClient) -> Result<()> { } } } + +#[cfg(windows)] +fn send_sigterm(process: &Child, home: &std::path::Path) -> Result<()> { + std::fs::write( + home.join("daemon.shutdown"), + process.id().context("server pid")?.to_string(), + )?; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 028ee5918f..160bf3ccce 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -16,7 +16,7 @@ mod compaction; mod config_requirements_in_app_browser; mod config_rpc; mod connection_handling_websocket; -#[cfg(unix)] +#[cfg(any(unix, windows))] mod connection_handling_websocket_unix; #[cfg(unix)] mod curated_mcp_sync;