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
This commit is contained in:
Charlie Marsh
2026-09-01 18:53:06 +00:00
committed by copyberry
parent ef76e6ac30
commit ea7e0714da
11 changed files with 855 additions and 166 deletions
+1
View File
@@ -54,6 +54,7 @@ tokio = { workspace = true, features = [
"io-util",
"macros",
"process",
"signal",
"rt-multi-thread",
"sync",
"io-std",
@@ -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<Mutex<Option<ChildStdin>>>,
stdout: BufReader<ChildStdout>,
pending_line: Vec<u8>,
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<RxJsonRpcMessage<RoleClient>> {
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<RxJsonRpcMessage<RoleClient>> {
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<RoleClient> 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<RoleClient>,
) -> impl Future<Output = Result<(), Self::Error>> + 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<Output = Option<RxJsonRpcMessage<RoleClient>>> + Send {
self.receive_message()
}
async fn close(&mut self) -> Result<(), Self::Error> {
self.stdin.lock().await.take();
Ok(())
}
}
@@ -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;
+2
View File
@@ -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;
+34
View File
@@ -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<LocalChild> {
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()
}
}
+42 -115
View File
@@ -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<Mutex<Option<ChildStdin>>>,
stdout: BufReader<ChildStdout>,
pending_line: Vec<u8>,
program_name: String,
child: LocalChild,
transport: StdioTransport,
}
enum StdioTransport {
/// Preserve rmcp's existing framing for servers using the initialize handshake.
Legacy(AsyncRwTransport<RoleClient, ChildStdout, ChildStdin>),
/// 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<ChildStderr>)> {
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<u32> {
self.child.id()
}
async fn receive_message(&mut self) -> Option<RxJsonRpcMessage<RoleClient>> {
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<RxJsonRpcMessage<RoleClient>> {
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<RoleClient> 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<RoleClient>,
) -> impl Future<Output = Result<(), Self::Error>> + 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<Output = io::Result<()>> + 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<Output = Option<RxJsonRpcMessage<RoleClient>>> + 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,
}
}
+376
View File
@@ -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<ChildStdin>,
pub(crate) stdout: Option<ChildStdout>,
pub(crate) stderr: Option<ChildStderr>,
}
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<Self> {
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<u32> {
match &self.inner {
ChildKind::Tokio(child) => child.id(),
ChildKind::Native(child) => child.id(),
}
}
pub(crate) async fn wait(&mut self) -> io::Result<ExitStatus> {
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<libc::pid_t>,
status: Option<ExitStatus>,
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<Option<(Self, ChildStdin, ChildStdout, ChildStderr)>> {
let program = c_string(command.get_program())?;
let args = std::iter::once(command.get_program())
.chain(command.get_args())
.map(c_string)
.collect::<io::Result<Vec<_>>>()?;
let argv = args
.iter()
.map(|arg| arg.as_ptr().cast_mut())
.chain(std::iter::once(ptr::null_mut()))
.collect::<Vec<_>>();
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::<io::Result<Vec<_>>>()?;
let envp = env
.iter()
.map(|entry| entry.as_ptr().cast_mut())
.chain(std::iter::once(ptr::null_mut()))
.collect::<Vec<_>>();
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<u32> {
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<Option<ExitStatus>> {
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<ExitStatus> {
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> {
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<OwnedFd> {
// 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;
@@ -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(())
}
@@ -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<RoleClient> 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<RoleClient> 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<RoleClient> 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<tokio::process::ChildStderr>,
Option<u32>,
)> {
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)]
+34 -6
View File
@@ -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(())
}
@@ -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?;