diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index d57f141bf5..ea48d2a4fd 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4332,6 +4332,7 @@ dependencies = [ "tiny_http", "tokio", "tracing", + "tracing-subscriber", "tracing-test", "url", "urlencoding", diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index fb3fd4a21b..c3c0f07db3 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -72,6 +72,7 @@ codex-utils-cargo-bin = { workspace = true } pretty_assertions = { workspace = true } serial_test = { workspace = true } tempfile = { workspace = true } +tracing-subscriber = { workspace = true } tracing-test = { workspace = true, features = ["no-env-filter"] } wiremock = { workspace = true } diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index f942378bb8..84cc008e0f 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -26,7 +26,6 @@ use std::sync::atomic::Ordering; use std::thread::sleep; #[cfg(unix)] use std::thread::spawn; -#[cfg(unix)] use std::time::Duration; use anyhow::Result; @@ -56,6 +55,8 @@ use rmcp::transport::Transport; use tokio::io::AsyncBufReadExt; use tokio::io::BufReader; use tokio::process::Command; +use tokio::sync::watch; +use tokio::time::Instant; use tracing::info; use tracing::warn; @@ -216,6 +217,10 @@ impl StdioServerLauncher for LocalStdioServerLauncher { #[cfg(unix)] const PROCESS_GROUP_TERM_GRACE_PERIOD: Duration = Duration::from_secs(2); +// Keep queued stderr diagnostics before closing the reader, even when an +// escaped descendant prevents the pipe from reaching EOF. +const STDERR_READER_DRAIN_GRACE_PERIOD: Duration = Duration::from_millis(250); + #[cfg(unix)] struct LocalProcessTerminator { process_group_id: u32, @@ -239,6 +244,8 @@ struct StdioServerProcessHandleInner { program_name: String, kind: StdioServerProcessKind, terminated: AtomicBool, + // An escaped descendant can keep stderr open after the MCP server exits. + stderr_reader: Option>, } enum StdioServerProcessKind { @@ -349,25 +356,44 @@ impl LocalStdioServerLauncher { }; #[cfg(not(windows))] let terminator = process_id.map(LocalProcessTerminator::new); - let process = StdioServerProcessHandle::local(program_name.clone(), terminator); - - if let Some(stderr) = stderr { - tokio::spawn(async move { + let stderr_reader = stderr.map(|stderr| { + let program_name = program_name.clone(); + let (stop_tx, mut stop_rx) = watch::channel(()); + std::mem::drop(tokio::spawn(async move { let mut reader = BufReader::new(stderr).lines(); + // Give queued diagnostics time to reach the logs without waiting + // indefinitely for a descendant that still has stderr open. + let drain_deadline = tokio::time::sleep(STDERR_READER_DRAIN_GRACE_PERIOD); + tokio::pin!(drain_deadline); + let mut draining = false; loop { - match reader.next_line().await { - Ok(Some(line)) => { - info!("MCP server stderr ({program_name}): {line}"); - } - Ok(None) => break, - Err(error) => { - warn!("Failed to read MCP server stderr ({program_name}): {error}"); - break; + tokio::select! { + biased; + _ = &mut drain_deadline, if draining => break, + _ = stop_rx.changed(), if !draining => { + draining = true; + drain_deadline.as_mut().reset( + Instant::now() + STDERR_READER_DRAIN_GRACE_PERIOD + ); } + line = reader.next_line() => { + match line { + Ok(Some(line)) => { + info!("MCP server stderr ({program_name}): {line}"); + } + Ok(None) => break, + Err(error) => { + warn!("Failed to read MCP server stderr ({program_name}): {error}"); + break; + } + } + }, } } - }); - } + })); + stop_tx + }); + let process = StdioServerProcessHandle::local(program_name, terminator, stderr_reader); Ok(StdioServerTransport { inner: transport, @@ -428,12 +454,17 @@ impl LocalProcessTerminator { } impl StdioServerProcessHandle { - fn local(program_name: String, terminator: Option) -> Self { + fn local( + program_name: String, + terminator: Option, + stderr_reader: Option>, + ) -> Self { Self { inner: Arc::new(StdioServerProcessHandleInner { program_name, kind: StdioServerProcessKind::Local(terminator), terminated: AtomicBool::new(false), + stderr_reader, }), } } @@ -444,6 +475,7 @@ impl StdioServerProcessHandle { program_name, kind: StdioServerProcessKind::Executor(process), terminated: AtomicBool::new(false), + stderr_reader: None, }), } } @@ -453,7 +485,7 @@ impl StdioServerProcessHandle { return Ok(()); } - match &self.inner.kind { + let result = match &self.inner.kind { StdioServerProcessKind::Local(Some(terminator)) => { terminator.terminate(); Ok(()) @@ -466,7 +498,11 @@ impl StdioServerProcessHandle { Err(io::Error::other(error)) } }, + }; + if let Some(stderr_reader) = &self.inner.stderr_reader { + stderr_reader.send_replace(()); } + result } } @@ -501,6 +537,9 @@ impl Drop for StdioServerProcessHandleInner { })); } } + if let Some(stderr_reader) = &self.stderr_reader { + stderr_reader.send_replace(()); + } } } diff --git a/codex-rs/rmcp-client/tests/stdio_stderr_cleanup.rs b/codex-rs/rmcp-client/tests/stdio_stderr_cleanup.rs new file mode 100644 index 0000000000..fa8b63c48c --- /dev/null +++ b/codex-rs/rmcp-client/tests/stdio_stderr_cleanup.rs @@ -0,0 +1,222 @@ +//! The local stderr reader must close when its MCP transport ends, even when +//! a descendant outside the server process group still has stderr open. + +#![cfg(unix)] + +use std::fs; +use std::io::Write; +use std::path::Path; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use anyhow::Result; +use codex_rmcp_client::ElicitationAction; +use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::LocalStdioServerLauncher; +use codex_rmcp_client::RmcpClient; +use futures::FutureExt as _; +use pretty_assertions::assert_eq; +use rmcp::model::ClientCapabilities; +use rmcp::model::Implementation; +use rmcp::model::InitializeRequestParams; +use rmcp::model::ProtocolVersion; + +struct EscapedChild(i32); + +enum Cleanup { + Shutdown, + Drop, +} + +#[derive(Clone)] +struct TestLogWriter(Arc>>); + +impl Write for TestLogWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 + .lock() + .map_err(|_| std::io::Error::other("log buffer lock"))? + .extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Drop for EscapedChild { + fn drop(&mut self) { + // Only this test's PID, written after setsid(), is eligible for cleanup. + // The fixture also exits independently after 30 seconds. + unsafe { libc::kill(self.0, libc::SIGKILL) }; + } +} + +fn fd_count() -> Result { + let directory = if cfg!(target_os = "linux") { + "/proc/self/fd" + } else { + "/dev/fd" + }; + Ok(fs::read_dir(directory)?.count()) +} + +async fn read_pid(path: &Path) -> Result { + for _ in 0..100 { + if let Ok(text) = fs::read_to_string(path) + && let Ok(pid) = text.trim().parse::() + && pid > 1 + { + return Ok(pid); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + anyhow::bail!("fixture PID unavailable: {}", path.display()) +} + +#[tokio::test(flavor = "current_thread")] +async fn client_teardown_closes_stderr_fd_and_logs_queued_diagnostics() -> Result<()> { + let logs = Arc::new(Mutex::new(Vec::new())); + let writer_logs = Arc::clone(&logs); + // The reader task runs on this thread, so its logs use this subscriber. + let _subscriber_guard = tracing::subscriber::set_default( + tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing::Level::INFO) + .with_writer(move || TestLogWriter(Arc::clone(&writer_logs))) + .finish(), + ); + // FD counts are process-wide, so run the two teardown paths sequentially. + assert_stderr_reader_closed(Cleanup::Shutdown, &logs).await?; + assert_stderr_reader_closed(Cleanup::Drop, &logs).await +} + +async fn assert_stderr_reader_closed(cleanup: Cleanup, logs: &Arc>>) -> Result<()> { + let temporary = tempfile::tempdir()?; + let server_pid_file = temporary.path().join("server.pid"); + let child_pid_file = temporary.path().join("escaped.pid"); + let trigger_file = temporary.path().join("emit-stderr"); + let written_file = temporary.path().join("stderr-written"); + let diagnostic = match &cleanup { + Cleanup::Shutdown => "queued diagnostic before shutdown", + Cleanup::Drop => "queued diagnostic before drop", + }; + let script = temporary.path().join("server.py"); + fs::write( + &script, + r#"import json, os, pathlib, signal, sys, threading, time +# Keep either process bounded if test cleanup fails. +MAX_FIXTURE_LIFETIME_SECONDS = 30 +server_pid_path = pathlib.Path(sys.argv[1]) +escaped_pid_path = pathlib.Path(sys.argv[2]) +trigger_path = pathlib.Path(sys.argv[3]) +written_path = pathlib.Path(sys.argv[4]) +signal.alarm(MAX_FIXTURE_LIFETIME_SECONDS) +server_pid_path.write_text(str(os.getpid())) +if os.fork() == 0: # fork() returns zero in the child. + os.setsid() + signal.alarm(MAX_FIXTURE_LIFETIME_SECONDS) + os.close(sys.stdin.fileno()) + os.close(sys.stdout.fileno()) + escaped_pid_path.write_text(str(os.getpid())) + # Exit even if SIGALRM was ignored; the test normally kills this child sooner. + time.sleep(MAX_FIXTURE_LIFETIME_SECONDS) + os._exit(os.EX_OK) +def emit_diagnostic(): + while not trigger_path.exists(): + time.sleep(0.01) + print(sys.argv[5], file=sys.stderr, flush=True) + written_path.write_text("written") +threading.Thread(target=emit_diagnostic, daemon=True).start() +for line in sys.stdin: + message = json.loads(line) + if message.get("method") == "initialize": + result = {"protocolVersion": "2025-06-18", "capabilities": {}, "serverInfo": {"name": "stderr-reader-fixture", "version": "1"}} + print(json.dumps({"jsonrpc": "2.0", "id": message["id"], "result": result}), flush=True) +"#, + )?; + let baseline = fd_count()?; + let client = RmcpClient::new_stdio_client( + which::which("python3")?.into(), + vec![ + script.into_os_string(), + server_pid_file.clone().into_os_string(), + child_pid_file.clone().into_os_string(), + trigger_file.clone().into_os_string(), + written_file.clone().into_os_string(), + diagnostic.into(), + ], + /*env*/ None, + &[], + /*cwd*/ None, + Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)), + ) + .await?; + let server_pid = read_pid(&server_pid_file).await?; + let escaped = EscapedChild(read_pid(&child_pid_file).await?); + assert_eq!(unsafe { libc::getpgid(server_pid) }, server_pid); + assert_eq!(unsafe { libc::getpgid(escaped.0) }, escaped.0); + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("stderr-reader-test", "1"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + Some(Duration::from_secs(5)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Decline, + content: None, + meta: None, + }) + } + .boxed() + }), + ) + .await?; + // Keep the current-thread Tokio runtime blocked until the server has + // written stderr. The reader cannot run before teardown starts. + fs::write(&trigger_file, "")?; + for _ in 0..250 { + if written_file.exists() { + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + assert!(written_file.exists(), "fixture did not write stderr"); + // For Shutdown, keep the client alive until after the FD assertion so + // final Drop cannot hide missing cleanup in the explicit shutdown path. + if matches!(cleanup, Cleanup::Shutdown) { + client.shutdown().await; + } else { + drop(client); + } + let mut after_shutdown = fd_count()?; + for _ in 0..250 { + if after_shutdown == baseline { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + after_shutdown = fd_count()?; + } + assert_eq!( + after_shutdown, baseline, + "stderr reader still owns a file descriptor while the escaped child is alive" + ); + assert_eq!(unsafe { libc::kill(escaped.0, 0) }, 0); + let captured_logs = String::from_utf8( + logs.lock() + .map_err(|_| anyhow::anyhow!("log buffer lock"))? + .clone(), + )?; + assert!( + captured_logs.contains(diagnostic), + "stderr diagnostic written before teardown was lost: {captured_logs}" + ); + drop(escaped); + Ok(()) +}