From 27198bfe11bd1f909c9ed0b17f923b04b53c2785 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 8 May 2025 23:45:54 -0700 Subject: [PATCH 1/3] fix: make McpConnectionManager tolerant of MCPs that fail to start (#854) I added a typo in my `config.toml` such that the `command` for one of my `mcp_servers` did not exist and I verified that the error was surfaced in the TUI (and that I was still able to use Codex). ![image](https://github.com/user-attachments/assets/f13cc08c-f4c6-40ec-9ab4-a9d75e03152f) --- codex-rs/core/src/codex.rs | 41 ++++++++++++++++----- codex-rs/core/src/mcp_connection_manager.rs | 35 ++++++++++++------ 2 files changed, 56 insertions(+), 20 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 7d056adcd9..5cd5a6799d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -561,15 +561,35 @@ async fn submission_loop( let writable_roots = Mutex::new(get_writable_roots(&cwd)); - let mcp_connection_manager = + // Error messages to dispatch after SessionConfigured is sent. + let mut mcp_connection_errors = Vec::::new(); + let (mcp_connection_manager, failed_clients) = match McpConnectionManager::new(config.mcp_servers.clone()).await { - Ok(mgr) => mgr, + Ok((mgr, failures)) => (mgr, failures), Err(e) => { - error!("Failed to create MCP connection manager: {e:#}"); - McpConnectionManager::default() + let message = format!("Failed to create MCP connection manager: {e:#}"); + error!("{message}"); + mcp_connection_errors.push(Event { + id: sub.id.clone(), + msg: EventMsg::Error { message }, + }); + (McpConnectionManager::default(), Default::default()) } }; + // Surface individual client start-up failures to the user. + if !failed_clients.is_empty() { + for (server_name, err) in failed_clients { + let message = + format!("MCP client for `{server_name}` failed to start: {err:#}"); + error!("{message}"); + mcp_connection_errors.push(Event { + id: sub.id.clone(), + msg: EventMsg::Error { message }, + }); + } + } + // Attempt to create a RolloutRecorder *before* moving the // `instructions` value into the Session struct. let rollout_recorder = match RolloutRecorder::new(instructions.clone()).await { @@ -596,12 +616,15 @@ async fn submission_loop( })); // ack - let event = Event { - id: sub.id, + let events = std::iter::once(Event { + id: sub.id.clone(), msg: EventMsg::SessionConfigured { model }, - }; - if tx_event.send(event).await.is_err() { - return; + }) + .chain(mcp_connection_errors.into_iter()); + for event in events { + if let Err(e) = tx_event.send(event).await { + error!("failed to send event: {e:?}"); + } } } Op::UserInput { items } => { diff --git a/codex-rs/core/src/mcp_connection_manager.rs b/codex-rs/core/src/mcp_connection_manager.rs index 734c351478..e4124b9099 100644 --- a/codex-rs/core/src/mcp_connection_manager.rs +++ b/codex-rs/core/src/mcp_connection_manager.rs @@ -29,6 +29,10 @@ const MCP_TOOL_NAME_DELIMITER: &str = "__OAI_CODEX_MCP__"; /// Timeout for the `tools/list` request. const LIST_TOOLS_TIMEOUT: Duration = Duration::from_secs(10); +/// Map that holds a startup error for every MCP server that could **not** be +/// spawned successfully. +pub type ClientStartErrors = HashMap; + fn fully_qualified_tool_name(server: &str, tool: &str) -> String { format!("{server}{MCP_TOOL_NAME_DELIMITER}{tool}") } @@ -60,40 +64,49 @@ impl McpConnectionManager { /// * `mcp_servers` – Map loaded from the user configuration where *keys* /// are human-readable server identifiers and *values* are the spawn /// instructions. - pub async fn new(mcp_servers: HashMap) -> Result { + /// + /// Servers that fail to start are reported in `ClientStartErrors`: the + /// user should be informed about these errors. + pub async fn new( + mcp_servers: HashMap, + ) -> Result<(Self, ClientStartErrors)> { // Early exit if no servers are configured. if mcp_servers.is_empty() { - return Ok(Self::default()); + return Ok((Self::default(), ClientStartErrors::default())); } - // Spin up all servers concurrently. + // Launch all configured servers concurrently. let mut join_set = JoinSet::new(); - // Spawn tasks to launch each server. for (server_name, cfg) in mcp_servers { // TODO: Verify server name: require `^[a-zA-Z0-9_-]+$`? join_set.spawn(async move { let McpServerConfig { command, args, env } = cfg; let client_res = McpClient::new_stdio_client(command, args, env).await; - (server_name, client_res) }); } let mut clients: HashMap> = HashMap::with_capacity(join_set.len()); + let mut errors = ClientStartErrors::new(); + while let Some(res) = join_set.join_next().await { - let (server_name, client_res) = res?; + let (server_name, client_res) = res?; // JoinError propagation - let client = client_res - .with_context(|| format!("failed to spawn MCP server `{server_name}`"))?; - - clients.insert(server_name, std::sync::Arc::new(client)); + match client_res { + Ok(client) => { + clients.insert(server_name, std::sync::Arc::new(client)); + } + Err(e) => { + errors.insert(server_name, e.into()); + } + } } let tools = list_all_tools(&clients).await?; - Ok(Self { clients, tools }) + Ok((Self { clients, tools }, errors)) } /// Returns a single map that contains **all** tools. Each key is the From 644429b46f170d40b023d63ca3337209e846d9c0 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Fri, 9 May 2025 10:25:05 -0700 Subject: [PATCH 2/3] chore: refactor exec() into spawn_child() and exec_child_and_truncate_output() --- codex-rs/core/src/exec.rs | 70 ++++++++++++++++++++++----------------- 1 file changed, 40 insertions(+), 30 deletions(-) diff --git a/codex-rs/core/src/exec.rs b/codex-rs/core/src/exec.rs index d1939d2c22..0b7fabcc9e 100644 --- a/codex-rs/core/src/exec.rs +++ b/codex-rs/core/src/exec.rs @@ -12,6 +12,7 @@ use std::time::Instant; use tokio::io::AsyncRead; use tokio::io::AsyncReadExt; use tokio::io::BufReader; +use tokio::process::Child; use tokio::process::Command; use tokio::sync::Notify; @@ -228,40 +229,49 @@ pub struct ExecToolCallOutput { pub duration: Duration, } -pub async fn exec( - ExecParams { +pub async fn exec(params: ExecParams, ctrl_c: Arc) -> Result { + let timeout_ms = params.timeout_ms; + let child = spawn_child(params).await?; + consume_truncated_output(child, ctrl_c, timeout_ms).await +} + +/// Spawns the appropriate child process for the ExecParams. +async fn spawn_child(params: ExecParams) -> std::io::Result { + let ExecParams { command, cwd, - timeout_ms, - }: ExecParams, + timeout_ms: _, + } = params; + if command.is_empty() { + return Err(std::io::Error::new( + io::ErrorKind::InvalidInput, + "command args are empty", + )); + } + + let mut cmd = Command::new(&command[0]); + cmd.args(&command[1..]); + cmd.current_dir(cwd); + + // Do not create a file descriptor for stdin because otherwise some + // commands may hang forever waiting for input. For example, ripgrep has + // a heuristic where it may try to read from stdin as explained here: + // https://github.com/BurntSushi/ripgrep/blob/e2362d4d5185d02fa857bf381e7bd52e66fafc73/crates/core/flags/hiargs.rs#L1101-L1103 + cmd.stdin(Stdio::null()); + + cmd.stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() +} + +/// Consumes the output of a child process, truncating it so it is suitable for +/// use as the output of a `shell` tool call. Also enforces specified timeout. +async fn consume_truncated_output( + mut child: Child, ctrl_c: Arc, + timeout_ms: Option, ) -> Result { - let mut child = { - if command.is_empty() { - return Err(CodexErr::Io(io::Error::new( - io::ErrorKind::InvalidInput, - "command args are empty", - ))); - } - - let mut cmd = Command::new(&command[0]); - if command.len() > 1 { - cmd.args(&command[1..]); - } - cmd.current_dir(cwd); - - // Do not create a file descriptor for stdin because otherwise some - // commands may hang forever waiting for input. For example, ripgrep has - // a heuristic where it may try to read from stdin as explained here: - // https://github.com/BurntSushi/ripgrep/blob/e2362d4d5185d02fa857bf381e7bd52e66fafc73/crates/core/flags/hiargs.rs#L1101-L1103 - cmd.stdin(Stdio::null()); - - cmd.stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true) - .spawn()? - }; - let stdout_handle = tokio::spawn(read_capped( BufReader::new(child.stdout.take().expect("stdout is not piped")), MAX_STREAM_OUTPUT, From 63ab6984ed3f6e948fa142e7e96a1deb995d65ef Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Fri, 9 May 2025 10:38:13 -0700 Subject: [PATCH 3/3] feat: experimental env var: CODEX_SANDBOX_NETWORK_DISABLED Previous to this change: ``` $ cargo run --bin codex -- debug seatbelt --full-auto -- cargo test ---- keeps_previous_response_id_between_tasks stdout ---- thread 'keeps_previous_response_id_between_tasks' panicked at /Users/mbolin/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/wiremock-0.6.3/src/mock_server/builder.rs:107:46: Failed to bind an OS port for a mock server.: Os { code: 1, kind: PermissionDenied, message: "Operation not permitted" } note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace failures: keeps_previous_response_id_between_tasks test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s error: test failed, to rerun pass `-p codex-core --test previous_response_id` ``` --- codex-rs/cli/src/landlock.rs | 6 +- codex-rs/cli/src/seatbelt.rs | 8 +-- codex-rs/core/src/exec.rs | 72 ++++++++++++++------- codex-rs/core/src/linux.rs | 6 +- codex-rs/core/tests/previous_response_id.rs | 8 +++ codex-rs/mcp-client/src/mcp_client.rs | 1 + 6 files changed, 69 insertions(+), 32 deletions(-) diff --git a/codex-rs/cli/src/landlock.rs b/codex-rs/cli/src/landlock.rs index bc43eb57cd..6e55c02e27 100644 --- a/codex-rs/cli/src/landlock.rs +++ b/codex-rs/cli/src/landlock.rs @@ -3,6 +3,7 @@ //! On Linux the command is executed inside a Landlock + seccomp sandbox by //! calling the low-level `exec_linux` helper from `codex_core::linux`. +use codex_core::exec::spawn_child; use codex_core::protocol::SandboxPolicy; use std::os::unix::process::ExitStatusExt; use std::process; @@ -19,8 +20,9 @@ pub fn run_landlock(command: Vec, sandbox_policy: SandboxPolicy) -> anyh // Spawn a new thread and apply the sandbox policies there. let handle = std::thread::spawn(move || -> anyhow::Result { let cwd = std::env::current_dir()?; - codex_core::linux::apply_sandbox_policy_to_current_thread(sandbox_policy, &cwd)?; - let status = Command::new(&command[0]).args(&command[1..]).status()?; + codex_core::linux::apply_sandbox_policy_to_current_thread(&sandbox_policy, &cwd)?; + let child = spawn_child(command, cwd, sandbox_policy)?; + let status = child.status()?; Ok(status) }); let status = handle diff --git a/codex-rs/cli/src/seatbelt.rs b/codex-rs/cli/src/seatbelt.rs index 3c7ec2ba93..ba62b150fb 100644 --- a/codex-rs/cli/src/seatbelt.rs +++ b/codex-rs/cli/src/seatbelt.rs @@ -1,4 +1,4 @@ -use codex_core::exec::create_seatbelt_command; +use codex_core::exec::spawn_command_under_seatbelt; use codex_core::protocol::SandboxPolicy; pub async fn run_seatbelt( @@ -6,10 +6,8 @@ pub async fn run_seatbelt( sandbox_policy: SandboxPolicy, ) -> anyhow::Result<()> { let cwd = std::env::current_dir().expect("failed to get cwd"); - let seatbelt_command = create_seatbelt_command(command, &sandbox_policy, &cwd); - let status = tokio::process::Command::new(seatbelt_command[0].clone()) - .args(&seatbelt_command[1..]) - .spawn() + let child = spawn_command_under_seatbelt(command, &sandbox_policy, cwd).await; + let status = child .map_err(|e| anyhow::anyhow!("Failed to spawn command: {}", e))? .wait() .await diff --git a/codex-rs/core/src/exec.rs b/codex-rs/core/src/exec.rs index 0b7fabcc9e..b8bb25b53d 100644 --- a/codex-rs/core/src/exec.rs +++ b/codex-rs/core/src/exec.rs @@ -42,6 +42,16 @@ const MACOS_SEATBELT_BASE_POLICY: &str = include_str!("seatbelt_base_policy.sbpl /// already has root access. const MACOS_PATH_TO_SEATBELT_EXECUTABLE: &str = "/usr/bin/sandbox-exec"; +/// Experimental environment variable that will be set to some non-empty value +/// if both of the following are true: +/// +/// 1. The process was spawned by Codex as part of a shell tool call. +/// 2. SandboxPolicy.has_full_network_access() was false for the tool call. +/// +/// We may try to have just one environment variable for all sandboxing +/// attributes, so this may change in the future. +pub const CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR: &str = "CODEX_SANDBOX_NETWORK_DISABLED"; + #[derive(Debug, Clone)] pub struct ExecParams { pub command: Vec, @@ -90,23 +100,15 @@ pub async fn process_exec_tool_call( let start = Instant::now(); let raw_output_result = match sandbox_type { - SandboxType::None => exec(params, ctrl_c).await, + SandboxType::None => exec(params, sandbox_policy, ctrl_c).await, SandboxType::MacosSeatbelt => { let ExecParams { command, cwd, timeout_ms, } = params; - let seatbelt_command = create_seatbelt_command(command, sandbox_policy, &cwd); - exec( - ExecParams { - command: seatbelt_command, - cwd, - timeout_ms, - }, - ctrl_c, - ) - .await + let child = spawn_command_under_seatbelt(command, sandbox_policy, cwd).await?; + consume_truncated_output(child, ctrl_c, timeout_ms).await } SandboxType::LinuxSeccomp => exec_linux(params, ctrl_c, sandbox_policy).await, }; @@ -151,7 +153,16 @@ pub async fn process_exec_tool_call( } } -pub fn create_seatbelt_command( +pub async fn spawn_command_under_seatbelt( + command: Vec, + sandbox_policy: &SandboxPolicy, + cwd: PathBuf, +) -> std::io::Result { + let seatbelt_command = create_seatbelt_command(command, sandbox_policy, &cwd); + spawn_child(seatbelt_command, cwd, sandbox_policy).await +} + +fn create_seatbelt_command( command: Vec, sandbox_policy: &SandboxPolicy, cwd: &Path, @@ -229,21 +240,34 @@ pub struct ExecToolCallOutput { pub duration: Duration, } -pub async fn exec(params: ExecParams, ctrl_c: Arc) -> Result { - let timeout_ms = params.timeout_ms; - let child = spawn_child(params).await?; +pub async fn exec( + params: ExecParams, + sandbox_policy: &SandboxPolicy, + ctrl_c: Arc, +) -> Result { + let ExecParams { + command, + cwd, + timeout_ms, + } = params; + let child = spawn_child(command, cwd, sandbox_policy).await?; consume_truncated_output(child, ctrl_c, timeout_ms).await } /// Spawns the appropriate child process for the ExecParams. -async fn spawn_child(params: ExecParams) -> std::io::Result { - let ExecParams { - command, - cwd, - timeout_ms: _, - } = params; +async fn spawn_child( + command: Vec, + cwd: PathBuf, + sandbox_policy: &SandboxPolicy, +) -> std::io::Result { + // For now, we take `SandboxPolicy` as a parameter to exec() because we need + // to determine whether to set the `CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR` + // environment variable. Ultimately, we should be stricter about the + // environment variables that are set for the command (as we are when + // spawning an MCP server), so instead of SandboxPolicy, we should take the + // exact env to use for the Command (i.e., `env_clear().envs(env)`). if command.is_empty() { - return Err(std::io::Error::new( + return Err(io::Error::new( io::ErrorKind::InvalidInput, "command args are empty", )); @@ -253,6 +277,10 @@ async fn spawn_child(params: ExecParams) -> std::io::Result { cmd.args(&command[1..]); cmd.current_dir(cwd); + if !sandbox_policy.has_full_network_access() { + cmd.env(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR, "1"); + } + // Do not create a file descriptor for stdin because otherwise some // commands may hang forever waiting for input. For example, ripgrep has // a heuristic where it may try to read from stdin as explained here: diff --git a/codex-rs/core/src/linux.rs b/codex-rs/core/src/linux.rs index 9928cfee4e..b09ad88295 100644 --- a/codex-rs/core/src/linux.rs +++ b/codex-rs/core/src/linux.rs @@ -49,8 +49,8 @@ pub async fn exec_linux( .expect("Failed to create runtime"); rt.block_on(async { - apply_sandbox_policy_to_current_thread(sandbox_policy, ¶ms.cwd)?; - exec(params, ctrl_c_copy).await + apply_sandbox_policy_to_current_thread(&sandbox_policy, ¶ms.cwd)?; + exec(params, sandbox_policy, ctrl_c_copy).await }) }) .join(); @@ -68,7 +68,7 @@ pub async fn exec_linux( /// Apply sandbox policies inside this thread so only the child inherits /// them, not the entire CLI process. pub fn apply_sandbox_policy_to_current_thread( - sandbox_policy: SandboxPolicy, + sandbox_policy: &SandboxPolicy, cwd: &Path, ) -> Result<()> { if !sandbox_policy.has_full_network_access() { diff --git a/codex-rs/core/tests/previous_response_id.rs b/codex-rs/core/tests/previous_response_id.rs index c318f38ba5..2c899df0e9 100644 --- a/codex-rs/core/tests/previous_response_id.rs +++ b/codex-rs/core/tests/previous_response_id.rs @@ -3,6 +3,7 @@ use std::time::Duration; use codex_core::Codex; use codex_core::ModelProviderInfo; use codex_core::config::Config; +use codex_core::exec::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR; use codex_core::protocol::InputItem; use codex_core::protocol::Op; use serde_json::Value; @@ -50,6 +51,13 @@ data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"{}\",\"output\": async fn keeps_previous_response_id_between_tasks() { #![allow(clippy::unwrap_used)] + if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() { + println!( + "Skipping test because it cannot execute when network is disabled in a Codex sandbox." + ); + return; + } + // Mock server let server = MockServer::start().await; diff --git a/codex-rs/mcp-client/src/mcp_client.rs b/codex-rs/mcp-client/src/mcp_client.rs index 1c6a765c57..641de0e89a 100644 --- a/codex-rs/mcp-client/src/mcp_client.rs +++ b/codex-rs/mcp-client/src/mcp_client.rs @@ -81,6 +81,7 @@ impl McpClient { ) -> std::io::Result { let mut child = Command::new(program) .args(args) + .env_clear() .envs(create_env_for_mcp_server(env)) .stdin(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped())