From ea7e0714daeee7ebad9a0061ed44483a8c2a50e8 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Tue, 1 Sep 2026 18:43:19 +0000 Subject: [PATCH] Fix relative MCP server spawning on macOS (#42117) ## Why Rust falls back to `fork` when a macOS command combines a relative executable path with a working directory, due to a historical `posix_spawnp` issue. This can make local MCP server startup less reliable. ## What changed - Spawn relative macOS MCP executables directly with `posix_spawn`, preserving the configured path, `argv[0]`, working directory, environment, stdio, and process group. - Keep the existing launcher for `PATH` lookup and executable text files without a shebang. - Use one local child wrapper for legacy and 2026-07-28 protocol framing, with cancellation-safe waiting and cleanup that kills and reaps dropped children. ## Testing Add macOS regression coverage for relative scripts, argument and descriptor preservation, launch errors, cancelled waits, and cleanup after runtime shutdown. GitOrigin-RevId: 1cbc6b90aed6f67e0a38f551964d80496ebc4471 --- codex-rs/rmcp-client/Cargo.toml | 1 + .../src/bounded_stdio_transport.rs | 132 ++++++ .../rmcp-client/src/http_client_adapter.rs | 2 +- codex-rs/rmcp-client/src/lib.rs | 2 + codex-rs/rmcp-client/src/local_child.rs | 34 ++ .../rmcp-client/src/local_stdio_transport.rs | 157 ++------ codex-rs/rmcp-client/src/macos_stdio.rs | 376 ++++++++++++++++++ codex-rs/rmcp-client/src/macos_stdio_tests.rs | 211 ++++++++++ .../rmcp-client/src/stdio_server_launcher.rs | 48 +-- codex-rs/rmcp-client/tests/mcp_2026_stdio.rs | 40 +- .../tests/process_group_cleanup.rs | 18 +- 11 files changed, 855 insertions(+), 166 deletions(-) create mode 100644 codex-rs/rmcp-client/src/bounded_stdio_transport.rs create mode 100644 codex-rs/rmcp-client/src/local_child.rs create mode 100644 codex-rs/rmcp-client/src/macos_stdio.rs create mode 100644 codex-rs/rmcp-client/src/macos_stdio_tests.rs diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index a24df4a4a2..d567c52fec 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -54,6 +54,7 @@ tokio = { workspace = true, features = [ "io-util", "macros", "process", + "signal", "rt-multi-thread", "sync", "io-std", diff --git a/codex-rs/rmcp-client/src/bounded_stdio_transport.rs b/codex-rs/rmcp-client/src/bounded_stdio_transport.rs new file mode 100644 index 0000000000..fdde9f698e --- /dev/null +++ b/codex-rs/rmcp-client/src/bounded_stdio_transport.rs @@ -0,0 +1,132 @@ +//! Bounded, compatibility-aware transport for modern locally spawned MCP servers. + +use std::future::Future; +use std::io; +use std::sync::Arc; + +use memchr::memchr; +use rmcp::service::RoleClient; +use rmcp::service::RxJsonRpcMessage; +use rmcp::service::TxJsonRpcMessage; +use rmcp::transport::Transport; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::process::ChildStdin; +use tokio::process::ChildStdout; +use tokio::sync::Mutex; +use tracing::debug; +use tracing::warn; + +/// Match the existing executor stdio transport and the production MCP limit. +pub(crate) const MAX_MCP_STDIO_LINE_BYTES: usize = 8 * 1024 * 1024; + +pub(super) struct BoundedStdioTransport { + stdin: Arc>>, + stdout: BufReader, + pending_line: Vec, + program_name: String, +} + +impl BoundedStdioTransport { + pub(super) fn new(stdin: ChildStdin, stdout: ChildStdout, program_name: String) -> Self { + Self { + stdin: Arc::new(Mutex::new(Some(stdin))), + stdout: BufReader::new(stdout), + pending_line: Vec::new(), + program_name, + } + } + + async fn receive_message(&mut self) -> Option> { + loop { + let bytes = match self.stdout.fill_buf().await { + Ok(bytes) => bytes, + Err(error) => { + warn!( + "Failed to read MCP server stdout ({}): {error}", + self.program_name + ); + return None; + } + }; + + if bytes.is_empty() { + if self.pending_line.is_empty() { + return None; + } + return self.decode_pending_message(); + } + + let newline = memchr(b'\n', bytes); + let content_len = newline.unwrap_or(bytes.len()); + if content_len > MAX_MCP_STDIO_LINE_BYTES.saturating_sub(self.pending_line.len()) { + warn!( + "MCP stdio line exceeds {MAX_MCP_STDIO_LINE_BYTES} bytes ({}); closing transport", + self.program_name + ); + self.pending_line.clear(); + return None; + } + + self.pending_line.extend_from_slice(&bytes[..content_len]); + let consumed = content_len + usize::from(newline.is_some()); + self.stdout.consume(consumed); + + if newline.is_some() + && let Some(message) = self.decode_pending_message() + { + return Some(message); + } + } + } + + fn decode_pending_message(&mut self) -> Option> { + let line = std::mem::take(&mut self.pending_line); + let line = line.strip_suffix(b"\r").unwrap_or(&line); + match serde_json::from_slice(line) { + Ok(message) => Some(message), + Err(error) => { + debug!( + "Failed to parse local MCP server message ({}): {error}", + self.program_name + ); + None + } + } + } +} + +impl Transport for BoundedStdioTransport { + type Error = io::Error; + + #[expect( + clippy::await_holding_invalid_type, + reason = "complete JSON-RPC frames must hold the stdin lock across writes" + )] + fn send( + &mut self, + item: TxJsonRpcMessage, + ) -> impl Future> + Send + 'static { + let stdin = Arc::clone(&self.stdin); + async move { + let mut message = serde_json::to_vec(&item).map_err(io::Error::other)?; + message.push(b'\n'); + let mut guard = stdin.lock().await; + let stdin = guard + .as_mut() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "MCP stdin closed"))?; + stdin.write_all(&message).await?; + stdin.flush().await + } + } + + fn receive(&mut self) -> impl Future>> + Send { + self.receive_message() + } + + async fn close(&mut self) -> Result<(), Self::Error> { + self.stdin.lock().await.take(); + Ok(()) + } +} diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index b887a682b4..0371341aaa 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -55,9 +55,9 @@ use sse_stream::Sse; use sse_stream::SseStream; use tokio::sync::oneshot; +use crate::bounded_stdio_transport::MAX_MCP_STDIO_LINE_BYTES; use crate::event_notification_transport::MAX_EVENT_NOTIFICATION_BYTES; use crate::http_client_redirect::SameOriginRedirectHttpClient; -use crate::local_stdio_transport::MAX_MCP_STDIO_LINE_BYTES; use crate::www_authenticate::insufficient_scope_challenge; diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index 3ca80072e6..0ca24001a9 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -1,4 +1,5 @@ mod auth_status; +mod bounded_stdio_transport; mod elicitation_client_service; mod ema_auth_policy; mod ema_claims; @@ -10,6 +11,7 @@ mod http_client_adapter; mod http_client_redirect; mod http_headers; mod in_process_transport; +mod local_child; mod local_stdio_transport; mod logging_client_handler; mod oauth; diff --git a/codex-rs/rmcp-client/src/local_child.rs b/codex-rs/rmcp-client/src/local_child.rs new file mode 100644 index 0000000000..fdb98303e5 --- /dev/null +++ b/codex-rs/rmcp-client/src/local_child.rs @@ -0,0 +1,34 @@ +//! Uniform child-process API for local MCP servers, with platform-specific spawning. +//! +//! Commands must use the launcher's cleared environment, process group, and default +//! argv[0]. Both implementations expose Tokio stdio handles and kill on drop. + +use std::io; +use std::process::Stdio; + +use tokio::process::Command; + +#[cfg(target_os = "macos")] +#[path = "macos_stdio.rs"] +mod macos; + +#[cfg(target_os = "macos")] +pub(super) use macos::LocalChild; +#[cfg(not(target_os = "macos"))] +pub(super) use tokio::process::Child as LocalChild; + +pub(super) fn spawn(mut command: Command) -> io::Result { + command + .kill_on_drop(true) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + #[cfg(target_os = "macos")] + { + LocalChild::spawn(command) + } + #[cfg(not(target_os = "macos"))] + { + command.spawn() + } +} diff --git a/codex-rs/rmcp-client/src/local_stdio_transport.rs b/codex-rs/rmcp-client/src/local_stdio_transport.rs index 40e0e72946..b94c42584b 100644 --- a/codex-rs/rmcp-client/src/local_stdio_transport.rs +++ b/codex-rs/rmcp-client/src/local_stdio_transport.rs @@ -1,49 +1,46 @@ -//! Bounded, compatibility-aware transport for modern locally spawned MCP servers. +//! Protocol framing and child lifetime for locally spawned MCP servers. +//! +//! Process creation is platform-specific; protocol selection and shutdown are shared. use std::future::Future; use std::io; -use std::process::Stdio; -use std::sync::Arc; use std::time::Duration; -use memchr::memchr; +use futures::FutureExt; use rmcp::service::RoleClient; use rmcp::service::RxJsonRpcMessage; use rmcp::service::TxJsonRpcMessage; use rmcp::transport::Transport; -use tokio::io::AsyncBufReadExt; -use tokio::io::AsyncWriteExt; -use tokio::io::BufReader; -use tokio::process::Child; +use rmcp::transport::async_rw::AsyncRwTransport; use tokio::process::ChildStderr; use tokio::process::ChildStdin; use tokio::process::ChildStdout; use tokio::process::Command; -use tokio::sync::Mutex; -use tracing::debug; -use tracing::warn; -/// Match the existing executor stdio transport and the production MCP limit. -pub(crate) const MAX_MCP_STDIO_LINE_BYTES: usize = 8 * 1024 * 1024; +use crate::bounded_stdio_transport::BoundedStdioTransport; +use crate::local_child; +use crate::local_child::LocalChild; +use crate::protocol_mode::McpProtocolMode; pub(super) struct LocalStdioTransport { - child: Child, - stdin: Arc>>, - stdout: BufReader, - pending_line: Vec, - program_name: String, + child: LocalChild, + transport: StdioTransport, +} + +enum StdioTransport { + /// Preserve rmcp's existing framing for servers using the initialize handshake. + Legacy(AsyncRwTransport), + /// Bound frames and skip messages unknown to the client during 2026-07-28 discovery. + V20260728(BoundedStdioTransport), } impl LocalStdioTransport { pub(super) fn spawn( - mut command: Command, + command: Command, program_name: String, + protocol_mode: McpProtocolMode, ) -> io::Result<(Self, Option)> { - command - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()); - let mut child = command.spawn()?; + let mut child = local_child::spawn(command)?; let stdin = child .stdin .take() @@ -53,117 +50,47 @@ impl LocalStdioTransport { .take() .ok_or_else(|| io::Error::other("MCP server stdout was not piped"))?; let stderr = child.stderr.take(); - - Ok(( - Self { - child, - stdin: Arc::new(Mutex::new(Some(stdin))), - stdout: BufReader::new(stdout), - pending_line: Vec::new(), - program_name, - }, - stderr, - )) + let transport = match protocol_mode { + McpProtocolMode::Legacy => StdioTransport::Legacy(AsyncRwTransport::new(stdout, stdin)), + McpProtocolMode::V20260728 => { + StdioTransport::V20260728(BoundedStdioTransport::new(stdin, stdout, program_name)) + } + }; + Ok((Self { child, transport }, stderr)) } pub(super) fn id(&self) -> Option { self.child.id() } - - async fn receive_message(&mut self) -> Option> { - loop { - let bytes = match self.stdout.fill_buf().await { - Ok(bytes) => bytes, - Err(error) => { - warn!( - "Failed to read MCP server stdout ({}): {error}", - self.program_name - ); - return None; - } - }; - - if bytes.is_empty() { - if self.pending_line.is_empty() { - return None; - } - return self.decode_pending_message(); - } - - let newline = memchr(b'\n', bytes); - let content_len = newline.unwrap_or(bytes.len()); - if content_len > MAX_MCP_STDIO_LINE_BYTES.saturating_sub(self.pending_line.len()) { - warn!( - "MCP stdio line exceeds {MAX_MCP_STDIO_LINE_BYTES} bytes ({}); closing transport", - self.program_name - ); - self.pending_line.clear(); - return None; - } - - self.pending_line.extend_from_slice(&bytes[..content_len]); - let consumed = content_len + usize::from(newline.is_some()); - self.stdout.consume(consumed); - - if newline.is_some() - && let Some(message) = self.decode_pending_message() - { - return Some(message); - } - } - } - - fn decode_pending_message(&mut self) -> Option> { - let line = std::mem::take(&mut self.pending_line); - let line = line.strip_suffix(b"\r").unwrap_or(&line); - match serde_json::from_slice(line) { - Ok(message) => Some(message), - Err(error) => { - debug!( - "Failed to parse local MCP server message ({}): {error}", - self.program_name - ); - None - } - } - } } impl Transport for LocalStdioTransport { type Error = io::Error; - #[expect( - clippy::await_holding_invalid_type, - reason = "complete JSON-RPC frames must hold the stdin lock across writes" - )] fn send( &mut self, item: TxJsonRpcMessage, - ) -> impl Future> + Send + 'static { - let stdin = Arc::clone(&self.stdin); - async move { - let mut message = serde_json::to_vec(&item).map_err(io::Error::other)?; - message.push(b'\n'); - let mut guard = stdin.lock().await; - let stdin = guard - .as_mut() - .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "MCP stdin closed"))?; - stdin.write_all(&message).await?; - stdin.flush().await + ) -> impl Future> + Send + 'static { + match &mut self.transport { + StdioTransport::Legacy(transport) => transport.send(item).boxed(), + StdioTransport::V20260728(transport) => transport.send(item).boxed(), } } fn receive(&mut self) -> impl Future>> + Send { - self.receive_message() + match &mut self.transport { + StdioTransport::Legacy(transport) => transport.receive().boxed(), + StdioTransport::V20260728(transport) => transport.receive().boxed(), + } } - async fn close(&mut self) -> Result<(), Self::Error> { - self.stdin.lock().await.take(); + async fn close(&mut self) -> io::Result<()> { + match &mut self.transport { + StdioTransport::Legacy(transport) => transport.close().await?, + StdioTransport::V20260728(transport) => transport.close().await?, + } match tokio::time::timeout(Duration::from_secs(3), self.child.wait()).await { - Ok(status) => { - status?; - Ok(()) - } + Ok(status) => status.map(|_| ()), Err(_) => self.child.kill().await, } } diff --git a/codex-rs/rmcp-client/src/macos_stdio.rs b/codex-rs/rmcp-client/src/macos_stdio.rs new file mode 100644 index 0000000000..78dc330ce2 --- /dev/null +++ b/codex-rs/rmcp-client/src/macos_stdio.rs @@ -0,0 +1,376 @@ +//! Native spawning for relative macOS MCP executables, without rewriting script paths. +//! +//! Rust falls back to fork for a historical relative-path/cwd bug in Apple's +//! `posix_spawnp`. Calling `posix_spawn` directly avoids that wrapper. This module only +//! accepts the launcher's cleared-environment command shape, with piped stdio, +//! a new process group, and default `argv[0]`. PATH lookup and executable files +//! without shebangs retain the existing launcher. Each native child owns its PID +//! until it has been reaped. + +use std::ffi::CString; +use std::ffi::OsStr; +use std::io; +use std::os::fd::AsRawFd; +use std::os::fd::FromRawFd; +use std::os::fd::OwnedFd; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::process::ExitStatusExt; +use std::path::Path; +use std::process::ExitStatus; +use std::ptr; + +use tokio::process::Child; +use tokio::process::ChildStderr; +use tokio::process::ChildStdin; +use tokio::process::ChildStdout; +use tokio::process::Command; +use tokio::signal::unix::Signal; +use tokio::signal::unix::SignalKind; +use tokio::signal::unix::signal; + +/// Matches Tokio's child API while keeping native spawning private to macOS. +pub(crate) struct LocalChild { + inner: ChildKind, + pub(crate) stdin: Option, + pub(crate) stdout: Option, + pub(crate) stderr: Option, +} + +enum ChildKind { + Tokio(Child), + Native(NativeChild), +} + +impl LocalChild { + /// Uses native spawning for relative paths, retaining Tokio's fallback for + /// PATH lookup and executable text without a shebang. + pub(super) fn spawn(mut command: Command) -> io::Result { + let program = command.as_std().get_program(); + if Path::new(program).is_relative() + && program.as_bytes().contains(&b'/') + && let Some((child, stdin, stdout, stderr)) = NativeChild::spawn(command.as_std())? + { + return Ok(Self { + inner: ChildKind::Native(child), + stdin: Some(stdin), + stdout: Some(stdout), + stderr: Some(stderr), + }); + } + let mut child = command.spawn()?; + Ok(Self { + stdin: child.stdin.take(), + stdout: child.stdout.take(), + stderr: child.stderr.take(), + inner: ChildKind::Tokio(child), + }) + } + + pub(crate) fn id(&self) -> Option { + match &self.inner { + ChildKind::Tokio(child) => child.id(), + ChildKind::Native(child) => child.id(), + } + } + + pub(crate) async fn wait(&mut self) -> io::Result { + self.stdin.take(); + match &mut self.inner { + ChildKind::Tokio(child) => child.wait().await, + ChildKind::Native(child) => child.wait().await, + } + } + + pub(crate) async fn kill(&mut self) -> io::Result<()> { + self.stdin.take(); + match &mut self.inner { + ChildKind::Tokio(child) => child.kill().await, + ChildKind::Native(child) => child.kill().await, + } + } +} + +// libc does not expose this Apple extension. It is available since macOS 10.15, +// before Codex's minimum supported macOS version (12). +unsafe extern "C" { + fn posix_spawn_file_actions_addchdir_np( + actions: *mut libc::posix_spawn_file_actions_t, + path: *const libc::c_char, + ) -> libc::c_int; +} + +/// Owns a child PID until reaping, so cancellation cannot lose or reuse it. +/// Dropping a live child kills it and reaps it independently of the Tokio runtime. +struct NativeChild { + pid: Option, + status: Option, + sigchld: Signal, +} + +impl NativeChild { + /// Spawns the MCP command without changing its executable path or `argv[0]`. + /// The caller must clear inherited environment variables before setting the + /// child's environment, because only explicit command entries are copied. + fn spawn( + command: &std::process::Command, + ) -> io::Result> { + let program = c_string(command.get_program())?; + let args = std::iter::once(command.get_program()) + .chain(command.get_args()) + .map(c_string) + .collect::>>()?; + let argv = args + .iter() + .map(|arg| arg.as_ptr().cast_mut()) + .chain(std::iter::once(ptr::null_mut())) + .collect::>(); + let env = command + .get_envs() + .filter_map(|(key, value)| value.map(|value| (key, value))) + .map(|(key, value)| { + let mut entry = key.to_os_string(); + entry.push("="); + entry.push(value); + c_string(&entry) + }) + .collect::>>()?; + let envp = env + .iter() + .map(|entry| entry.as_ptr().cast_mut()) + .chain(std::iter::once(ptr::null_mut())) + .collect::>(); + let cwd = command + .get_current_dir() + .map(|cwd| c_string(cwd.as_os_str())) + .transpose()?; + + // Subscribe before spawning so a child that exits immediately cannot be missed. + let sigchld = signal(SignalKind::child())?; + let (stdin_read, stdin_write) = io::pipe()?; + let (stdout_read, stdout_write) = io::pipe()?; + let (stderr_read, stderr_write) = io::pipe()?; + let child_fds = [ + child_fd(stdin_read.into())?, + child_fd(stdout_write.into())?, + child_fd(stderr_write.into())?, + ]; + let stdin = ChildStdin::from_std(OwnedFd::from(stdin_write).into())?; + let stdout = ChildStdout::from_std(OwnedFd::from(stdout_read).into())?; + let stderr = ChildStderr::from_std(OwnedFd::from(stderr_read).into())?; + + let mut actions = FileActions(ptr::null_mut()); + let mut attrs = Attributes(ptr::null_mut()); + let mut pid = 0; + // SAFETY: All C strings and pipe descriptors outlive this synchronous + // spawn. The initialized action/attribute objects are destroyed by RAII. + let result = unsafe { + cvt(libc::posix_spawn_file_actions_init(&mut actions.0))?; + cvt(libc::posix_spawnattr_init(&mut attrs.0))?; + if let Some(cwd) = &cwd { + cvt(posix_spawn_file_actions_addchdir_np( + &mut actions.0, + cwd.as_ptr(), + ))?; + } + for (target, source) in child_fds.iter().enumerate() { + cvt(libc::posix_spawn_file_actions_adddup2( + &mut actions.0, + source.as_raw_fd(), + target as i32, + ))?; + } + cvt(libc::posix_spawnattr_setpgroup( + &mut attrs.0, + /*pgroup*/ 0, + ))?; + let mut defaults = 0; + cvt_errno(libc::sigemptyset(&mut defaults))?; + cvt_errno(libc::sigaddset(&mut defaults, libc::SIGPIPE))?; + cvt(libc::posix_spawnattr_setsigdefault(&mut attrs.0, &defaults))?; + // Match Command's descriptor inheritance: honor FD_CLOEXEC rather + // than introducing a different policy with CLOEXEC_DEFAULT. + cvt(libc::posix_spawnattr_setflags( + &mut attrs.0, + (libc::POSIX_SPAWN_SETPGROUP | libc::POSIX_SPAWN_SETSIGDEF) as _, + ))?; + libc::posix_spawn( + &mut pid, + program.as_ptr(), + &actions.0, + &attrs.0, + argv.as_ptr(), + envp.as_ptr(), + ) + }; + // execvp also supports executable text files without a shebang. Leave + // that compatibility behavior to the existing Command-based launcher. + if result == libc::ENOEXEC { + return Ok(None); + } + cvt(result)?; + let child = Self { + pid: Some(pid), + status: None, + sigchld, + }; + Ok(Some((child, stdin, stdout, stderr))) + } + + fn id(&self) -> Option { + self.pid.map(|pid| pid as u32) + } + + /// Polls and caches the exit status, relinquishing the PID once it is reaped. + /// `ECHILD` also relinquishes it to prevent later signaling of a reused PID. + fn try_wait(&mut self) -> io::Result> { + if self.status.is_some() { + return Ok(self.status); + } + let pid = self + .pid + .ok_or_else(|| io::Error::from_raw_os_error(libc::ECHILD))?; + let mut status = 0; + // SAFETY: We own this child PID and provide writable status storage. + match unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) } { + 0 => Ok(None), + -1 => { + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::ECHILD) { + self.pid = None; + } + Err(error) + } + _ => { + self.pid = None; + self.status = Some(ExitStatus::from_raw(status)); + Ok(self.status) + } + } + } + + /// Waits without transferring child ownership into the future, so callers + /// may cancel a wait and then wait again or kill the same child. + async fn wait(&mut self) -> io::Result { + loop { + match self.try_wait() { + Ok(Some(status)) => return Ok(status), + Ok(None) => {} + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + self.sigchld + .recv() + .await + .ok_or_else(|| io::Error::other("SIGCHLD stream closed"))?; + } + } + + /// Sends SIGKILL if still owned, then waits for the child to be reaped. + async fn kill(&mut self) -> io::Result<()> { + if let Some(pid) = self.pid { + // SAFETY: An unreaped child retains its PID, even after it exits. + let result = unsafe { libc::kill(pid, libc::SIGKILL) }; + if result == -1 && io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH) { + return Err(io::Error::last_os_error()); + } + } + self.wait().await.map(|_| ()) + } +} + +impl Drop for NativeChild { + fn drop(&mut self) { + let _ = self.try_wait(); + let Some(pid) = self.pid.take() else { return }; + // SAFETY: This child has not been reaped, so its PID cannot be reused. + unsafe { + libc::kill(pid, libc::SIGKILL); + } + // Drop may run during runtime shutdown. Reap independently of Tokio, + // without blocking its worker threads while the killed process exits. + if std::thread::Builder::new() + .name("mcp-child-reaper".into()) + .spawn(move || reap(pid)) + .is_err() + { + // Resource exhaustion must not turn a dropped child into a zombie. + reap(pid); + } + } +} + +/// Reaps the child transferred by `Drop`, retrying interrupted waits without +/// requiring a live Tokio runtime. The caller relinquishes ownership of `pid`. +fn reap(pid: libc::pid_t) { + loop { + // SAFETY: The caller transferred exclusive ownership of this unreaped PID. + let result = unsafe { + libc::waitpid(pid, ptr::null_mut(), /*options*/ 0) + }; + if result != -1 || io::Error::last_os_error().kind() != io::ErrorKind::Interrupted { + break; + } + } +} + +fn c_string(value: &OsStr) -> io::Result { + CString::new(value.as_bytes()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "nul byte in MCP command")) +} + +/// Keeps a child pipe source above stdio so `dup2` cannot clobber another source +/// when the parent has closed standard descriptors. Close-on-exec disposes of +/// this extra descriptor after the spawn actions duplicate it onto stdio. +fn child_fd(fd: OwnedFd) -> io::Result { + // SAFETY: fcntl duplicates this live descriptor; the returned fd is newly owned. + let duplicate = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_DUPFD_CLOEXEC, 3) }; + cvt_errno(duplicate)?; + Ok(unsafe { OwnedFd::from_raw_fd(duplicate) }) +} + +/// Converts a spawn API's returned error number, which does not use `errno`. +fn cvt(result: libc::c_int) -> io::Result<()> { + if result == 0 { + Ok(()) + } else { + Err(io::Error::from_raw_os_error(result)) + } +} + +/// Converts a syscall's `-1` sentinel using the thread's current `errno`. +fn cvt_errno(result: libc::c_int) -> io::Result<()> { + if result == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + } +} + +struct FileActions(libc::posix_spawn_file_actions_t); +struct Attributes(libc::posix_spawnattr_t); + +impl Drop for FileActions { + fn drop(&mut self) { + if !self.0.is_null() { + // SAFETY: This object was initialized by posix_spawn_file_actions_init. + unsafe { + libc::posix_spawn_file_actions_destroy(&mut self.0); + } + } + } +} + +impl Drop for Attributes { + fn drop(&mut self) { + if !self.0.is_null() { + // SAFETY: This object was initialized by posix_spawnattr_init. + unsafe { + libc::posix_spawnattr_destroy(&mut self.0); + } + } + } +} + +#[cfg(test)] +#[path = "macos_stdio_tests.rs"] +mod tests; diff --git a/codex-rs/rmcp-client/src/macos_stdio_tests.rs b/codex-rs/rmcp-client/src/macos_stdio_tests.rs new file mode 100644 index 0000000000..f475ba5e1d --- /dev/null +++ b/codex-rs/rmcp-client/src/macos_stdio_tests.rs @@ -0,0 +1,211 @@ +//! Regression coverage for native spawn compatibility, descriptor inheritance, and reaping. + +use super::*; +use pretty_assertions::assert_eq; +use std::fs; +use std::os::unix::ffi::OsStringExt; +use std::os::unix::fs::PermissionsExt; +use std::os::unix::fs::symlink; +use std::time::Duration; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; + +#[tokio::test] +async fn relative_script_preserves_paths_stdio_environment_and_process_group() -> anyhow::Result<()> +{ + let root = tempfile::tempdir()?; + fs::create_dir_all(root.path().join("actual/bin"))?; + symlink("actual/bin", root.path().join("link"))?; + let script = root.path().join("actual/server"); + fs::write( + &script, + "#!/bin/sh\nread -r input\nprintf '%s\\n' \"$0\" \"$1\" \"$2\" \"$MCP_TEST\" \"$input\"\nprintf diagnostic >&2\nexit 23\n", + )?; + fs::set_permissions(script, fs::Permissions::from_mode(0o755))?; + let mut command = std::process::Command::new("./link/../server"); + command + .current_dir(root.path()) + .env_clear() + .env("MCP_TEST", "kept") + .arg("spaces ; literal $arg") + .arg(std::ffi::OsString::from_vec(b"raw-\xff".to_vec())); + let (mut child, mut stdin, mut stdout, mut stderr) = + NativeChild::spawn(&command)?.expect("native child"); + let pid = child.id().expect("live PID") as libc::pid_t; + // SAFETY: getpgid only inspects the live child, which waits for input below. + assert_eq!(unsafe { libc::getpgid(pid) }, pid); + stdin.write_all(b"hello\n").await?; + drop(stdin); + let mut output = Vec::new(); + let mut diagnostic = String::new(); + let (_, _, status) = tokio::try_join!( + stdout.read_to_end(&mut output), + stderr.read_to_string(&mut diagnostic), + child.wait() + )?; + assert_eq!( + (output.as_slice(), diagnostic.as_str(), status.code()), + ( + b"./link/../server\nspaces ; literal $arg\nraw-\xff\nkept\nhello\n".as_slice(), + "diagnostic", + Some(23) + ) + ); + assert_eq!(child.wait().await?, status); + assert_eq!(child.id(), None); + Ok(()) +} + +#[tokio::test] +async fn native_executable_preserves_argv0() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + symlink("/bin/sh", root.path().join("shell"))?; + let program = Path::new("./shell"); + let mut command = std::process::Command::new(program); + command + .current_dir(root.path()) + .env_clear() + .args(["-c", "printf '%s' \"$0\""]); + let (mut child, stdin, mut stdout, _stderr) = + NativeChild::spawn(&command)?.expect("native child"); + drop(stdin); + let mut output = Vec::new(); + stdout.read_to_end(&mut output).await?; + assert!(child.wait().await?.success()); + assert_eq!(output, program.as_os_str().as_bytes()); + Ok(()) +} + +#[tokio::test] +async fn cancelled_wait_can_still_kill_and_reap_child() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + symlink("/bin/cat", root.path().join("server"))?; + let mut command = std::process::Command::new("./server"); + command.current_dir(root.path()).env_clear(); + let (mut child, _stdin, _stdout, _stderr) = + NativeChild::spawn(&command)?.expect("native child"); + assert!( + tokio::time::timeout(Duration::from_millis(20), child.wait()) + .await + .is_err() + ); + child.kill().await?; + assert_eq!(child.wait().await?.signal(), Some(libc::SIGKILL)); + assert_eq!(child.id(), None); + Ok(()) +} + +#[tokio::test] +async fn descriptor_inheritance_matches_command() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + symlink("/bin/sh", root.path().join("shell"))?; + let file = fs::File::open("/dev/null")?; + for (operation, expected) in [ + (libc::F_DUPFD, "inherited"), + (libc::F_DUPFD_CLOEXEC, "closed"), + ] { + // SAFETY: Duplicate a harmless descriptor with the requested inheritance flag. + let fd = unsafe { libc::fcntl(file.as_raw_fd(), operation, 200) }; + cvt_errno(fd)?; + // SAFETY: fcntl returned a new owned descriptor. + let _sentinel = unsafe { OwnedFd::from_raw_fd(fd) }; + let mut command = std::process::Command::new("./shell"); + command + .current_dir(root.path()) + .env_clear() + .env("SENTINEL", fd.to_string()) + .args([ + "-c", + "if [ -e /dev/fd/\"$SENTINEL\" ]; then printf inherited; else printf closed; fi", + ]); + let (mut child, stdin, mut stdout, _stderr) = + NativeChild::spawn(&command)?.expect("native child"); + drop(stdin); + let mut output = String::new(); + stdout.read_to_string(&mut output).await?; + assert!(child.wait().await?.success()); + assert_eq!(output, expected); + assert_eq!(command.output()?.stdout, output.as_bytes()); + } + Ok(()) +} + +#[tokio::test] +async fn launch_failures_preserve_os_errors() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + fs::write(root.path().join("not-executable"), "#!/bin/sh\nexit 0\n")?; + for (program, cwd, errno) in [ + ("./missing", root.path().to_path_buf(), libc::ENOENT), + ("./not-executable", root.path().to_path_buf(), libc::EACCES), + ("/bin/sh", root.path().join("missing"), libc::ENOENT), + ] { + let mut command = std::process::Command::new(program); + command.env_clear().current_dir(cwd); + let error = NativeChild::spawn(&command) + .err() + .expect("spawn should fail"); + assert_eq!(error.raw_os_error(), Some(errno)); + } + Ok(()) +} + +#[tokio::test] +async fn executable_text_without_shebang_retains_command_fallback() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let script = root.path().join("server"); + fs::write(&script, "printf '%s' \"$0\"\n")?; + fs::set_permissions(script, fs::Permissions::from_mode(0o755))?; + let mut command = Command::new("./server"); + command.current_dir(root.path()).env_clear(); + let mut child = crate::local_child::spawn(command)?; + let mut output = Vec::new(); + child + .stdout + .take() + .expect("piped stdout") + .read_to_end(&mut output) + .await?; + assert!(child.wait().await?.success()); + assert_eq!(output, b"./server"); + Ok(()) +} + +#[test] +fn dropping_after_runtime_shutdown_kills_and_reaps_child() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + symlink("/bin/cat", root.path().join("server"))?; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let (child, stdin, _stdout, _stderr) = runtime + .block_on(async { + let mut command = std::process::Command::new("./server"); + command.current_dir(root.path()).env_clear(); + NativeChild::spawn(&command) + })? + .expect("native child"); + let pid = child.id().expect("live PID") as libc::pid_t; + drop(runtime); + drop(child); + drop(stdin); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + // SAFETY: Signal zero probes existence without changing the process. + if unsafe { libc::kill(pid, 0) } == -1 { + assert_eq!(io::Error::last_os_error().raw_os_error(), Some(libc::ESRCH)); + break; + } + anyhow::ensure!(std::time::Instant::now() < deadline, "child was not reaped"); + std::thread::sleep(Duration::from_millis(10)); + } + // SAFETY: WNOHANG verifies our drop reaper already collected this child. + assert_eq!( + unsafe { libc::waitpid(pid, ptr::null_mut(), libc::WNOHANG) }, + -1 + ); + assert_eq!( + io::Error::last_os_error().raw_os_error(), + Some(libc::ECHILD) + ); + Ok(()) +} diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index 1f6ab22be0..7c636e3f3d 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -19,7 +19,6 @@ use std::io; use std::os::windows::io::OwnedHandle; use std::path::Path; use std::path::PathBuf; -use std::process::Stdio; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -54,7 +53,6 @@ use rmcp::service::RoleClient; use rmcp::service::RxJsonRpcMessage; use rmcp::service::TxJsonRpcMessage; use rmcp::transport::Transport; -use rmcp::transport::child_process::TokioChildProcess; use tokio::io::AsyncBufReadExt; use tokio::io::BufReader; use tokio::process::Command; @@ -107,8 +105,7 @@ pub struct StdioServerTransport { } enum StdioServerTransportInner { - LocalLegacy(TokioChildProcess), - LocalModern(LocalStdioTransport), + Local(LocalStdioTransport), Executor(ExecutorProcessTransport), } @@ -123,8 +120,7 @@ impl Transport for StdioServerTransport { // wrapper keeps process placement private while leaving rmcp's send // semantics unchanged. match &mut self.inner { - StdioServerTransportInner::LocalLegacy(transport) => transport.send(item).boxed(), - StdioServerTransportInner::LocalModern(transport) => transport.send(item).boxed(), + StdioServerTransportInner::Local(transport) => transport.send(item).boxed(), StdioServerTransportInner::Executor(transport) => transport.send(item).boxed(), } } @@ -134,8 +130,7 @@ impl Transport for StdioServerTransport { // executor variant turns pushed process-output events back into the // line-delimited JSON stream expected by rmcp. match &mut self.inner { - StdioServerTransportInner::LocalLegacy(transport) => transport.receive().boxed(), - StdioServerTransportInner::LocalModern(transport) => transport.receive().boxed(), + StdioServerTransportInner::Local(transport) => transport.receive().boxed(), StdioServerTransportInner::Executor(transport) => transport.receive().boxed(), } } @@ -143,8 +138,7 @@ impl Transport for StdioServerTransport { async fn close(&mut self) -> std::result::Result<(), Self::Error> { self.process.terminate().await?; match &mut self.inner { - StdioServerTransportInner::LocalLegacy(transport) => transport.close().await, - StdioServerTransportInner::LocalModern(transport) => transport.close().await, + StdioServerTransportInner::Local(transport) => transport.close().await, StdioServerTransportInner::Executor(transport) => transport.close().await, } } @@ -280,9 +274,6 @@ impl LocalStdioServerLauncher { let build_command = || { let mut command = Command::new(&resolved_program); command - .kill_on_drop(true) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) .current_dir(&cwd) .env_clear() .envs(&envs) @@ -312,29 +303,14 @@ impl LocalStdioServerLauncher { Option, Option, )> { - match protocol_mode { - McpProtocolMode::Legacy => { - let (transport, stderr) = TokioChildProcess::builder(command) - .stderr(Stdio::piped()) - .spawn()?; - let process_id = transport.id(); - Ok(( - StdioServerTransportInner::LocalLegacy(transport), - stderr, - process_id, - )) - } - McpProtocolMode::V20260728 => { - let (transport, stderr) = - LocalStdioTransport::spawn(command, program_name.clone())?; - let process_id = transport.id(); - Ok(( - StdioServerTransportInner::LocalModern(transport), - stderr, - process_id, - )) - } - } + let (transport, stderr) = + LocalStdioTransport::spawn(command, program_name.clone(), protocol_mode)?; + let process_id = transport.id(); + Ok(( + StdioServerTransportInner::Local(transport), + stderr, + process_id, + )) }; let (transport, stderr, process_id) = spawn_transport(command)?; #[cfg(windows)] diff --git a/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs b/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs index 3cb7bf5cbc..53ba7fbb95 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs @@ -39,18 +39,39 @@ async fn exercise_stdio_server( } else { Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)) }; - let env = opt_in.then(|| { - HashMap::from([( + let mut env = HashMap::new(); + if opt_in { + env.insert( OsString::from("CODEX_MCP_PROTOCOL_VERSION"), OsString::from("2026-07-28"), - )]) - }); + ); + } + let cwd = std::env::current_dir()?; + #[cfg(unix)] + let root = tempfile::tempdir()?; + #[cfg(unix)] + let (server, cwd) = if use_executor { + (server, cwd) + } else { + use std::os::unix::fs::PermissionsExt; + let wrapper = root.path().join("server"); + std::fs::write( + &wrapper, + "#!/bin/sh\nprintf '%s' \"$0\" > argv0\nexec \"$MCP_SERVER\" \"$@\"\n", + )?; + std::fs::set_permissions(wrapper, std::fs::Permissions::from_mode(0o755))?; + env.insert(OsString::from("MCP_SERVER"), server.into_os_string()); + ( + std::path::PathBuf::from("./server"), + root.path().to_path_buf(), + ) + }; let client = RmcpClient::new_stdio_client_with_protocol_mode( server.into(), vec![OsString::from(server_mode)], - env, + Some(env), &[], - /*cwd*/ Some(std::env::current_dir()?.to_string_lossy().into_owned()), + Some(cwd.to_string_lossy().into_owned()), launcher, protocol_mode, ) @@ -166,6 +187,13 @@ async fn exercise_stdio_server( }) ); assert_eq!(elicitation_count.load(Ordering::Relaxed), 1); + #[cfg(unix)] + if !use_executor { + assert_eq!( + std::fs::read_to_string(root.path().join("argv0"))?, + "./server" + ); + } client.shutdown().await; Ok(()) } diff --git a/codex-rs/rmcp-client/tests/process_group_cleanup.rs b/codex-rs/rmcp-client/tests/process_group_cleanup.rs index ac446233f8..1dcd5586ef 100644 --- a/codex-rs/rmcp-client/tests/process_group_cleanup.rs +++ b/codex-rs/rmcp-client/tests/process_group_cleanup.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use std::ffi::OsString; use std::fs; +use std::os::unix::fs::PermissionsExt; use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -85,21 +86,22 @@ async fn drop_kills_wrapper_process_group() -> Result<()> { let temp_dir = tempfile::tempdir()?; let child_pid_file = temp_dir.path().join("child.pid"); let child_pid_file_str = child_pid_file.to_string_lossy().into_owned(); + let wrapper = temp_dir.path().join("wrapper"); + fs::write( + &wrapper, + "#!/bin/sh\nsleep 300 & child_pid=$!; echo \"$child_pid\" > \"$CHILD_PID_FILE\"; cat >/dev/null\n", + )?; + fs::set_permissions(wrapper, fs::Permissions::from_mode(0o755))?; let client = RmcpClient::new_stdio_client( - OsString::from("/bin/sh"), - vec![ - OsString::from("-c"), - OsString::from( - "sleep 300 & child_pid=$!; echo \"$child_pid\" > \"$CHILD_PID_FILE\"; cat >/dev/null", - ), - ], + OsString::from("./wrapper"), + vec![], Some(HashMap::from([( OsString::from("CHILD_PID_FILE"), OsString::from(child_pid_file_str), )])), &[], - /*cwd*/ None, + Some(temp_dir.path().to_string_lossy().into_owned()), Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)), ) .await?;