From 713caa89f389acd9cbcd77016edbb607273826af Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 10 Sep 2026 12:54:02 +0000 Subject: [PATCH] Bound app-server stdio shutdown and handle Unix SIGTERM gracefully (#44523) ## Why Stdio shutdown can hang on open pipes or blocked I/O, and Unix `SIGTERM` needs to run connection cleanup so owned commands terminate and session-end hooks execute. ## What changed - Handle `SIGTERM` on a dedicated signal thread and close the stdio connection through the existing cleanup path. - Move blocking stdin and stdout operations to dedicated threads so they do not hold the Tokio runtime open. Cancel message forwarding on `SIGTERM` to release blocked producers. - When shutdown signal handling is enabled on Unix, start a shared 45-second watchdog on EOF or `SIGTERM`. Preserve the first deadline and exit with status `1` if cleanup or runtime teardown stalls, even when logging is blocked. ## Testing Add Unix integration coverage for shutdown with open pipes, blocked stdout, blocked stderr, and a stalled filesystem write. Verify command and child-process termination, preserve the EOF deadline when `SIGTERM` follows, and extend session-end hook coverage to `SIGTERM`. GitOrigin-RevId: 43a80b8277eb50519cdc293eb3b5dc2cad00b3aa --- codex-rs/Cargo.lock | 1 + codex-rs/Cargo.toml | 1 + codex-rs/app-server-transport/Cargo.toml | 3 + .../src/transport/stdio.rs | 158 ++++++--- codex-rs/app-server/src/lib.rs | 5 +- .../tests/common/test_app_server.rs | 15 + .../v2/connection_handling_stdio_tests.rs | 329 ++++++++++++++++++ codex-rs/app-server/tests/suite/v2/mod.rs | 3 + .../app-server/tests/suite/v2/session_end.rs | 31 +- 9 files changed, 501 insertions(+), 45 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/connection_handling_stdio_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index d735f833ad..30a9c2f963 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2411,6 +2411,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", + "signal-hook", "tempfile", "time", "tokio", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index e07725207c..788afe6e80 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -451,6 +451,7 @@ sha1 = "0.10.6" scopeguard = "1.2.0" sha2 = "0.10" shlex = "1.3.0" +signal-hook = "0.3.18" similar = "2.7.0" symphonia = { version = "0.6.0", default-features = false, features = [ "isomp4", diff --git a/codex-rs/app-server-transport/Cargo.toml b/codex-rs/app-server-transport/Cargo.toml index b6c3241307..aa8150de07 100644 --- a/codex-rs/app-server-transport/Cargo.toml +++ b/codex-rs/app-server-transport/Cargo.toml @@ -55,6 +55,9 @@ tracing = { workspace = true, features = ["log"] } url = { workspace = true } uuid = { workspace = true, features = ["serde", "v7"] } +[target.'cfg(unix)'.dependencies] +signal-hook = { workspace = true } + [dev-dependencies] chrono = { workspace = true } codex-config = { workspace = true } diff --git a/codex-rs/app-server-transport/src/transport/stdio.rs b/codex-rs/app-server-transport/src/transport/stdio.rs index c2efdaea71..12552800cf 100644 --- a/codex-rs/app-server-transport/src/transport/stdio.rs +++ b/codex-rs/app-server-transport/src/transport/stdio.rs @@ -1,3 +1,8 @@ +//! Stdio transport with EOF cleanup and Unix SIGTERM shutdown. Dedicated I/O +//! threads let SIGTERM abandon blocked pipes without holding the runtime open. +//! A dedicated signal thread arms the shared EOF/SIGTERM deadline even when +//! logging or the Tokio runtime is blocked. + use super::CHANNEL_CAPACITY; use super::ConnectionOrigin; use super::TransportEvent; @@ -8,24 +13,42 @@ use crate::outgoing_message::QueuedOutgoingMessage; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCRequest; +use std::io::BufRead; use std::io::ErrorKind; use std::io::Result as IoResult; -use tokio::io; -use tokio::io::AsyncBufReadExt; -use tokio::io::AsyncWriteExt; -use tokio::io::BufReader; +use std::io::Write; use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use tracing::debug; use tracing::error; use tracing::info; pub async fn start_stdio_connection( transport_event_tx: mpsc::Sender, - stdio_handles: &mut Vec>, initialize_client_name_tx: oneshot::Sender, -) -> IoResult<()> { + install_shutdown_signal_handler: bool, +) -> IoResult> { + let shutdown_signal = CancellationToken::new(); + #[cfg(unix)] + if install_shutdown_signal_handler { + use signal_hook::consts::SIGTERM; + use signal_hook::iterator::Signals; + + // Register before accepting requests. Receiving SIGTERM and arming the + // watchdog must not depend on Tokio or synchronous transport logging. + let mut signals = Signals::new([SIGTERM])?; + let shutdown_signal = shutdown_signal.clone(); + std::thread::Builder::new() + .name("app-server-signal".to_string()) + .spawn(move || { + if signals.forever().next().is_some() { + start_shutdown_watchdog(); + shutdown_signal.cancel(); + } + })?; + } let connection_id = next_connection_id(); let (writer_tx, mut writer_rx) = mpsc::channel::(CHANNEL_CAPACITY); let writer_tx_for_reader = writer_tx.clone(); @@ -40,55 +63,84 @@ pub async fn start_stdio_connection( .await .map_err(|_| std::io::Error::new(ErrorKind::BrokenPipe, "processor unavailable"))?; - let transport_event_tx_for_reader = transport_event_tx.clone(); - stdio_handles.push(tokio::spawn(async move { - let stdin = io::stdin(); - let reader = BufReader::new(stdin); - let mut lines = reader.lines(); - let mut initialize_client_name_tx = Some(initialize_client_name_tx); - - loop { - match lines.next_line().await { - Ok(Some(line)) => { - if let Some(client_name) = stdio_initialize_client_name(&line) - && let Some(initialize_client_name_tx) = initialize_client_name_tx.take() - { - let _ = initialize_client_name_tx.send(client_name); - } - if !forward_incoming_message( - &transport_event_tx_for_reader, - &writer_tx_for_reader, - connection_id, - &line, - ) - .await - { - break; - } + // Tokio's stdin uses an uncancellable blocking task, which keeps its runtime + // alive while the client leaves stdin open. This process-owned thread may + // remain blocked until process exit, but is not joined by the runtime. + let (stdin_tx, mut stdin_rx) = mpsc::channel(/*buffer*/ 1); + std::thread::Builder::new() + .name("app-server-stdin".to_string()) + .spawn(move || { + for line in std::io::stdin().lock().lines() { + if stdin_tx.blocking_send(line).is_err() { + break; } - Ok(None) => break, + } + })?; + + // Keep stdout's blocking writes off Tokio's pool too. The forwarding future + // owns writer_rx so cancelling it also releases producers stuck on a full queue. + let (stdout_tx, mut stdout_rx) = + mpsc::channel::<(String, oneshot::Sender<()>)>(/*buffer*/ 1); + std::thread::Builder::new() + .name("app-server-stdout".to_string()) + .spawn(move || { + let mut stdout = std::io::stdout().lock(); + while let Some((json, written_tx)) = stdout_rx.blocking_recv() { + if let Err(err) = stdout.write_all(json.as_bytes()) { + error!("Failed to write to stdout: {err}"); + break; + } + let _ = written_tx.send(()); + } + })?; + + let transport_event_tx_for_reader = transport_event_tx.clone(); + let read_messages = async move { + let mut initialize_client_name_tx = Some(initialize_client_name_tx); + while let Some(line) = stdin_rx.recv().await { + let line = match line { + Ok(line) => line, Err(err) => { error!("Failed reading stdin: {err}"); break; } + }; + if let Some(client_name) = stdio_initialize_client_name(&line) + && let Some(initialize_client_name_tx) = initialize_client_name_tx.take() + { + let _ = initialize_client_name_tx.send(client_name); + } + if !forward_incoming_message( + &transport_event_tx_for_reader, + &writer_tx_for_reader, + connection_id, + &line, + ) + .await + { + break; } } + // EOF can finish the transport before RPC or runtime cleanup. Start + // the same process deadline even if no SIGTERM arrives. + if cfg!(unix) && install_shutdown_signal_handler { + start_shutdown_watchdog(); + } let _ = transport_event_tx_for_reader .send(TransportEvent::ConnectionClosed { connection_id }) .await; debug!("stdin reader finished (EOF)"); - })); + }; - stdio_handles.push(tokio::spawn(async move { - let mut stdout = io::stdout(); + let write_messages = async move { while let Some(queued_message) = writer_rx.recv().await { let Some(mut json) = serialize_outgoing_message(queued_message.message) else { continue; }; json.push('\n'); - if let Err(err) = stdout.write_all(json.as_bytes()).await { - error!("Failed to write to stdout: {err}"); + let (written_tx, written_rx) = oneshot::channel(); + if stdout_tx.send((json, written_tx)).await.is_err() || written_rx.await.is_err() { break; } if let Some(write_complete_tx) = queued_message.write_complete_tx { @@ -96,9 +148,37 @@ pub async fn start_stdio_connection( } } info!("stdout writer exited (channel closed)"); - })); + }; - Ok(()) + Ok(tokio::spawn(async move { + tokio::select! { + _ = shutdown_signal.cancelled() => { + // Cancelling both forwarding futures drops their queues before + // connection teardown, including when EOF already began draining. + info!("SIGTERM received; closing stdio connection (45s shutdown deadline)"); + let _ = transport_event_tx + .send(TransportEvent::ConnectionClosed { connection_id }) + .await; + } + _ = async move { tokio::join!(read_messages, write_messages); } => {} + } + })) +} + +fn start_shutdown_watchdog() { + // EOF and SIGTERM can both start cleanup; keep the first process deadline. + static STARTED: std::sync::Once = std::sync::Once::new(); + STARTED.call_once(|| { + std::thread::Builder::new() + .name("app-server-shutdown".to_string()) + .spawn(|| { + // Allow the processor's 30s RPC drain, then bound even Tokio's + // runtime teardown. Do not log here: stderr may also be blocked. + std::thread::sleep(std::time::Duration::from_secs(45)); + std::process::exit(/*code*/ 1); + }) + .unwrap_or_else(|_| std::process::exit(/*code*/ 1)); + }); } fn stdio_initialize_client_name(line: &str) -> Option { diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index b48813de2b..aa565b6b34 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -762,12 +762,13 @@ pub async fn run_main_with_transport_options( AppServerTransport::Stdio => { let (stdio_client_name_tx, stdio_client_name_rx) = oneshot::channel::(); app_server_client_name_rx = Some(stdio_client_name_rx); - start_stdio_connection( + let handle = start_stdio_connection( transport_event_tx.clone(), - &mut transport_accept_handles, stdio_client_name_tx, + runtime_options.install_shutdown_signal_handler, ) .await?; + transport_accept_handles.push(handle); } AppServerTransport::UnixSocket { socket_path } => { let accept_handle = start_control_socket_acceptor( diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index 048f2eeb8e..568916e93b 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -192,6 +192,21 @@ impl TestAppServer { self.process.wait().await } + #[cfg(unix)] + pub fn send_sigterm(&self) -> anyhow::Result<()> { + let pid = self.process.id().context("app-server has no pid")?; + let status = std::process::Command::new("kill") + .args(["-TERM", &pid.to_string()]) + .status()?; + ensure!(status.success(), "failed to signal app-server: {status}"); + Ok(()) + } + + /// Waits for output without consuming it, for transport backpressure tests. + pub async fn peek_stdout(&mut self) -> std::io::Result<&[u8]> { + self.stdout.fill_buf().await + } + /// Closes stdio and waits for app-server's graceful thread teardown to finish. pub async fn shutdown_gracefully(&mut self) -> std::io::Result { drop(self.stdin.take()); diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_stdio_tests.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_stdio_tests.rs new file mode 100644 index 0000000000..434335f3e4 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_stdio_tests.rs @@ -0,0 +1,329 @@ +//! Stdio shutdown cleans up owned commands and bounds teardown when I/O stalls. + +use anyhow::Context; +use anyhow::Result; +use app_test_support::DISABLE_PLUGIN_STARTUP_TASKS_ARG; +use app_test_support::TestAppServer; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use codex_app_server_protocol::CommandExecParams; +use codex_app_server_protocol::FsReadFileParams; +use codex_app_server_protocol::FsWriteFileParams; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::SandboxPolicy; +use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_cargo_bin::cargo_bin; +use core_test_support::skip_if_remote; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::io::ErrorKind; +use std::io::Read; +use std::process::Command; +use std::process::Stdio; +use std::time::Instant; +use tempfile::TempDir; +use test_case::test_case; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::net::unix::pipe; +use tokio::time::Duration; +use tokio::time::sleep; +use tokio::time::timeout; + +#[tokio::test] +async fn stdio_sigterm_exits_with_open_pipes() -> Result<()> { + let mut app = TestAppServer::builder().build_initialized().await?; + + app.send_sigterm()?; + // wait_for_exit leaves both pipes open and does not drain stdout. Success catches an + // unhandled SIGTERM, which exits before any cleanup can run. + let status = timeout(Duration::from_secs(10), app.wait_for_exit()) + .await + .context("app-server did not exit after SIGTERM")??; + assert!( + status.success(), + "app-server did not exit cleanly: {status}" + ); + Ok(()) +} + +#[tokio::test] +async fn stdio_sigterm_times_out_with_blocked_stderr() -> Result<()> { + // TestAppServer drains stderr, so keep this subprocess's stderr pipe unread. + let codex_home = TempDir::new()?; + let mut process = tokio::process::Command::new(cargo_bin("codex-app-server")?) + .arg(DISABLE_PLUGIN_STARTUP_TASKS_ARG) + .env("CODEX_HOME", codex_home.path()) + .env( + "CODEX_APP_SERVER_MANAGED_CONFIG_PATH", + codex_home.path().join("managed_config.toml"), + ) + .env("RUST_LOG", "codex_app_server_transport=error") + .current_dir(codex_home.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let mut stdin = process.stdin.take().context("app-server has no stdin")?; + let mut stdout = + BufReader::new(process.stdout.take().context("app-server has no stdout")?).lines(); + let initialize = json!({ + "id": 1, + "method": "initialize", + "params": { "clientInfo": { "name": "stdio-stderr-test", "version": "1" } } + }); + stdin + .write_all(format!("{initialize}\n").as_bytes()) + .await?; + let response = timeout(Duration::from_secs(10), stdout.next_line()) + .await?? + .context("app-server did not initialize")?; + assert!(matches!( + serde_json::from_str::(&response)?, + JSONRPCMessage::Response(_) + )); + + // Invalid JSON logs synchronously in the reader. Fill stderr until the + // transport stops consuming stdin, then keep all three pipes open. + let invalid_input = b"{\n".repeat(1024 * 1024); + assert!( + timeout(Duration::from_secs(2), stdin.write_all(&invalid_input)) + .await + .is_err(), + "expected unread stderr to block the transport" + ); + let signalled_at = Instant::now(); + let status = Command::new("kill") + .arg("-TERM") + .arg(process.id().context("app-server has no pid")?.to_string()) + .status()?; + assert!(status.success(), "failed to signal app-server: {status}"); + let status = timeout(Duration::from_secs(55), process.wait()) + .await + .context("blocked stderr prevented SIGTERM shutdown")??; + assert_eq!(status.code(), Some(1)); + assert!(signalled_at.elapsed() >= Duration::from_secs(45)); + drop(stdin); + Ok(()) +} + +#[test_case(Shutdown::Sigterm; "sigterm")] +#[test_case(Shutdown::Eof; "eof_then_sigterm")] +#[tokio::test] +async fn stdio_shutdown_times_out_with_blocked_file_write(shutdown: Shutdown) -> Result<()> { + let codex_home = TempDir::new()?; + let path = codex_home.path().join("blocked-write"); + let status = Command::new("mkfifo").arg(&path).status()?; + assert!(status.success(), "failed to create FIFO: {status}"); + let mut reader = std::fs::File::from( + pipe::OpenOptions::new() + .open_receiver(&path)? + .into_nonblocking_fd()?, + ); + let mut app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + // The FIFO belongs to the app-server's local filesystem, even in remote CI. + .without_auto_env() + .build_initialized() + .await?; + app.send_fs_write_file_request(FsWriteFileParams { + path: AbsolutePathBuf::try_from(path)?, + data_base64: STANDARD.encode(vec![b'x'; 1024 * 1024]), + }) + .await?; + + // Confirm the write has started, then leave the FIFO unread. The blocking + // filesystem write cannot finish, even if its async request is cancelled. + let mut first_byte = [0]; + timeout(Duration::from_secs(10), async { + loop { + match reader.read(&mut first_byte) { + Ok(0) => {} + Ok(_) => break, + Err(err) if err.kind() == ErrorKind::WouldBlock => {} + Err(err) => return Err(err), + } + sleep(Duration::from_millis(20)).await; + } + Ok::<_, std::io::Error>(()) + }) + .await + .context("filesystem write did not start")??; + assert_eq!(first_byte, [b'x']); + + let shutdown_at = Instant::now(); + match shutdown { + Shutdown::Sigterm => app.send_sigterm()?, + Shutdown::Eof => { + // Close stdin, then let the 30s RPC drain finish and enter runtime + // teardown before SIGTERM. Keep the FIFO blocked throughout. + assert!( + timeout(Duration::from_secs(35), app.shutdown_gracefully()) + .await + .is_err(), + "app-server exited before allowing graceful cleanup" + ); + app.send_sigterm()?; + } + } + let remaining = Duration::from_secs(55).saturating_sub(shutdown_at.elapsed()); + let status = timeout(remaining, app.wait_for_exit()) + .await + .context("app-server did not exit within the shutdown deadline")??; + assert_eq!(status.code(), Some(1)); + assert!( + shutdown_at.elapsed() >= Duration::from_secs(45), + "app-server exited before allowing graceful cleanup" + ); + drop(reader); + Ok(()) +} + +#[derive(Clone, Copy)] +enum Shutdown { + Eof, + Sigterm, +} + +#[test_case(Shutdown::Eof; "eof")] +#[test_case(Shutdown::Sigterm; "sigterm")] +#[tokio::test] +async fn stdio_shutdown_terminates_commands_and_children(shutdown: Shutdown) -> Result<()> { + skip_if_remote!( + Ok(()), + "asserts that processes owned by the local app-server exit" + ); + let workspace = TempDir::new()?; + let mut app = TestAppServer::builder().build_initialized().await?; + + let script = r#"sleep 120 & child=$!; printf '%s %s\n' "$$" "$child" > pids; wait "$child""#; + app.send_command_exec_request(CommandExecParams { + command: vec!["/bin/sh".to_string(), "-c".to_string(), script.to_string()], + process_id: Some("shutdown-test".to_string()), + tty: false, + stream_stdin: false, + stream_stdout_stderr: true, + output_bytes_cap: None, + disable_output_cap: false, + disable_timeout: true, + timeout_ms: None, + cwd: Some(workspace.path().to_path_buf()), + env: None, + size: None, + sandbox_policy: Some(SandboxPolicy::DangerFullAccess), + permission_profile: None, + }) + .await?; + + let mut children = ProcessCleanup(Vec::new()); + timeout(Duration::from_secs(10), async { + loop { + if let Ok(pids) = tokio::fs::read_to_string(workspace.path().join("pids")).await { + let pids = pids + .split_whitespace() + .map(str::parse::) + .collect::, _>>()?; + if pids.len() == 2 { + children.0 = pids; + break; + } + } + sleep(Duration::from_millis(20)).await; + } + Ok::<_, anyhow::Error>(()) + }) + .await + .context("command did not start")??; + children.0.sort_unstable(); + assert_eq!(running_process_ids(&children.0)?, children.0); + + if matches!(shutdown, Shutdown::Sigterm) { + let path = workspace.path().join("large-response"); + std::fs::write(&path, vec![b'x'; 1024 * 1024])?; + app.send_fs_read_file_request(FsReadFileParams { + path: AbsolutePathBuf::try_from(path)?, + }) + .await?; + timeout(Duration::from_secs(10), async { + while !String::from_utf8_lossy(app.peek_stdout().await?).contains("dataBase64") { + app.read_next_message().await?; + } + Ok::<_, anyhow::Error>(()) + }) + .await + .context("large response did not start")??; + // Leave the response unread so stdout remains blocked during shutdown. + } + + let status = match shutdown { + Shutdown::Eof => timeout(Duration::from_secs(10), app.shutdown_gracefully()).await??, + Shutdown::Sigterm => { + app.send_sigterm()?; + timeout(Duration::from_secs(10), app.wait_for_exit()) + .await + .context("app-server did not exit after SIGTERM")?? + } + }; + assert!( + status.success(), + "app-server did not exit cleanly: {status}" + ); + + timeout(Duration::from_secs(5), async { + loop { + if running_process_ids(&children.0)?.is_empty() { + children.0.clear(); + return Ok::<_, anyhow::Error>(()); + } + sleep(Duration::from_millis(20)).await; + } + }) + .await + .context("command or its child survived app-server shutdown")??; + Ok(()) +} + +fn running_process_ids(pids: &[u32]) -> Result> { + let output = Command::new("ps") + .args(["-o", "pid=,stat=", "-p"]) + .arg( + pids.iter() + .map(u32::to_string) + .collect::>() + .join(","), + ) + .output()?; + anyhow::ensure!( + output.status.success() || output.status.code() == Some(1), + "ps failed: {}", + String::from_utf8_lossy(&output.stderr), + ); + let mut running = Vec::new(); + for line in String::from_utf8(output.stdout)?.lines() { + let mut fields = line.split_whitespace(); + let pid = fields.next().context("missing process id")?.parse()?; + let state = fields.next().context("missing process state")?; + // An orphaned zombie has exited but may not yet have been reaped + // by init (notably in Linux test containers). + if !state.starts_with('Z') { + running.push(pid); + } + } + running.sort_unstable(); + Ok(running) +} + +struct ProcessCleanup(Vec); + +impl Drop for ProcessCleanup { + fn drop(&mut self) { + for pid in &self.0 { + let _ = Command::new("kill") + .arg("-KILL") + .arg(pid.to_string()) + .output(); + } + } +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 4846897d71..67fe1b9541 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -18,6 +18,9 @@ mod config_requirements_application; mod config_requirements_browser_use; mod config_requirements_in_app_browser; mod config_rpc; +#[cfg(unix)] +#[path = "connection_handling_stdio_tests.rs"] +mod connection_handling_stdio; mod connection_handling_websocket; #[cfg(unix)] mod connection_handling_websocket_unix; diff --git a/codex-rs/app-server/tests/suite/v2/session_end.rs b/codex-rs/app-server/tests/suite/v2/session_end.rs index 93592f2967..2a478aacb1 100644 --- a/codex-rs/app-server/tests/suite/v2/session_end.rs +++ b/codex-rs/app-server/tests/suite/v2/session_end.rs @@ -21,6 +21,7 @@ use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use tempfile::TempDir; +use test_case::test_case; use tokio::time::timeout; const READ_TIMEOUT: Duration = Duration::from_secs(20); @@ -94,8 +95,18 @@ async fn run_removal_session_end_test(operation: &str) -> Result<()> { Ok(()) } +#[derive(Clone, Copy)] +enum Shutdown { + Eof, + Sigterm, +} + +#[test_case(Shutdown::Eof; "eof")] +#[test_case(Shutdown::Sigterm; "sigterm")] #[tokio::test] -async fn app_server_shutdown_runs_session_end_for_all_loaded_threads() -> Result<()> { +async fn app_server_shutdown_runs_session_end_for_all_loaded_threads( + shutdown: Shutdown, +) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; let log_path = write_config_and_hook(codex_home.path(), &server.uri())?; @@ -106,15 +117,27 @@ async fn app_server_shutdown_runs_session_end_for_all_loaded_threads() -> Result let first = start_thread(&mut app_server).await?; let second = start_thread(&mut app_server).await?; - let status = timeout(READ_TIMEOUT, app_server.shutdown_gracefully()).await??; + let status = match shutdown { + Shutdown::Eof => timeout(READ_TIMEOUT, app_server.shutdown_gracefully()).await??, + Shutdown::Sigterm => { + app_server.send_sigterm()?; + timeout(READ_TIMEOUT, app_server.wait_for_exit()).await?? + } + }; assert!(status.success(), "app-server did not exit successfully"); let mut actual = read_hook_log(&log_path)? .into_iter() .map(|payload| { ( - payload["session_id"].as_str().unwrap().to_string(), - payload["reason"].as_str().unwrap().to_string(), + payload["session_id"] + .as_str() + .expect("session end session ID") + .to_string(), + payload["reason"] + .as_str() + .expect("session end reason") + .to_string(), ) }) .collect::>();