Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions codex-rs/rmcp-client/src/bin/test_stdio_server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -900,10 +900,52 @@ fn parse_data_url(url: &str) -> Option<(String, String)> {

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
#[cfg(windows)]
if std::env::var_os("MCP_TEST_DESCENDANT_ROLE").is_some() {
tokio::time::sleep(Duration::from_secs(30)).await;
return Ok(());
}

eprintln!("starting rmcp test server");
if let Ok(pid_file) = std::env::var("MCP_TEST_PID_FILE") {
std::fs::write(pid_file, std::process::id().to_string())?;
}
#[cfg(windows)]
if let Ok(marker_file) = std::env::var("MCP_TEST_BREAKAWAY_DENIED_FILE") {
use std::os::windows::process::CommandExt;

const CREATE_BREAKAWAY_FROM_JOB: u32 = 0x0100_0000;
const ERROR_ACCESS_DENIED: i32 = 5;

let escaped = std::process::Command::new(std::env::current_exe()?)
.creation_flags(CREATE_BREAKAWAY_FROM_JOB)
.env("MCP_TEST_DESCENDANT_ROLE", "1")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn();
match escaped {
Err(error) if error.raw_os_error() == Some(ERROR_ACCESS_DENIED) => {
std::fs::write(marker_file, "denied")?;
}
Err(error) => return Err(error.into()),
Ok(mut child) => {
let _ = child.kill();
let _ = child.wait();
return Err("MCP descendant unexpectedly escaped its Windows job".into());
}
}
}
#[cfg(windows)]
if let Ok(pid_file) = std::env::var("MCP_TEST_DESCENDANT_PID_FILE") {
let child = std::process::Command::new(std::env::current_exe()?)
.env("MCP_TEST_DESCENDANT_ROLE", "1")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()?;
std::fs::write(pid_file, child.id().to_string())?;
}
// Run the server with STDIO transport. If the client disconnects we simply
// bubble up the error so the process exits.
let service = TestToolServer::new();
Expand Down
167 changes: 114 additions & 53 deletions codex-rs/rmcp-client/src/stdio_server_launcher.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ use std::collections::HashMap;
use std::ffi::OsString;
use std::future::Future;
use std::io;
#[cfg(windows)]
use std::os::windows::io::OwnedHandle;
use std::path::Path;
use std::path::PathBuf;
use std::process::Stdio;
Expand Down Expand Up @@ -226,8 +228,9 @@ struct LocalProcessTerminator {
}

#[cfg(windows)]
struct LocalProcessTerminator {
pid: u32,
enum LocalProcessTerminator {
Job(codex_utils_pty::JobObject),
Process(OwnedHandle),
}

#[cfg(not(any(unix, windows)))]
Expand Down Expand Up @@ -274,45 +277,103 @@ impl LocalStdioServerLauncher {
let resolved_program =
program_resolver::resolve(program, &envs, &cwd).map_err(io::Error::other)?;

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)
.args(args);
#[cfg(unix)]
command.process_group(0);

let (transport, stderr, process_id) = match protocol_mode {
McpProtocolMode::Legacy => {
let (transport, stderr) = TokioChildProcess::builder(command)
.stderr(Stdio::piped())
.spawn()?;
let process_id = transport.id();
(
StdioServerTransportInner::LocalLegacy(transport),
stderr,
process_id,
)
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)
.args(&args);
#[cfg(unix)]
command.process_group(0);
command
};
#[cfg(windows)]
let mut command = build_command();
#[cfg(not(windows))]
let command = build_command();
#[cfg(windows)]
let job = match codex_utils_pty::JobObject::create_without_breakaway() {
Ok(job) => {
job.prepare_suspended_spawn(&mut command);
Some(job)
}
McpProtocolMode::V20260728 => {
let (transport, stderr) =
LocalStdioTransport::spawn(command, program_name.clone())?;
let process_id = transport.id();
(
StdioServerTransportInner::LocalModern(transport),
stderr,
process_id,
)
Err(error) => {
warn!("Windows MCP process job containment unavailable: {error}");
None
}
};
let process = StdioServerProcessHandle::local(
program_name.clone(),
process_id.map(LocalProcessTerminator::new),
);

let spawn_transport = |command: Command| -> io::Result<(
StdioServerTransportInner,
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, process_id) = spawn_transport(command)?;
#[cfg(windows)]
let (transport, stderr, process_id, job) = match job {
Some(job) => match process_id
.ok_or_else(|| io::Error::other("missing suspended MCP server process id"))
.and_then(|process_id| job.assign_and_resume_process(process_id))
{
Ok(true) => (transport, stderr, process_id, Some(job)),
Ok(false) => (transport, stderr, process_id, None),
Err(error) => {
warn!(
"Windows MCP process job containment failed; retrying without it: {error}"
);
drop(stderr);
drop(transport);
drop(job);
let (transport, stderr, process_id) = spawn_transport(build_command())?;
(transport, stderr, process_id, None)
}
},
None => (transport, stderr, process_id, None),
};
#[cfg(windows)]
let terminator = match job {
Some(job) => Some(LocalProcessTerminator::Job(job)),
None => process_id.and_then(|process_id| {
match codex_utils_pty::JobObject::open_process_handle(process_id) {
Ok(handle) => Some(LocalProcessTerminator::Process(handle)),
Err(error) => {
warn!("Windows MCP process handle unavailable: {error}");
None
}
}
}),
};
#[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 {
Expand Down Expand Up @@ -340,17 +401,12 @@ impl LocalStdioServerLauncher {
}

impl LocalProcessTerminator {
#[cfg(not(windows))]
fn new(process_group_id: u32) -> Self {
#[cfg(unix)]
{
Self { process_group_id }
}
#[cfg(windows)]
{
Self {
pid: process_group_id,
}
}
#[cfg(not(any(unix, windows)))]
{
let _ = process_group_id;
Expand Down Expand Up @@ -380,15 +436,15 @@ impl LocalProcessTerminator {

#[cfg(windows)]
fn terminate(&self) {
let _ = std::process::Command::new("taskkill")
.arg("/PID")
.arg(self.pid.to_string())
.arg("/T")
.arg("/F")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status();
let result = match self {
Self::Job(job) => job.terminate(),
Self::Process(process_handle) => {
codex_utils_pty::JobObject::terminate_process_handle(process_handle)
}
};
if let Err(error) = result {
warn!("Failed to terminate Windows MCP process: {error}");
}
}

#[cfg(not(any(unix, windows)))]
Expand Down Expand Up @@ -479,6 +535,11 @@ impl Drop for StdioServerProcessHandleInner {
/// MCP framing still runs in the orchestrator. The executor only owns the
/// child process and transports raw stdin/stdout/stderr bytes, so it does not
/// need to know about MCP methods such as `initialize` or `tools/list`.
///
/// Windows executor-backed servers retain the executor's normal descendant
/// lifetime. MCP-specific containment requires negotiated process ownership:
/// caller-controlled process IDs cannot safely select a destructive policy,
/// and a wrapper may exit while its descendants continue serving requests.
#[derive(Clone)]
pub struct ExecutorStdioServerLauncher {
exec_backend: Arc<dyn ExecBackend>,
Expand Down
Loading
Loading