From 996aa23e4ce900468047ed3ec57d1e7271f8d6de Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sat, 18 Apr 2026 21:47:43 -0700 Subject: [PATCH] [5/6] Wire executor-backed MCP stdio (#18212) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - Add the executor-backed RMCP stdio transport. - Wire MCP stdio placement through the executor environment config. - Cover local and executor-backed stdio paths with the existing MCP test helpers. ## Stack ```text o #18027 [6/6] Fail exec client operations after disconnect │ @ #18212 [5/6] Wire executor-backed MCP stdio │ o #18087 [4/6] Abstract MCP stdio server launching │ o #18020 [3/6] Add pushed exec process events │ o #18086 [2/6] Support piped stdin in exec process API │ o #18085 [1/6] Add MCP server environment config │ o main ``` --------- Co-authored-by: Codex --- codex-rs/Cargo.lock | 2 + .../app-server/src/codex_message_processor.rs | 37 +- codex-rs/cli/tests/mcp_list.rs | 2 +- codex-rs/codex-mcp/Cargo.toml | 1 + codex-rs/codex-mcp/src/lib.rs | 1 + codex-rs/codex-mcp/src/mcp/mod.rs | 26 +- .../codex-mcp/src/mcp_connection_manager.rs | 101 ++- codex-rs/config/src/lib.rs | 1 + codex-rs/config/src/mcp_edit.rs | 21 +- codex-rs/config/src/mcp_types.rs | 68 +- codex-rs/config/src/mcp_types_tests.rs | 55 +- codex-rs/config/src/types.rs | 1 + codex-rs/core/config.schema.json | 24 +- codex-rs/core/src/config/config_tests.rs | 61 +- codex-rs/core/src/config/edit.rs | 21 +- codex-rs/core/src/config/edit_tests.rs | 2 +- codex-rs/core/src/connectors.rs | 3 + codex-rs/core/src/session/mcp.rs | 7 + codex-rs/core/src/session/mod.rs | 1 + codex-rs/core/src/session/session.rs | 8 +- codex-rs/core/tests/suite/rmcp_client.rs | 589 +++++++++++++++++- codex-rs/rmcp-client/Cargo.toml | 1 + .../rmcp-client/src/bin/rmcp_test_server.rs | 4 +- .../rmcp-client/src/bin/test_stdio_server.rs | 47 +- .../src/executor_process_transport.rs | 376 +++++++++++ codex-rs/rmcp-client/src/lib.rs | 2 + codex-rs/rmcp-client/src/program_resolver.rs | 2 +- codex-rs/rmcp-client/src/rmcp_client.rs | 3 +- .../rmcp-client/src/stdio_server_launcher.rs | 272 +++++++- codex-rs/rmcp-client/src/utils.rs | 135 +++- codex-rs/utils/cli/src/format_env_display.rs | 17 +- 31 files changed, 1815 insertions(+), 76 deletions(-) create mode 100644 codex-rs/rmcp-client/src/executor_process_transport.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 4091c0d24..e0c399a3b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2414,6 +2414,7 @@ dependencies = [ "async-channel", "codex-async-utils", "codex-config", + "codex-exec-server", "codex-login", "codex-otel", "codex-plugin", @@ -2715,6 +2716,7 @@ dependencies = [ "axum", "codex-client", "codex-config", + "codex-exec-server", "codex-keyring-store", "codex-protocol", "codex-utils-cargo-bin", diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 69cbd6cf4..4cd97e99a 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -279,6 +279,7 @@ use codex_login::default_client::set_default_client_residency_requirement; use codex_login::login_with_api_key; use codex_login::request_device_code; use codex_login::run_login_server; +use codex_mcp::McpRuntimeEnvironment; use codex_mcp::McpServerStatusSnapshot; use codex_mcp::McpSnapshotDetail; use codex_mcp::collect_mcp_server_status_snapshot_with_detail; @@ -5724,10 +5725,40 @@ impl CodexMessageProcessor { .to_mcp_config(self.thread_manager.plugins_manager().as_ref()) .await; let auth = self.auth_manager.auth().await; + let runtime_environment = match self.thread_manager.environment_manager().current().await { + Ok(Some(environment)) => { + // Status listing has no turn cwd. This fallback is used only + // by executor-backed stdio MCPs whose config omits `cwd`. + McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf()) + } + Ok(None) => McpRuntimeEnvironment::new( + Arc::new(codex_exec_server::Environment::default()), + config.cwd.to_path_buf(), + ), + Err(err) => { + // TODO(aibrahim): Investigate degrading MCP status listing when + // executor environment creation fails. + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to create environment: {err}"), + data: None, + }; + self.outgoing.send_error(request, error).await; + return; + } + }; tokio::spawn(async move { - Self::list_mcp_server_status_task(outgoing, request, params, config, mcp_config, auth) - .await; + Self::list_mcp_server_status_task( + outgoing, + request, + params, + config, + mcp_config, + auth, + runtime_environment, + ) + .await; }); } @@ -5738,6 +5769,7 @@ impl CodexMessageProcessor { config: Config, mcp_config: codex_mcp::McpConfig, auth: Option, + runtime_environment: McpRuntimeEnvironment, ) { let detail = match params.detail.unwrap_or(McpServerStatusDetail::Full) { McpServerStatusDetail::Full => McpSnapshotDetail::Full, @@ -5748,6 +5780,7 @@ impl CodexMessageProcessor { &mcp_config, auth.as_ref(), request_id.request_id.to_string(), + runtime_environment, detail, ) .await; diff --git a/codex-rs/cli/tests/mcp_list.rs b/codex-rs/cli/tests/mcp_list.rs index d41a3cc62..bed350598 100644 --- a/codex-rs/cli/tests/mcp_list.rs +++ b/codex-rs/cli/tests/mcp_list.rs @@ -55,7 +55,7 @@ async fn list_and_get_render_expected_output() -> Result<()> { .expect("docs server should exist after add"); match &mut docs_entry.transport { McpServerTransportConfig::Stdio { env_vars, .. } => { - *env_vars = vec!["APP_TOKEN".to_string(), "WORKSPACE_ID".to_string()]; + *env_vars = vec!["APP_TOKEN".into(), "WORKSPACE_ID".into()]; } other => panic!("unexpected transport: {other:?}"), } diff --git a/codex-rs/codex-mcp/Cargo.toml b/codex-rs/codex-mcp/Cargo.toml index adc38d409..0aec1f3aa 100644 --- a/codex-rs/codex-mcp/Cargo.toml +++ b/codex-rs/codex-mcp/Cargo.toml @@ -16,6 +16,7 @@ anyhow = { workspace = true } async-channel = { workspace = true } codex-async-utils = { workspace = true } codex-config = { workspace = true } +codex-exec-server = { workspace = true } codex-login = { workspace = true } codex-otel = { workspace = true } codex-plugin = { workspace = true } diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index ed0d9b412..766465f05 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -38,6 +38,7 @@ pub use mcp_connection_manager::CodexAppsToolsCacheKey; pub use mcp_connection_manager::DEFAULT_STARTUP_TIMEOUT; pub use mcp_connection_manager::MCP_SANDBOX_STATE_META_CAPABILITY; pub use mcp_connection_manager::McpConnectionManager; +pub use mcp_connection_manager::McpRuntimeEnvironment; pub use mcp_connection_manager::SandboxState; pub use mcp_connection_manager::ToolInfo; pub use mcp_connection_manager::codex_apps_tools_cache_key; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 1dc9db078..448259d5b 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -35,6 +35,7 @@ use codex_protocol::protocol::SandboxPolicy; use serde_json::Value; use crate::mcp_connection_manager::McpConnectionManager; +use crate::mcp_connection_manager::McpRuntimeEnvironment; use crate::mcp_connection_manager::codex_apps_tools_cache_key; pub type McpManager = McpConnectionManager; @@ -321,14 +322,23 @@ pub async fn collect_mcp_snapshot( config: &McpConfig, auth: Option<&CodexAuth>, submit_id: String, + runtime_environment: McpRuntimeEnvironment, ) -> McpListToolsResponseEvent { - collect_mcp_snapshot_with_detail(config, auth, submit_id, McpSnapshotDetail::Full).await + collect_mcp_snapshot_with_detail( + config, + auth, + submit_id, + runtime_environment, + McpSnapshotDetail::Full, + ) + .await } pub async fn collect_mcp_snapshot_with_detail( config: &McpConfig, auth: Option<&CodexAuth>, submit_id: String, + runtime_environment: McpRuntimeEnvironment, detail: McpSnapshotDetail, ) -> McpListToolsResponseEvent { let mcp_servers = effective_mcp_servers(config, auth); @@ -356,6 +366,7 @@ pub async fn collect_mcp_snapshot_with_detail( submit_id, tx_event, SandboxPolicy::new_read_only_policy(), + runtime_environment, config.codex_home.clone(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, @@ -386,15 +397,23 @@ pub async fn collect_mcp_server_status_snapshot( config: &McpConfig, auth: Option<&CodexAuth>, submit_id: String, + runtime_environment: McpRuntimeEnvironment, ) -> McpServerStatusSnapshot { - collect_mcp_server_status_snapshot_with_detail(config, auth, submit_id, McpSnapshotDetail::Full) - .await + collect_mcp_server_status_snapshot_with_detail( + config, + auth, + submit_id, + runtime_environment, + McpSnapshotDetail::Full, + ) + .await } pub async fn collect_mcp_server_status_snapshot_with_detail( config: &McpConfig, auth: Option<&CodexAuth>, submit_id: String, + runtime_environment: McpRuntimeEnvironment, detail: McpSnapshotDetail, ) -> McpServerStatusSnapshot { let mcp_servers = effective_mcp_servers(config, auth); @@ -422,6 +441,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( submit_id, tx_event, SandboxPolicy::new_read_only_policy(), + runtime_environment, config.codex_home.clone(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, diff --git a/codex-rs/codex-mcp/src/mcp_connection_manager.rs b/codex-rs/codex-mcp/src/mcp_connection_manager.rs index 7a290973f..10c48e040 100644 --- a/codex-rs/codex-mcp/src/mcp_connection_manager.rs +++ b/codex-rs/codex-mcp/src/mcp_connection_manager.rs @@ -36,6 +36,7 @@ use codex_async_utils::CancelErr; use codex_async_utils::OrCancelExt; use codex_config::Constrained; use codex_config::types::OAuthCredentialsStoreMode; +use codex_exec_server::Environment; use codex_protocol::ToolName; use codex_protocol::approvals::ElicitationRequest; use codex_protocol::approvals::ElicitationRequestEvent; @@ -50,6 +51,7 @@ use codex_protocol::protocol::McpStartupStatus; use codex_protocol::protocol::McpStartupUpdateEvent; use codex_protocol::protocol::SandboxPolicy; use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::ExecutorStdioServerLauncher; use codex_rmcp_client::LocalStdioServerLauncher; use codex_rmcp_client::RmcpClient; use codex_rmcp_client::SendElicitation; @@ -493,6 +495,7 @@ impl AsyncManagedClient { elicitation_requests: ElicitationRequestManager, codex_apps_tools_cache_context: Option, tool_plugin_provenance: Arc, + runtime_environment: McpRuntimeEnvironment, ) -> Self { let tool_filter = ToolFilter::from_config(&config); let startup_snapshot = load_startup_cached_codex_apps_tools_snapshot( @@ -509,8 +512,15 @@ impl AsyncManagedClient { return Err(error.into()); } - let client = - Arc::new(make_rmcp_client(&server_name, config.transport, store_mode).await?); + let client = Arc::new( + make_rmcp_client( + &server_name, + config.clone(), + store_mode, + runtime_environment, + ) + .await?, + ); match start_server_task( server_name, client, @@ -650,6 +660,37 @@ pub struct McpConnectionManager { elicitation_requests: ElicitationRequestManager, } +/// Runtime placement information used when starting MCP server transports. +/// +/// `McpConfig` describes what servers exist. This value describes where those +/// servers should run for the current caller. Keep it explicit at manager +/// construction time so status/snapshot paths and real sessions make the same +/// local-vs-remote decision. `fallback_cwd` is not a per-server override; it is +/// used only when an executor-backed stdio server omits `cwd` and the executor +/// API still needs a concrete process working directory. +#[derive(Clone)] +pub struct McpRuntimeEnvironment { + environment: Arc, + fallback_cwd: PathBuf, +} + +impl McpRuntimeEnvironment { + pub fn new(environment: Arc, fallback_cwd: PathBuf) -> Self { + Self { + environment, + fallback_cwd, + } + } + + fn environment(&self) -> Arc { + Arc::clone(&self.environment) + } + + fn fallback_cwd(&self) -> PathBuf { + self.fallback_cwd.clone() + } +} + impl McpConnectionManager { pub fn configured_servers(&self, config: &McpConfig) -> HashMap { configured_mcp_servers(config) @@ -710,6 +751,7 @@ impl McpConnectionManager { submit_id: String, tx_event: Sender, initial_sandbox_policy: SandboxPolicy, + runtime_environment: McpRuntimeEnvironment, codex_home: PathBuf, codex_apps_tools_cache_key: CodexAppsToolsCacheKey, tool_plugin_provenance: ToolPluginProvenance, @@ -754,6 +796,7 @@ impl McpConnectionManager { elicitation_requests.clone(), codex_apps_tools_cache_context, Arc::clone(&tool_plugin_provenance), + runtime_environment.clone(), ); clients.insert(server_name.clone(), async_managed_client.clone()); let tx_event = tx_event.clone(); @@ -1484,9 +1527,25 @@ struct StartServerTaskParams { async fn make_rmcp_client( server_name: &str, - transport: McpServerTransportConfig, + config: McpServerConfig, store_mode: OAuthCredentialsStoreMode, + runtime_environment: McpRuntimeEnvironment, ) -> Result { + let McpServerConfig { + transport, + experimental_environment, + .. + } = config; + let remote_environment = match experimental_environment.as_deref() { + None | Some("local") => false, + Some("remote") => true, + Some(environment) => { + return Err(StartupOutcomeError::from(anyhow!( + "unsupported experimental_environment `{environment}` for MCP server `{server_name}`" + ))); + } + }; + match transport { McpServerTransportConfig::Stdio { command, @@ -1502,7 +1561,24 @@ async fn make_rmcp_client( .map(|(key, value)| (key.into(), value.into())) .collect::>() }); - let launcher = Arc::new(LocalStdioServerLauncher) as Arc; + let launcher = if remote_environment { + let exec_environment = runtime_environment.environment(); + if !exec_environment.is_remote() { + return Err(StartupOutcomeError::from(anyhow!( + "remote MCP server `{server_name}` requires a remote executor environment" + ))); + } + Arc::new(ExecutorStdioServerLauncher::new( + exec_environment.get_exec_backend(), + runtime_environment.fallback_cwd(), + )) + } else { + Arc::new(LocalStdioServerLauncher) as Arc + }; + + // `RmcpClient` always sees a launched MCP stdio server. The + // launcher hides whether that means a local child process or an + // executor process whose stdin/stdout bytes cross the process API. RmcpClient::new_stdio_client(command_os, args_os, env_os, &env_vars, cwd, launcher) .await .map_err(|err| StartupOutcomeError::from(anyhow!(err))) @@ -1513,6 +1589,23 @@ async fn make_rmcp_client( env_http_headers, bearer_token_env_var, } => { + if remote_environment { + if !runtime_environment.environment().is_remote() { + return Err(StartupOutcomeError::from(anyhow!( + "remote MCP server `{server_name}` requires a remote executor environment" + ))); + } + return Err(StartupOutcomeError::from(anyhow!( + // Remote HTTP needs the future low-level executor + // `network/request` API so reqwest runs on the executor side. + // Do not fall back to local HTTP here; the config explicitly + // asked for remote placement. + "remote streamable HTTP MCP server `{server_name}` is not implemented yet" + ))); + } + + // Local streamable HTTP remains the existing reqwest path from + // the orchestrator process. let resolved_bearer_token = match resolve_bearer_token(server_name, bearer_token_env_var.as_deref()) { Ok(token) => token, diff --git a/codex-rs/config/src/lib.rs b/codex-rs/config/src/lib.rs index c1d904169..a26d5afe0 100644 --- a/codex-rs/config/src/lib.rs +++ b/codex-rs/config/src/lib.rs @@ -72,6 +72,7 @@ pub use mcp_edit::load_global_mcp_servers; pub use mcp_types::AppToolApproval; pub use mcp_types::McpServerConfig; pub use mcp_types::McpServerDisabledReason; +pub use mcp_types::McpServerEnvVar; pub use mcp_types::McpServerToolConfig; pub use mcp_types::McpServerTransportConfig; pub use mcp_types::RawMcpServerConfig; diff --git a/codex-rs/config/src/mcp_edit.rs b/codex-rs/config/src/mcp_edit.rs index c4bb38c54..f5881a125 100644 --- a/codex-rs/config/src/mcp_edit.rs +++ b/codex-rs/config/src/mcp_edit.rs @@ -14,6 +14,7 @@ use toml_edit::value; use crate::AppToolApproval; use crate::CONFIG_TOML_FILE; use crate::McpServerConfig; +use crate::McpServerEnvVar; use crate::McpServerTransportConfig; pub async fn load_global_mcp_servers( @@ -142,7 +143,7 @@ fn serialize_mcp_server(config: &McpServerConfig) -> TomlItem { entry["env"] = table_from_pairs(env.iter()); } if !env_vars.is_empty() { - entry["env_vars"] = array_from_strings(env_vars); + entry["env_vars"] = array_from_env_vars(env_vars); } if let Some(cwd) = cwd { entry["cwd"] = value(cwd.to_string_lossy().to_string()); @@ -247,6 +248,24 @@ fn array_from_strings(values: &[String]) -> TomlItem { TomlItem::Value(array.into()) } +fn array_from_env_vars(env_vars: &[McpServerEnvVar]) -> TomlItem { + let mut array = toml_edit::Array::new(); + for env_var in env_vars { + match env_var { + McpServerEnvVar::Name(name) => array.push(name.clone()), + McpServerEnvVar::Config { name, source } => { + let mut table = toml_edit::InlineTable::new(); + table.insert("name", name.clone().into()); + if let Some(source) = source { + table.insert("source", source.clone().into()); + } + array.push(table); + } + } + } + TomlItem::Value(array.into()) +} + fn table_from_pairs<'a, I>(pairs: I) -> TomlItem where I: IntoIterator, diff --git a/codex-rs/config/src/mcp_types.rs b/codex-rs/config/src/mcp_types.rs index 75b68c3f9..a276cd707 100644 --- a/codex-rs/config/src/mcp_types.rs +++ b/codex-rs/config/src/mcp_types.rs @@ -56,6 +56,64 @@ pub struct McpServerToolConfig { pub approval_mode: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema)] +#[serde(untagged, deny_unknown_fields)] +pub enum McpServerEnvVar { + Name(String), + Config { + name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, + }, +} + +impl McpServerEnvVar { + pub fn name(&self) -> &str { + match self { + McpServerEnvVar::Name(name) => name, + McpServerEnvVar::Config { name, .. } => name, + } + } + + pub fn source(&self) -> Option<&str> { + match self { + McpServerEnvVar::Name(_) => None, + McpServerEnvVar::Config { source, .. } => source.as_deref(), + } + } + + pub fn is_remote_source(&self) -> bool { + self.source() == Some("remote") + } + + pub fn validate_source(&self) -> Result<(), String> { + match self.source() { + None | Some("local") | Some("remote") => Ok(()), + Some(source) => Err(format!( + "unsupported env_vars source `{source}`; expected `local` or `remote`" + )), + } + } +} + +impl From for McpServerEnvVar { + fn from(value: String) -> Self { + Self::Name(value) + } +} + +impl From<&str> for McpServerEnvVar { + fn from(value: &str) -> Self { + Self::Name(value.to_string()) + } +} + +impl AsRef for McpServerEnvVar { + fn as_ref(&self) -> &str { + self.name() + } +} + #[derive(Serialize, Debug, Clone, PartialEq)] pub struct McpServerConfig { #[serde(flatten)] @@ -133,7 +191,7 @@ pub struct RawMcpServerConfig { #[serde(default)] pub env: Option>, #[serde(default)] - pub env_vars: Option>, + pub env_vars: Option>, #[serde(default)] pub cwd: Option, pub http_headers: Option>, @@ -235,11 +293,15 @@ impl TryFrom for McpServerConfig { throw_if_set("stdio", "http_headers", http_headers.as_ref())?; throw_if_set("stdio", "env_http_headers", env_http_headers.as_ref())?; throw_if_set("stdio", "oauth_resource", oauth_resource.as_ref())?; + let env_vars = env_vars.unwrap_or_default(); + for env_var in &env_vars { + env_var.validate_source()?; + } McpServerTransportConfig::Stdio { command, args: args.unwrap_or_default(), env, - env_vars: env_vars.unwrap_or_default(), + env_vars, cwd, } } else if let Some(url) = url { @@ -303,7 +365,7 @@ pub enum McpServerTransportConfig { #[serde(default, skip_serializing_if = "Option::is_none")] env: Option>, #[serde(default, skip_serializing_if = "Vec::is_empty")] - env_vars: Vec, + env_vars: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] cwd: Option, }, diff --git a/codex-rs/config/src/mcp_types_tests.rs b/codex-rs/config/src/mcp_types_tests.rs index dff4a6bbf..50f705d0f 100644 --- a/codex-rs/config/src/mcp_types_tests.rs +++ b/codex-rs/config/src/mcp_types_tests.rs @@ -91,12 +91,65 @@ fn deserialize_stdio_command_server_config_with_env_vars() { command: "echo".to_string(), args: vec![], env: None, - env_vars: vec!["FOO".to_string(), "BAR".to_string()], + env_vars: vec!["FOO".into(), "BAR".into()], cwd: None, } ); } +#[test] +fn deserialize_stdio_command_server_config_with_env_var_sources() { + let cfg: McpServerConfig = toml::from_str( + r#" + command = "echo" + env_vars = [ + "LEGACY_TOKEN", + { name = "LOCAL_TOKEN", source = "local" }, + { name = "REMOTE_TOKEN", source = "remote" }, + ] + "#, + ) + .expect("should deserialize command config with sourced env_vars"); + + assert_eq!( + cfg.transport, + McpServerTransportConfig::Stdio { + command: "echo".to_string(), + args: vec![], + env: None, + env_vars: vec![ + McpServerEnvVar::Name("LEGACY_TOKEN".to_string()), + McpServerEnvVar::Config { + name: "LOCAL_TOKEN".to_string(), + source: Some("local".to_string()), + }, + McpServerEnvVar::Config { + name: "REMOTE_TOKEN".to_string(), + source: Some("remote".to_string()), + }, + ], + cwd: None, + } + ); +} + +#[test] +fn deserialize_stdio_command_server_config_rejects_unknown_env_var_source() { + let err = toml::from_str::( + r#" + command = "echo" + env_vars = [{ name = "TOKEN", source = "elsewhere" }] + "#, + ) + .expect_err("unsupported env var source should be rejected"); + + assert!( + err.to_string() + .contains("unsupported env_vars source `elsewhere`"), + "unexpected error: {err}" + ); +} + #[test] fn deserialize_stdio_command_server_config_with_cwd() { let cfg: McpServerConfig = toml::from_str( diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index ae62216d2..50a24db53 100644 --- a/codex-rs/config/src/types.rs +++ b/codex-rs/config/src/types.rs @@ -6,6 +6,7 @@ pub use crate::mcp_types::AppToolApproval; pub use crate::mcp_types::McpServerConfig; pub use crate::mcp_types::McpServerDisabledReason; +pub use crate::mcp_types::McpServerEnvVar; pub use crate::mcp_types::McpServerToolConfig; pub use crate::mcp_types::McpServerTransportConfig; pub use crate::mcp_types::RawMcpServerConfig; diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 5135e4b0d..240d84fdb 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -876,6 +876,28 @@ ], "type": "string" }, + "McpServerEnvVar": { + "anyOf": [ + { + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "name": { + "type": "string" + }, + "source": { + "type": "string" + } + }, + "required": [ + "name" + ], + "type": "object" + } + ] + }, "McpServerToolConfig": { "additionalProperties": false, "description": "Per-tool approval settings for a single MCP server tool.", @@ -1570,7 +1592,7 @@ "env_vars": { "default": null, "items": { - "type": "string" + "$ref": "#/definitions/McpServerEnvVar" }, "type": "array" }, diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index bc65aac61..edf5b9ebc 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -32,6 +32,7 @@ use codex_config::types::ApprovalsReviewer; use codex_config::types::BundledSkillsConfig; use codex_config::types::FeedbackConfigToml; use codex_config::types::HistoryPersistence; +use codex_config::types::McpServerEnvVar; use codex_config::types::McpServerToolConfig; use codex_config::types::McpServerTransportConfig; use codex_config::types::MemoriesConfig; @@ -2490,7 +2491,7 @@ async fn replace_mcp_servers_serializes_env_vars() -> anyhow::Result<()> { command: "docs-server".to_string(), args: Vec::new(), env: None, - env_vars: vec!["ALPHA".to_string(), "BETA".to_string()], + env_vars: vec!["ALPHA".into(), "BETA".into()], cwd: None, }, experimental_environment: None, @@ -2526,7 +2527,7 @@ async fn replace_mcp_servers_serializes_env_vars() -> anyhow::Result<()> { let docs = loaded.get("docs").expect("docs entry"); match &docs.transport { McpServerTransportConfig::Stdio { env_vars, .. } => { - assert_eq!(env_vars, &vec!["ALPHA".to_string(), "BETA".to_string()]); + assert_eq!(env_vars, &vec!["ALPHA".into(), "BETA".into()]); } other => panic!("unexpected transport {other:?}"), } @@ -2534,6 +2535,62 @@ async fn replace_mcp_servers_serializes_env_vars() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn replace_mcp_servers_serializes_sourced_env_vars() -> anyhow::Result<()> { + let codex_home = TempDir::new()?; + + let servers = BTreeMap::from([( + "docs".to_string(), + McpServerConfig { + transport: McpServerTransportConfig::Stdio { + command: "docs-server".to_string(), + args: Vec::new(), + env: None, + env_vars: vec![ + "LEGACY".into(), + McpServerEnvVar::Config { + name: "REMOTE_TOKEN".to_string(), + source: Some("remote".to_string()), + }, + ], + cwd: None, + }, + experimental_environment: None, + enabled: true, + required: false, + supports_parallel_tool_calls: false, + disabled_reason: None, + startup_timeout_sec: None, + tool_timeout_sec: None, + default_tools_approval_mode: None, + enabled_tools: None, + disabled_tools: None, + scopes: None, + oauth_resource: None, + tools: HashMap::new(), + }, + )]); + + apply_blocking( + codex_home.path(), + /*profile*/ None, + &[ConfigEdit::ReplaceMcpServers(servers.clone())], + )?; + + let config_path = codex_home.path().join(CONFIG_TOML_FILE); + let serialized = std::fs::read_to_string(&config_path)?; + assert!( + serialized + .contains(r#"env_vars = ["LEGACY", { name = "REMOTE_TOKEN", source = "remote" }]"#), + "serialized config missing sourced env_vars field:\n{serialized}" + ); + + let loaded = load_global_mcp_servers(codex_home.path()).await?; + assert_eq!(loaded, servers); + + Ok(()) +} + #[tokio::test] async fn replace_mcp_servers_serializes_cwd() -> anyhow::Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core/src/config/edit.rs b/codex-rs/core/src/config/edit.rs index 8f336f7ff..e7cf3651c 100644 --- a/codex-rs/core/src/config/edit.rs +++ b/codex-rs/core/src/config/edit.rs @@ -136,6 +136,7 @@ pub fn model_availability_nux_count_edits(shown_count: &HashMap) -> mod document_helpers { use codex_config::types::AppToolApproval; use codex_config::types::McpServerConfig; + use codex_config::types::McpServerEnvVar; use codex_config::types::McpServerToolConfig; use codex_config::types::McpServerTransportConfig; use toml_edit::Array as TomlArray; @@ -198,7 +199,7 @@ mod document_helpers { entry["env"] = table_from_pairs(env.iter()); } if !env_vars.is_empty() { - entry["env_vars"] = array_from_iter(env_vars.iter().cloned()); + entry["env_vars"] = array_from_env_vars(env_vars); } if let Some(cwd) = cwd { entry["cwd"] = value(cwd.to_string_lossy().to_string()); @@ -348,6 +349,24 @@ mod document_helpers { TomlItem::Value(array.into()) } + fn array_from_env_vars(env_vars: &[McpServerEnvVar]) -> TomlItem { + let mut array = TomlArray::new(); + for env_var in env_vars { + match env_var { + McpServerEnvVar::Name(name) => array.push(name.clone()), + McpServerEnvVar::Config { name, source } => { + let mut table = InlineTable::new(); + table.insert("name", name.clone().into()); + if let Some(source) = source { + table.insert("source", source.clone().into()); + } + array.push(table); + } + } + } + TomlItem::Value(array.into()) + } + fn table_from_pairs<'a, I>(pairs: I) -> TomlItem where I: IntoIterator, diff --git a/codex-rs/core/src/config/edit_tests.rs b/codex-rs/core/src/config/edit_tests.rs index 84a4a34ea..2a966d9b4 100644 --- a/codex-rs/core/src/config/edit_tests.rs +++ b/codex-rs/core/src/config/edit_tests.rs @@ -696,7 +696,7 @@ fn blocking_replace_mcp_servers_round_trips() { .into_iter() .collect(), ), - env_vars: vec!["FOO".to_string()], + env_vars: vec!["FOO".into()], cwd: None, }, experimental_environment: None, diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 480c16314..933ce0ac7 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -13,6 +13,7 @@ pub use codex_app_server_protocol::AppInfo; pub use codex_app_server_protocol::AppMetadata; use codex_connectors::AllConnectorsCacheKey; use codex_connectors::DirectoryListResponse; +use codex_exec_server::Environment; use codex_login::token_data::TokenData; use codex_protocol::protocol::SandboxPolicy; use codex_tools::DiscoverableTool; @@ -37,6 +38,7 @@ use codex_login::default_client::create_client; use codex_login::default_client::originator; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpConnectionManager; +use codex_mcp::McpRuntimeEnvironment; use codex_mcp::ToolInfo; use codex_mcp::ToolPluginProvenance; use codex_mcp::codex_apps_tools_cache_key; @@ -241,6 +243,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( INITIAL_SUBMIT_ID.to_owned(), tx_event, SandboxPolicy::new_read_only_policy(), + McpRuntimeEnvironment::new(Arc::new(Environment::default()), config.cwd.to_path_buf()), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth.as_ref()), ToolPluginProvenance::default(), diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 56628f054..8b7935a50 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -205,6 +205,13 @@ impl Session { turn_context.sub_id.clone(), self.get_tx_event(), turn_context.sandbox_policy.get().clone(), + McpRuntimeEnvironment::new( + turn_context + .environment + .clone() + .unwrap_or_else(|| Arc::new(Environment::default())), + turn_context.cwd.to_path_buf(), + ), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth.as_ref()), tool_plugin_provenance, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 594a2d103..5cda5debe 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -55,6 +55,7 @@ use codex_login::CodexAuth; use codex_login::auth_env_telemetry::collect_auth_env_telemetry; use codex_login::default_client::originator; use codex_mcp::McpConnectionManager; +use codex_mcp::McpRuntimeEnvironment; use codex_mcp::ToolInfo; use codex_mcp::codex_apps_tools_cache_key; #[cfg(test)] diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 766ac79ec..ac5be4421 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -676,7 +676,7 @@ impl Session { code_mode_service: crate::tools::code_mode::CodeModeService::new( config.js_repl_node_path.clone(), ), - environment, + environment: environment.clone(), }; services .model_client @@ -770,6 +770,12 @@ impl Session { INITIAL_SUBMIT_ID.to_owned(), tx_event.clone(), session_configuration.sandbox_policy.get().clone(), + McpRuntimeEnvironment::new( + environment + .clone() + .unwrap_or_else(|| Arc::new(Environment::default())), + session_configuration.cwd.to_path_buf(), + ), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index a32f21f7c..7875f0981 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -1,17 +1,26 @@ +#![allow(clippy::expect_used)] + +use anyhow::Context as _; +use anyhow::ensure; use std::collections::HashMap; use std::ffi::OsStr; use std::ffi::OsString; use std::fs; use std::net::TcpListener; use std::path::Path; +use std::path::PathBuf; +use std::process::Command as StdCommand; use std::sync::Arc; +use std::sync::Mutex; use std::time::Duration; use std::time::SystemTime; use std::time::UNIX_EPOCH; use codex_config::types::McpServerConfig; +use codex_config::types::McpServerEnvVar; use codex_config::types::McpServerTransportConfig; use codex_core::config::Config; +use codex_exec_server::CreateDirectoryOptions; use codex_features::Feature; use codex_login::CodexAuth; use codex_mcp::MCP_SANDBOX_STATE_META_CAPABILITY; @@ -34,6 +43,7 @@ use codex_protocol::protocol::SandboxPolicy; use codex_protocol::user_input::UserInput; use codex_utils_cargo_bin::cargo_bin; use core_test_support::assert_regex_match; +use core_test_support::remote_env_env_var; use core_test_support::responses; use core_test_support::responses::ev_custom_tool_call; use core_test_support::responses::mount_models_once; @@ -54,6 +64,7 @@ use tokio::process::Child; use tokio::process::Command; use tokio::time::Instant; use tokio::time::sleep; +use wiremock::MockServer; static OPENAI_PNG: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAD0AAAA9CAYAAAAeYmHpAAAE6klEQVR4Aeyau44UVxCGx1fZsmRLlm3Zoe0XcGQ5cUiCCIgJeS9CHgAhMkISQnIuGQgJEkBcxLW+nqnZ6uqqc+nuWRC7q/P3qetf9e+MtOwyX25O4Nep6JPyop++0qev9HrfgZ+F6r2DuB/vHOrt/UIkqdDHYvujOW6fO7h/CNEI+a5jc+pBR8uy0jVFsziYu5HtfSUk+Io34q921hLNctFSX0gwww+S8wce8K1LfCU+cYW4888aov8NxqvQILUPPReLOrm6zyLxa4i+6VZuFbJo8d1MOHZm+7VUtB/aIvhPWc/3SWg49JcwFLlHxuXKjtyloo+YNhuW3VS+WPBuUEMvCFKjEDVgFBQHXrnazpqiSxNZCkQ1kYiozsbm9Oz7l4i2Il7vGccGNWAc3XosDrZe/9P3ZnMmzHNEQw4smf8RQ87XEAMsC7Az0Au+dgXerfH4+sHvEc0SYGic8WBBUGqFH2gN7yDrazy7m2pbRTeRmU3+MjZmr1h6LJgPbGy23SI6GlYT0brQ71IY8Us4PNQCm+zepSbaD2BY9xCaAsD9IIj/IzFmKMSdHHonwdZATbTnYREf6/VZGER98N9yCWIvXQwXDoDdhZJoT8jwLnJXDB9w4Sb3e6nK5ndzlkTLnP3JBu4LKkbrYrU69gCVceV0JvpyuW1xlsUVngzhwMetn/XamtTORF9IO5YnWNiyeF9zCAfqR3fUW+vZZKLtgP+ts8BmQRBREAdRDhH3o8QuRh/YucNFz2BEjxbRN6LGzphfKmvP6v6QhqIQyZ8XNJ0W0X83MR1PEcJBNO2KC2Z1TW/v244scp9FwRViZxIOBF0Lctk7ZVSavdLvRlV1hz/ysUi9sr8CIcB3nvWBwA93ykTz18eAYxQ6N/K2DkPA1lv3iXCwmDUT7YkjIby9siXueIJj9H+pzSqJ9oIuJWTUgSSt4WO7o/9GGg0viR4VinNRUDoIj34xoCd6pxD3aK3zfdbnx5v1J3ZNNEJsE0sBG7N27ReDrJc4sFxz7dI/ZAbOmmiKvHBitQXpAdR6+F7v+/ol/tOouUV01EeMZQF2BoQDn6dP4XNr+j9GZEtEK1/L8pFw7bd3a53tsTa7WD+054jOFmPg1XBKPQgnqFfmFcy32ZRvjmiIIQTYFvyDxQ8nH8WIwwGwlyDjDznnilYyFr6njrlZwsKkBpO59A7OwgdzPEWRm+G+oeb7IfyNuzjEEVLrOVxJsxvxwF8kmCM6I2QYmJunz4u4TrADpfl7mlbRTWQ7VmrBzh3+C9f6Grc3YoGN9dg/SXFthpRsT6vobfXRs2VBlgBHXVMLHjDNbIZv1sZ9+X3hB09cXdH1JKViyG0+W9bWZDa/r2f9zAFR71sTzGpMSWz2iI4YssWjWo3REy1MDGjdwe5e0dFSiAC1JakBvu4/CUS8Eh6dqHdU0Or0ioY3W5ClSqDXAy7/6SRfgw8vt4I+tbvvNtFT2kVDhY5+IGb1rCqYaXNF08vSALsXCPmt0kQNqJT1p5eI1mkIV/BxCY1z85lOzeFbPBQHURkkPTlwTYK9gTVE25l84IbFFN+YJDHjdpn0gq6mrHht0dkcjbM4UL9283O5p77GN+SPW/QwVB4IUYg7Or+Kp7naR6qktP98LNF2UxWo9yObPIT9KYg+hK4i56no4rfnM0qeyFf6AwAAAP//trwR3wAAAAZJREFUAwBZ0sR75itw5gAAAABJRU5ErkJggg=="; @@ -86,6 +97,73 @@ enum McpCallEvent { End(String), } +const REMOTE_MCP_ENVIRONMENT: &str = "remote"; + +fn remote_aware_experimental_environment() -> Option { + // These tests run locally in normal CI and against the Docker-backed + // executor in full-ci. Match that shared test environment instead of + // parameterizing each stdio MCP test with its own local/remote cases. + std::env::var_os(remote_env_env_var()).map(|_| REMOTE_MCP_ENVIRONMENT.to_string()) +} + +/// Returns the stdio MCP test server command path for the active test placement. +/// +/// Local test runs can execute the host-built test binary directly. Remote-aware +/// runs start MCP stdio through the executor inside Docker, so the host path +/// would be meaningless to the process that actually launches the server. When +/// the remote test environment is active, copy the binary into the executor +/// container and return that in-container path instead. +fn remote_aware_stdio_server_bin() -> anyhow::Result { + let bin = stdio_server_bin()?; + let Some(container_name) = std::env::var_os(remote_env_env_var()) else { + return Ok(bin); + }; + let container_name = container_name + .into_string() + .map_err(|value| anyhow::anyhow!("remote env container name must be utf-8: {value:?}"))?; + + // Keep the Docker path rewrite scoped to tests that use `build_remote_aware`. + // Other MCP tests still start their stdio server from the orchestrator test + // process, even when the full-ci remote env is present. + // + // Remote-aware MCP tests run the executor inside Docker. The stdio test + // server is built on the host, so hand the executor a copied in-container + // path instead of the host build artifact path. + // Several remote-aware MCP tests can run in parallel; give each copied + // binary its own path so one test cannot replace another test's executable. + let unique_suffix = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos(); + let remote_path = format!( + "/tmp/codex-remote-env/test_stdio_server-{}-{unique_suffix}", + std::process::id() + ); + let container_target = format!("{container_name}:{remote_path}"); + let copy_output = StdCommand::new("docker") + .arg("cp") + .arg(&bin) + .arg(&container_target) + .output() + .with_context(|| format!("copy {bin} to remote MCP test env"))?; + ensure!( + copy_output.status.success(), + "docker cp test_stdio_server failed: stdout={} stderr={}", + String::from_utf8_lossy(©_output.stdout).trim(), + String::from_utf8_lossy(©_output.stderr).trim() + ); + + let chmod_output = StdCommand::new("docker") + .args(["exec", &container_name, "chmod", "+x", remote_path.as_str()]) + .output() + .context("mark remote test_stdio_server executable")?; + ensure!( + chmod_output.status.success(), + "docker chmod test_stdio_server failed: stdout={} stderr={}", + String::from_utf8_lossy(&chmod_output.stdout).trim(), + String::from_utf8_lossy(&chmod_output.stderr).trim() + ); + + Ok(remote_path) +} + async fn wait_for_mcp_tool(fixture: &TestCodex, tool_name: &str) -> anyhow::Result<()> { let tools_ready_deadline = Instant::now() + Duration::from_secs(30); loop { @@ -115,6 +193,7 @@ async fn wait_for_mcp_tool(fixture: &TestCodex, tool_name: &str) -> anyhow::Resu #[derive(Default)] struct TestMcpServerOptions { + experimental_environment: Option, supports_parallel_tool_calls: bool, tool_timeout_sec: Option, } @@ -122,14 +201,23 @@ struct TestMcpServerOptions { fn stdio_transport( command: String, env: Option>, - env_vars: Vec, + env_vars: Vec, +) -> McpServerTransportConfig { + stdio_transport_with_cwd(command, env, env_vars, /*cwd*/ None) +} + +fn stdio_transport_with_cwd( + command: String, + env: Option>, + env_vars: Vec, + cwd: Option, ) -> McpServerTransportConfig { McpServerTransportConfig::Stdio { command, args: Vec::new(), env, env_vars, - cwd: None, + cwd, } } @@ -144,7 +232,7 @@ fn insert_mcp_server( server_name.to_string(), McpServerConfig { transport, - experimental_environment: None, + experimental_environment: options.experimental_environment, enabled: true, required: false, supports_parallel_tool_calls: options.supports_parallel_tool_calls, @@ -164,6 +252,105 @@ fn insert_mcp_server( } } +async fn call_cwd_tool( + server: &MockServer, + fixture: &TestCodex, + server_name: &str, + call_id: &str, +) -> anyhow::Result { + let namespace = format!("mcp__{server_name}__"); + mount_sse_once( + server, + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call_with_namespace(call_id, &namespace, "cwd", r#"{}"#), + responses::ev_completed("resp-1"), + ]), + ) + .await; + mount_sse_once( + server, + responses::sse(vec![ + responses::ev_assistant_message("msg-1", "rmcp cwd tool completed successfully."), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + let session_model = fixture.session_configured.model.clone(); + fixture + .codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "call the rmcp cwd tool".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: fixture.config.cwd.to_path_buf(), + approval_policy: AskForApproval::Never, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + model: session_model, + effort: None, + summary: None, + service_tier: None, + collaboration_mode: None, + personality: None, + }) + .await?; + + wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallBegin(_)) + }) + .await; + let end_event = wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(end) = end_event else { + unreachable!("event guard guarantees McpToolCallEnd"); + }; + let structured_content = end + .result + .as_ref() + .expect("rmcp cwd tool should return success") + .structured_content + .as_ref() + .expect("structured content") + .clone(); + + wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + Ok(structured_content) +} + +fn assert_cwd_tool_output(structured: &Value, expected_cwd: &Path) { + let actual_cwd = structured + .get("cwd") + .and_then(Value::as_str) + .expect("cwd tool should return a string cwd"); + + if std::env::var_os(remote_env_env_var()).is_some() { + assert_eq!( + structured, + &json!({ + "cwd": expected_cwd.to_string_lossy(), + }) + ); + return; + } + + // Local Windows can report the same absolute directory through an 8.3 path. + // Canonical paths keep the assertion focused on cwd precedence. + assert_eq!( + Path::new(actual_cwd) + .canonicalize() + .expect("cwd tool path should exist"), + expected_cwd + .canonicalize() + .expect("expected cwd should exist"), + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial(mcp_test_value)] async fn stdio_server_round_trip() -> anyhow::Result<()> { @@ -199,7 +386,7 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> { .await; let expected_env_value = "propagated-env"; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { @@ -214,10 +401,13 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> { )])), Vec::new(), ), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -314,6 +504,118 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_cwd)] +async fn stdio_server_uses_configured_cwd_before_runtime_fallback() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let server_name = "rmcp_configured_cwd"; + let expected_cwd = Arc::new(Mutex::new(None::)); + let expected_cwd_for_config = Arc::clone(&expected_cwd); + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; + + let fixture = test_codex() + .with_workspace_setup(|cwd, fs| async move { + fs.create_directory( + &cwd.join("mcp-configured-cwd"), + CreateDirectoryOptions { recursive: true }, + /*sandbox*/ None, + ) + .await?; + Ok::<(), anyhow::Error>(()) + }) + .with_config(move |config| { + let configured_cwd = config.cwd.join("mcp-configured-cwd").into_path_buf(); + *expected_cwd_for_config + .lock() + .expect("expected cwd lock should not be poisoned") = Some(configured_cwd.clone()); + insert_mcp_server( + config, + server_name, + stdio_transport_with_cwd( + rmcp_test_server_bin, + Some(HashMap::from([( + "MCP_TEST_VALUE".to_string(), + "configured-cwd".to_string(), + )])), + Vec::new(), + Some(configured_cwd), + ), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, + ); + }) + .build_remote_aware(&server) + .await?; + + let expected_cwd = expected_cwd + .lock() + .expect("expected cwd lock should not be poisoned") + .clone() + .expect("test config should record configured MCP cwd"); + let structured = call_cwd_tool(&server, &fixture, server_name, "call-configured-cwd").await?; + + assert_cwd_tool_output(&structured, &expected_cwd); + server.verify().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_cwd)] +async fn remote_stdio_server_uses_runtime_fallback_cwd_when_config_omits_cwd() -> anyhow::Result<()> +{ + skip_if_no_network!(Ok(())); + if std::env::var_os(remote_env_env_var()).is_none() { + return Ok(()); + } + + let server = responses::start_mock_server().await; + let server_name = "rmcp_fallback_cwd"; + let expected_cwd = Arc::new(Mutex::new(None::)); + let expected_cwd_for_config = Arc::clone(&expected_cwd); + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; + + let fixture = test_codex() + .with_config(move |config| { + *expected_cwd_for_config + .lock() + .expect("expected cwd lock should not be poisoned") = + Some(config.cwd.to_path_buf()); + insert_mcp_server( + config, + server_name, + stdio_transport( + rmcp_test_server_bin, + Some(HashMap::from([( + "MCP_TEST_VALUE".to_string(), + "fallback-cwd".to_string(), + )])), + Vec::new(), + ), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, + ); + }) + .build_remote_aware(&server) + .await?; + + let expected_cwd = expected_cwd + .lock() + .expect("expected cwd lock should not be poisoned") + .clone() + .expect("test config should record runtime fallback cwd"); + let structured = call_cwd_tool(&server, &fixture, server_name, "call-fallback-cwd").await?; + + assert_cwd_tool_output(&structured, &expected_cwd); + server.verify().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn stdio_mcp_tool_call_includes_sandbox_state_meta() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); @@ -343,17 +645,20 @@ async fn stdio_mcp_tool_call_includes_sandbox_state_meta() -> anyhow::Result<()> ) .await; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { insert_mcp_server( config, server_name, stdio_transport(rmcp_test_server_bin, /*env*/ None, Vec::new()), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let tools_ready_deadline = Instant::now() + Duration::from_secs(30); @@ -415,7 +720,7 @@ async fn stdio_mcp_tool_call_includes_sandbox_state_meta() -> anyhow::Result<()> ); assert_eq!( sandbox_meta.get("sandboxCwd").and_then(Value::as_str), - fixture.cwd.path().to_str() + fixture.config.cwd.as_path().to_str() ); assert_eq!(sandbox_meta.get("useLegacyLandlock"), Some(&json!(false))); @@ -455,7 +760,7 @@ async fn stdio_mcp_parallel_tool_calls_default_false_runs_serially() -> anyhow:: ) .await; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { @@ -464,12 +769,13 @@ async fn stdio_mcp_parallel_tool_calls_default_false_runs_serially() -> anyhow:: server_name, stdio_transport(rmcp_test_server_bin, /*env*/ None, Vec::new()), TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), tool_timeout_sec: Some(Duration::from_secs(2)), ..Default::default() }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -586,7 +892,7 @@ async fn stdio_mcp_parallel_tool_calls_opt_in_runs_concurrently() -> anyhow::Res ) .await; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { @@ -595,12 +901,13 @@ async fn stdio_mcp_parallel_tool_calls_opt_in_runs_concurrently() -> anyhow::Res server_name, stdio_transport(rmcp_test_server_bin, /*env*/ None, Vec::new()), TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), supports_parallel_tool_calls: true, tool_timeout_sec: Some(Duration::from_secs(2)), }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -676,7 +983,7 @@ async fn stdio_image_responses_round_trip() -> anyhow::Result<()> { .await; // Build the stdio rmcp server and pass the image as data URL so it can construct ImageContent. - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { @@ -691,10 +998,13 @@ async fn stdio_image_responses_round_trip() -> anyhow::Result<()> { )])), Vec::new(), ), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -830,7 +1140,7 @@ async fn stdio_image_responses_preserve_original_detail_metadata() -> anyhow::Re ) .await; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_model("gpt-5.3-codex") @@ -839,10 +1149,13 @@ async fn stdio_image_responses_preserve_original_detail_metadata() -> anyhow::Re config, server_name, stdio_transport(rmcp_test_server_bin, /*env*/ None, Vec::new()), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -1053,7 +1366,7 @@ async fn stdio_image_responses_are_sanitized_for_text_only_model() -> anyhow::Re ) .await; - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) @@ -1069,10 +1382,13 @@ async fn stdio_image_responses_are_sanitized_for_text_only_model() -> anyhow::Re )])), Vec::new(), ), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; fixture @@ -1168,7 +1484,7 @@ async fn stdio_server_propagates_whitelisted_env_vars() -> anyhow::Result<()> { let expected_env_value = "propagated-env-from-whitelist"; let _guard = EnvVarGuard::set("MCP_TEST_VALUE", OsStr::new(expected_env_value)); - let rmcp_test_server_bin = stdio_server_bin()?; + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; let fixture = test_codex() .with_config(move |config| { @@ -1178,12 +1494,15 @@ async fn stdio_server_propagates_whitelisted_env_vars() -> anyhow::Result<()> { stdio_transport( rmcp_test_server_bin, /*env*/ None, - vec!["MCP_TEST_VALUE".to_string()], + vec!["MCP_TEST_VALUE".into()], ), - TestMcpServerOptions::default(), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, ); }) - .build(&server) + .build_remote_aware(&server) .await?; let session_model = fixture.session_configured.model.clone(); @@ -1262,6 +1581,222 @@ async fn stdio_server_propagates_whitelisted_env_vars() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_env_source)] +async fn stdio_server_propagates_explicit_local_env_var_source() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let call_id = "call-local-source"; + let server_name = "rmcp_local_source"; + let namespace = format!("mcp__{server_name}__"); + let env_name = "MCP_TEST_LOCAL_SOURCE"; + let expected_env_value = "propagated-explicit-local-source"; + + mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call_with_namespace( + call_id, + &namespace, + "echo", + &format!(r#"{{"message":"ping","env_var":"{env_name}"}}"#), + ), + responses::ev_completed("resp-1"), + ]), + ) + .await; + mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_assistant_message("msg-1", "rmcp echo tool completed successfully."), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + let _guard = EnvVarGuard::set(env_name, OsStr::new(expected_env_value)); + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; + + let fixture = test_codex() + .with_config(move |config| { + insert_mcp_server( + config, + server_name, + stdio_transport( + rmcp_test_server_bin, + /*env*/ None, + vec![McpServerEnvVar::Config { + name: env_name.to_string(), + source: Some("local".to_string()), + }], + ), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, + ); + }) + .build_remote_aware(&server) + .await?; + + let session_model = fixture.session_configured.model.clone(); + fixture + .codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "call the rmcp echo tool".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: fixture.cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + model: session_model, + effort: None, + summary: None, + service_tier: None, + collaboration_mode: None, + personality: None, + }) + .await?; + + wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallBegin(_)) + }) + .await; + let end_event = wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(end) = end_event else { + unreachable!("event guard guarantees McpToolCallEnd"); + }; + let structured = end + .result + .as_ref() + .expect("rmcp echo tool should return success") + .structured_content + .as_ref() + .expect("structured content"); + assert_eq!(structured["env"], expected_env_value); + + wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + server.verify().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_env_source)] +async fn remote_stdio_env_var_source_does_not_copy_local_env() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + if std::env::var_os(remote_env_env_var()).is_none() { + return Ok(()); + } + + let server = responses::start_mock_server().await; + let call_id = "call-remote-source"; + let server_name = "rmcp_remote_source"; + let namespace = format!("mcp__{server_name}__"); + let env_name = "MCP_TEST_REMOTE_SOURCE_ONLY"; + + mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call_with_namespace( + call_id, + &namespace, + "echo", + &format!(r#"{{"message":"ping","env_var":"{env_name}"}}"#), + ), + responses::ev_completed("resp-1"), + ]), + ) + .await; + mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_assistant_message("msg-1", "rmcp echo tool completed successfully."), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + let _guard = EnvVarGuard::set(env_name, OsStr::new("local-value-should-not-cross")); + let rmcp_test_server_bin = remote_aware_stdio_server_bin()?; + + let fixture = test_codex() + .with_config(move |config| { + insert_mcp_server( + config, + server_name, + stdio_transport( + rmcp_test_server_bin, + /*env*/ None, + vec![McpServerEnvVar::Config { + name: env_name.to_string(), + source: Some("remote".to_string()), + }], + ), + TestMcpServerOptions { + experimental_environment: remote_aware_experimental_environment(), + ..Default::default() + }, + ); + }) + .build_remote_aware(&server) + .await?; + + let session_model = fixture.session_configured.model.clone(); + fixture + .codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "call the rmcp echo tool".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: fixture.cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + model: session_model, + effort: None, + summary: None, + service_tier: None, + collaboration_mode: None, + personality: None, + }) + .await?; + + wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallBegin(_)) + }) + .await; + let end_event = wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(end) = end_event else { + unreachable!("event guard guarantees McpToolCallEnd"); + }; + let structured = end + .result + .as_ref() + .expect("rmcp echo tool should return success") + .structured_content + .as_ref() + .expect("structured content"); + assert_eq!(structured["env"], Value::Null); + + wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + server.verify().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index aa5ab5eee..b81c224a4 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -15,6 +15,7 @@ axum = { workspace = true, default-features = false, features = [ ] } codex-client = { workspace = true } codex-config = { workspace = true } +codex-exec-server = { workspace = true } codex-keyring-store = { workspace = true } codex-protocol = { workspace = true } codex-utils-pty = { workspace = true } diff --git a/codex-rs/rmcp-client/src/bin/rmcp_test_server.rs b/codex-rs/rmcp-client/src/bin/rmcp_test_server.rs index e513f1278..5bf32124d 100644 --- a/codex-rs/rmcp-client/src/bin/rmcp_test_server.rs +++ b/codex-rs/rmcp-client/src/bin/rmcp_test_server.rs @@ -74,7 +74,6 @@ impl TestToolServer { #[derive(Deserialize)] struct EchoArgs { message: String, - #[allow(dead_code)] env_var: Option, } @@ -125,9 +124,10 @@ impl ServerHandler for TestToolServer { }; let env_snapshot: HashMap = std::env::vars().collect(); + let env_name = args.env_var.as_deref().unwrap_or("MCP_TEST_VALUE"); let structured_content = json!({ "echo": args.message, - "env": env_snapshot.get("MCP_TEST_VALUE"), + "env": env_snapshot.get(env_name), }); Ok(CallToolResult { diff --git a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs index 82bc387d9..4d3a2ba77 100644 --- a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs +++ b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs @@ -68,6 +68,7 @@ impl TestToolServer { let tools = vec![ Self::echo_tool(), Self::echo_dash_tool(), + Self::cwd_tool(), Self::sync_tool(), Self::image_tool(), Self::image_scenario_tool(), @@ -124,7 +125,7 @@ impl TestToolServer { { "type": "string" }, { "type": "null" } ] - } + }, }, "required": ["echo", "env"], "additionalProperties": false @@ -135,6 +136,35 @@ impl TestToolServer { tool } + fn cwd_tool() -> Tool { + #[expect(clippy::expect_used)] + let schema: JsonObject = serde_json::from_value(json!({ + "type": "object", + "properties": {}, + "additionalProperties": false + })) + .expect("cwd tool schema should deserialize"); + + let mut tool = Tool::new( + Cow::Borrowed("cwd"), + Cow::Borrowed("Return the current working directory of this test server process."), + Arc::new(schema), + ); + #[expect(clippy::expect_used)] + let output_schema: JsonObject = serde_json::from_value(json!({ + "type": "object", + "properties": { + "cwd": { "type": "string" } + }, + "required": ["cwd"], + "additionalProperties": false + })) + .expect("cwd tool output schema should deserialize"); + tool.output_schema = Some(Arc::new(output_schema)); + tool.annotations = Some(ToolAnnotations::new().read_only(true)); + tool + } + fn sync_tool() -> Tool { #[expect(clippy::expect_used)] let schema: JsonObject = serde_json::from_value(json!({ @@ -292,7 +322,6 @@ impl TestToolServer { #[derive(Deserialize)] struct EchoArgs { message: String, - #[allow(dead_code)] env_var: Option, } @@ -453,6 +482,17 @@ impl ServerHandler for TestToolServer { is_error: Some(false), meta: None, }), + "cwd" => { + let cwd = std::env::current_dir() + .map(|path| path.to_string_lossy().into_owned()) + .map_err(|err| McpError::internal_error(err.to_string(), None))?; + Ok(CallToolResult { + content: Vec::new(), + structured_content: Some(json!({ "cwd": cwd })), + is_error: Some(false), + meta: None, + }) + } "echo" | "echo-tool" => { let args: EchoArgs = match request.arguments { Some(arguments) => serde_json::from_value(serde_json::Value::Object( @@ -468,9 +508,10 @@ impl ServerHandler for TestToolServer { }; let env_snapshot: HashMap = std::env::vars().collect(); + let env_name = args.env_var.as_deref().unwrap_or("MCP_TEST_VALUE"); let structured_content = json!({ "echo": format!("ECHOING: {}", args.message), - "env": env_snapshot.get("MCP_TEST_VALUE"), + "env": env_snapshot.get(env_name), }); Ok(CallToolResult { diff --git a/codex-rs/rmcp-client/src/executor_process_transport.rs b/codex-rs/rmcp-client/src/executor_process_transport.rs new file mode 100644 index 000000000..41f0b7660 --- /dev/null +++ b/codex-rs/rmcp-client/src/executor_process_transport.rs @@ -0,0 +1,376 @@ +//! rmcp transport adapter for an executor-managed MCP stdio process. +//! +//! This module owns the lower-level byte translation after +//! `stdio_server_launcher` has already started a process through +//! `ExecBackend::start`. It does not choose where the MCP server runs and it +//! does not implement MCP lifecycle behavior. MCP protocol ownership stays in +//! `RmcpClient` and rmcp: +//! +//! 1. rmcp serializes a JSON-RPC message and calls [`Transport::send`]. +//! 2. This transport appends the stdio newline delimiter and writes those bytes +//! to executor `process/write`. +//! 3. The executor writes the bytes to the child process stdin. +//! 4. The child writes newline-delimited JSON-RPC messages to stdout. +//! 5. The executor reports stdout bytes through pushed process events. +//! 6. This transport buffers stdout until it has one full line, deserializes +//! that line, and returns the rmcp message from [`Transport::receive`]. +//! +//! Stderr is deliberately not part of the MCP byte stream. It is logged for +//! diagnostics only, matching the local stdio implementation. + +use std::future::Future; +use std::io; +use std::mem::take; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use codex_exec_server::ExecOutputStream; +use codex_exec_server::ExecProcess; +use codex_exec_server::ExecProcessEvent; +use codex_exec_server::ExecProcessEventReceiver; +use codex_exec_server::ProcessId; +use codex_exec_server::ProcessOutputChunk; +use codex_exec_server::WriteStatus; +use rmcp::service::RoleClient; +use rmcp::service::RxJsonRpcMessage; +use rmcp::service::TxJsonRpcMessage; +use rmcp::transport::Transport; +use serde_json::from_slice; +use serde_json::to_vec; +use tokio::runtime::Handle; +use tokio::sync::broadcast; +use tracing::debug; +use tracing::info; +use tracing::warn; + +static PROCESS_COUNTER: AtomicUsize = AtomicUsize::new(1); + +// Remote public implementation. + +/// A client-side rmcp transport backed by an executor-managed process. +/// +/// The orchestrator owns this value and calls rmcp on it. The process it wraps +/// may be local or remote depending on the `ExecBackend` used to create it, but +/// for remote MCP stdio the process lives on the executor and all interaction +/// crosses the executor process RPC boundary. +pub(super) struct ExecutorProcessTransport { + /// Logical process handle returned by the executor process API. + /// + /// `write` forwards stdin bytes. `terminate` stops the child when rmcp + /// closes the transport. + process: Arc, + + /// Pushed output/lifecycle stream for the process. + /// + /// The executor process API still supports retained-output reads, but MCP + /// stdio is naturally streaming. This receiver lets rmcp wait for stdout + /// chunks without issuing `process/read` after each output notification. + events: ExecProcessEventReceiver, + + /// Human-readable program name used only in diagnostics. + program_name: String, + + /// Buffered child stdout bytes that have not yet formed a complete + /// newline-delimited JSON-RPC message. + stdout: Vec, + + /// Buffered stderr bytes for diagnostic logging. + stderr: Vec, + + /// Whether the executor has reported process closure or a terminal + /// subscription failure. Once closed, any remaining partial stdout line is + /// flushed once and then rmcp receives EOF. + closed: bool, + + /// Whether this transport already asked the executor to terminate the MCP + /// server process. + terminated: bool, + + /// Highest executor process event sequence observed by this transport. + /// + /// When the pushed event stream lags, use this as the retained-output read + /// cursor to recover missed stdout/stderr chunks from the executor. + last_seq: u64, +} + +impl ExecutorProcessTransport { + pub(super) fn new(process: Arc, program_name: String) -> Self { + // Subscribe before returning the transport to rmcp. Some test servers + // can emit output or exit quickly after `process/start`, and the + // process event log will replay anything that landed before this + // subscriber was attached. + let events = process.subscribe_events(); + Self { + process, + events, + program_name, + stdout: Vec::new(), + stderr: Vec::new(), + closed: false, + terminated: false, + last_seq: 0, + } + } + + pub(super) fn next_process_id() -> ProcessId { + // Process IDs are logical handles scoped to the executor connection, + // not OS pids. A monotonic client-side id is enough to avoid + // collisions between MCP servers started in the same session. + let index = PROCESS_COUNTER.fetch_add(1, Ordering::Relaxed); + ProcessId::from(format!("mcp-stdio-{index}")) + } +} + +impl Transport for ExecutorProcessTransport { + type Error = io::Error; + + fn send( + &mut self, + item: TxJsonRpcMessage, + ) -> impl Future> + Send + 'static { + let process = Arc::clone(&self.process); + async move { + // rmcp hands us a structured JSON-RPC message. Stdio transport on + // the wire is JSON plus one newline delimiter. + let mut bytes = to_vec(&item).map_err(io::Error::other)?; + bytes.push(b'\n'); + let response = process.write(bytes).await.map_err(io::Error::other)?; + match response.status { + WriteStatus::Accepted => Ok(()), + WriteStatus::UnknownProcess => { + Err(io::Error::new(io::ErrorKind::BrokenPipe, "unknown process")) + } + WriteStatus::StdinClosed => { + Err(io::Error::new(io::ErrorKind::BrokenPipe, "stdin closed")) + } + WriteStatus::Starting => Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process is starting", + )), + } + } + } + + fn receive(&mut self) -> impl Future>> + Send { + self.receive_message() + } + + async fn close(&mut self) -> std::result::Result<(), Self::Error> { + self.process.terminate().await.map_err(io::Error::other)?; + self.terminated = true; + Ok(()) + } +} + +impl ExecutorProcessTransport { + async fn receive_message(&mut self) -> Option> { + loop { + // rmcp stdio framing is line-oriented JSON. We first drain any + // complete line already buffered from an earlier process event. + if let Some(message) = self.take_stdout_message(/*allow_partial*/ self.closed) { + return Some(message); + } + if self.closed { + self.flush_stderr(); + return None; + } + + match self.events.recv().await { + Ok(ExecProcessEvent::Output(chunk)) => { + // The executor pushes raw process bytes. This is the only + // place where those bytes are split back into the stdout + // protocol stream and stderr diagnostics. + self.push_process_output_if_new(chunk); + } + Ok(ExecProcessEvent::Exited { seq, .. }) => { + self.note_seq(seq); + // Wait for `Closed` before ending the rmcp stream so any + // output flushed during process shutdown can still be + // decoded into JSON-RPC messages. + } + Ok(ExecProcessEvent::Closed { seq }) => { + self.note_seq(seq); + self.closed = true; + } + Ok(ExecProcessEvent::Failed(message)) => { + warn!( + "Remote MCP server process failed ({}): {message}", + self.program_name + ); + self.closed = true; + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + "Remote MCP server output stream lagged ({}): skipped {skipped} events", + self.program_name + ); + if let Err(error) = self.recover_lagged_events().await { + warn!( + "Failed to recover remote MCP server output stream ({}): {error}", + self.program_name + ); + self.closed = true; + } + } + Err(broadcast::error::RecvError::Closed) => { + self.closed = true; + } + } + } + } + + fn note_seq(&mut self, seq: u64) { + self.last_seq = self.last_seq.max(seq); + } + + fn should_accept_seq(&mut self, seq: u64) -> bool { + if seq <= self.last_seq { + return false; + } + self.last_seq = seq; + true + } + + async fn recover_lagged_events(&mut self) -> io::Result<()> { + let response = self + .process + .read( + Some(self.last_seq), + /*max_bytes*/ None, + /*wait_ms*/ Some(0), + ) + .await + .map_err(io::Error::other)?; + for chunk in response.chunks { + self.push_process_output_if_new(chunk); + } + self.last_seq = self.last_seq.max(response.next_seq.saturating_sub(1)); + if let Some(message) = response.failure { + warn!( + "Remote MCP server process failed ({}): {message}", + self.program_name + ); + self.closed = true; + } else if response.closed { + self.closed = true; + } + Ok(()) + } + + fn push_process_output_if_new(&mut self, chunk: ProcessOutputChunk) { + if !self.should_accept_seq(chunk.seq) { + return; + } + self.push_process_output(chunk); + } + + fn push_process_output(&mut self, chunk: ProcessOutputChunk) { + let bytes = chunk.chunk.into_inner(); + match chunk.stream { + // MCP stdio uses stdout as the protocol stream. PTY output is + // accepted defensively because the executor process API has a + // unified stream enum, but remote MCP starts with `tty=false`. + ExecOutputStream::Stdout | ExecOutputStream::Pty => { + self.stdout.extend_from_slice(&bytes); + } + // Stderr is intentionally out-of-band. It should help debug server + // startup failures without entering rmcp framing. + ExecOutputStream::Stderr => { + self.push_stderr(&bytes); + } + } + } + + fn take_stdout_message(&mut self, allow_partial: bool) -> Option> { + // A normal MCP stdio server emits one JSON-RPC message per newline. + // If the process has already closed, accept a final unterminated line + // so EOF after a complete JSON object behaves like local rmcp's + // `decode_eof` handling. + loop { + let line_end = self.stdout.iter().position(|byte| *byte == b'\n'); + let line = match (line_end, allow_partial && !self.stdout.is_empty()) { + (Some(index), _) => { + let mut line = self.stdout.drain(..=index).collect::>(); + line.pop(); + line + } + (None, true) => self.stdout.drain(..).collect(), + (None, false) => return None, + }; + let line = Self::trim_trailing_carriage_return(line); + match from_slice::>(&line) { + Ok(message) => return Some(message), + Err(error) => { + debug!( + "Failed to parse remote MCP server message ({}): {error}", + self.program_name + ); + } + } + } + } + + fn push_stderr(&mut self, bytes: &[u8]) { + // Keep stderr line-oriented in logs so a chatty MCP server does not + // produce one log record per byte chunk. + self.stderr.extend_from_slice(bytes); + while let Some(index) = self.stderr.iter().position(|byte| *byte == b'\n') { + let mut line = self.stderr.drain(..=index).collect::>(); + line.pop(); + if line.last() == Some(&b'\r') { + line.pop(); + } + info!( + "MCP server stderr ({}): {}", + self.program_name, + String::from_utf8_lossy(&line) + ); + } + } + + fn flush_stderr(&mut self) { + if self.stderr.is_empty() { + return; + } + let line = take(&mut self.stderr); + info!( + "MCP server stderr ({}): {}", + self.program_name, + String::from_utf8_lossy(&line) + ); + } + + fn trim_trailing_carriage_return(mut line: Vec) -> Vec { + if line.last() == Some(&b'\r') { + line.pop(); + } + line + } +} + +impl Drop for ExecutorProcessTransport { + fn drop(&mut self) { + if self.terminated { + return; + } + + let process = Arc::clone(&self.process); + let program_name = self.program_name.clone(); + let Ok(handle) = Handle::try_current() else { + warn!( + "Could not schedule remote MCP server process termination on drop ({}): no Tokio runtime is available", + self.program_name + ); + return; + }; + + std::mem::drop(handle.spawn(async move { + if let Err(error) = process.terminate().await { + warn!( + "Failed to terminate remote MCP server process on drop ({program_name}): {error}" + ); + } + })); + } +} diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index f02167b6f..f3870b931 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -1,5 +1,6 @@ mod auth_status; mod elicitation_client_service; +mod executor_process_transport; mod logging_client_handler; mod oauth; mod perform_oauth_login; @@ -30,5 +31,6 @@ pub use rmcp_client::ListToolsWithConnectorIdResult; pub use rmcp_client::RmcpClient; pub use rmcp_client::SendElicitation; pub use rmcp_client::ToolWithConnectorId; +pub use stdio_server_launcher::ExecutorStdioServerLauncher; pub use stdio_server_launcher::LocalStdioServerLauncher; pub use stdio_server_launcher::StdioServerLauncher; diff --git a/codex-rs/rmcp-client/src/program_resolver.rs b/codex-rs/rmcp-client/src/program_resolver.rs index d9be2741c..c20cac374 100644 --- a/codex-rs/rmcp-client/src/program_resolver.rs +++ b/codex-rs/rmcp-client/src/program_resolver.rs @@ -157,7 +157,7 @@ mod tests { #[cfg(windows)] extra_env.insert(OsString::from("PATHEXT"), Self::ensure_cmd_extension()); - let mcp_env = create_env_for_mcp_server(Some(extra_env), &[]); + let mcp_env = create_env_for_mcp_server(Some(extra_env), &[])?; Ok(Self { _temp_dir: temp_dir, diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index 6bdc5dd29..b80e335a5 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -13,6 +13,7 @@ use std::time::Instant; use anyhow::Result; use anyhow::anyhow; use codex_client::build_reqwest_client_with_custom_ca; +use codex_config::types::McpServerEnvVar; use futures::FutureExt; use futures::StreamExt; use futures::future::BoxFuture; @@ -503,7 +504,7 @@ impl RmcpClient { program: OsString, args: Vec, env: Option>, - env_vars: &[String], + env_vars: &[McpServerEnvVar], cwd: Option, launcher: Arc, ) -> io::Result { diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index 2fd610e5c..4e9020e36 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -1,12 +1,15 @@ //! Launch MCP stdio servers and return the transport rmcp should use. //! -//! This module owns the "where does the server process run?" boundary for -//! stdio MCP servers. In this PR there is only the local launcher, which keeps -//! the existing behavior: the orchestrator starts the configured command and -//! rmcp talks to the child process through local stdin/stdout pipes. +//! This module owns the "where does the server process run?" decision: //! -//! Later stack entries add an executor-backed launcher without changing -//! `RmcpClient`'s MCP lifecycle code. +//! - [`LocalStdioServerLauncher`] starts the configured command as a child of +//! the orchestrator process. +//! - [`ExecutorStdioServerLauncher`] starts the configured command through the +//! executor process API. +//! +//! Both paths return [`StdioServerTransport`], so `RmcpClient` can hand the +//! resulting byte stream to rmcp without knowing where the process lives. The +//! executor-specific byte adaptation lives in `executor_process_transport`. use std::collections::HashMap; use std::ffi::OsString; @@ -14,6 +17,7 @@ use std::future::Future; use std::io; use std::path::PathBuf; use std::process::Stdio; +use std::sync::Arc; #[cfg(unix)] use std::thread::sleep; #[cfg(unix)] @@ -21,6 +25,13 @@ use std::thread::spawn; #[cfg(unix)] use std::time::Duration; +use anyhow::Result; +use anyhow::anyhow; +use codex_config::types::McpServerEnvVar; +use codex_config::types::ShellEnvironmentPolicyInherit; +use codex_exec_server::ExecBackend; +use codex_exec_server::ExecEnvPolicy; +use codex_exec_server::ExecParams; #[cfg(unix)] use codex_utils_pty::process_group::kill_process_group; #[cfg(unix)] @@ -38,8 +49,11 @@ use tokio::process::Command; use tracing::info; use tracing::warn; +use crate::executor_process_transport::ExecutorProcessTransport; use crate::program_resolver; use crate::utils::create_env_for_mcp_server; +use crate::utils::create_env_overlay_for_remote_mcp_server; +use crate::utils::remote_mcp_env_var_names; // General purpose public code. @@ -63,7 +77,7 @@ pub struct StdioServerCommand { program: OsString, args: Vec, env: Option>, - env_vars: Vec, + env_vars: Vec, cwd: Option, } @@ -74,11 +88,16 @@ pub struct StdioServerCommand { /// directly to `rmcp::service::serve_client`. pub struct StdioServerTransport { inner: StdioServerTransportInner, + // Local child processes can leave subprocesses behind, so the local + // variant keeps a process-group guard with the transport. Executor-backed + // processes are owned and cleaned up by the executor, so that variant uses + // `None`. _process_group_guard: Option, } enum StdioServerTransportInner { Local(TokioChildProcess), + Executor(ExecutorProcessTransport), } impl Transport for StdioServerTransport { @@ -88,20 +107,29 @@ impl Transport for StdioServerTransport { &mut self, item: TxJsonRpcMessage, ) -> impl Future> + Send + 'static { + // Both variants already implement rmcp's transport contract. This + // wrapper keeps process placement private while leaving rmcp's send + // semantics unchanged. match &mut self.inner { StdioServerTransportInner::Local(transport) => transport.send(item).boxed(), + StdioServerTransportInner::Executor(transport) => transport.send(item).boxed(), } } fn receive(&mut self) -> impl Future>> + Send { + // rmcp reads from the same transport shape for both placements. The + // executor variant turns pushed process-output events back into the + // line-delimited JSON stream expected by rmcp. match &mut self.inner { StdioServerTransportInner::Local(transport) => transport.receive().boxed(), + StdioServerTransportInner::Executor(transport) => transport.receive().boxed(), } } async fn close(&mut self) -> std::result::Result<(), Self::Error> { match &mut self.inner { StdioServerTransportInner::Local(transport) => transport.close().await, + StdioServerTransportInner::Executor(transport) => transport.close().await, } } } @@ -113,7 +141,7 @@ impl StdioServerCommand { program: OsString, args: Vec, env: Option>, - env_vars: Vec, + env_vars: Vec, cwd: Option, ) -> Self { Self { @@ -174,7 +202,7 @@ impl LocalStdioServerLauncher { cwd, } = command; let program_name = program.to_string_lossy().into_owned(); - let envs = create_env_for_mcp_server(env, &env_vars); + let envs = create_env_for_mcp_server(env, &env_vars).map_err(io::Error::other)?; let resolved_program = program_resolver::resolve(program, &envs).map_err(io::Error::other)?; @@ -266,3 +294,229 @@ impl Drop for ProcessGroupGuard { } } } + +// Remote public implementation. + +/// Starts MCP stdio servers through the executor process API. +/// +/// 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`. +#[derive(Clone)] +pub struct ExecutorStdioServerLauncher { + exec_backend: Arc, + fallback_cwd: PathBuf, +} + +impl ExecutorStdioServerLauncher { + /// Creates a stdio server launcher backed by the executor process API. + /// + /// `fallback_cwd` is used only when the MCP server config omits `cwd`. + /// Executor `process/start` requires an explicit working directory, unlike + /// local `tokio::process::Command`, which can inherit the orchestrator cwd. + pub fn new(exec_backend: Arc, fallback_cwd: PathBuf) -> Self { + Self { + exec_backend, + fallback_cwd, + } + } +} + +impl StdioServerLauncher for ExecutorStdioServerLauncher { + fn launch( + &self, + command: StdioServerCommand, + ) -> BoxFuture<'static, io::Result> { + let exec_backend = Arc::clone(&self.exec_backend); + let fallback_cwd = self.fallback_cwd.clone(); + async move { Self::launch_server(command, exec_backend, fallback_cwd).await }.boxed() + } +} + +// Remote private implementation. + +impl private::Sealed for ExecutorStdioServerLauncher {} + +impl ExecutorStdioServerLauncher { + async fn launch_server( + command: StdioServerCommand, + exec_backend: Arc, + fallback_cwd: PathBuf, + ) -> io::Result { + let StdioServerCommand { + program, + args, + env, + env_vars, + cwd, + } = command; + let program_name = program.to_string_lossy().into_owned(); + let envs = create_env_overlay_for_remote_mcp_server(env, &env_vars); + let remote_env_vars = remote_mcp_env_var_names(&env_vars); + // The executor protocol carries argv/env as UTF-8 strings. Local stdio can + // accept arbitrary OsString values because it calls the OS directly; remote + // stdio must reject non-Unicode command, argument, or environment data + // before sending an executor request. + let argv = Self::process_api_argv(&program, &args).map_err(io::Error::other)?; + let env = Self::process_api_env(envs).map_err(io::Error::other)?; + let process_id = ExecutorProcessTransport::next_process_id(); + // Start the MCP server process on the executor with raw pipes. `tty=false` + // keeps stdout as a clean protocol stream, while `pipe_stdin=true` lets + // rmcp write JSON-RPC requests after the process starts. + let started = exec_backend + .start(ExecParams { + process_id, + argv, + cwd: cwd.unwrap_or(fallback_cwd), + env_policy: Some(Self::remote_env_policy(&remote_env_vars)), + env, + tty: false, + pipe_stdin: true, + arg0: None, + }) + .await + .map_err(io::Error::other)?; + + Ok(StdioServerTransport { + inner: StdioServerTransportInner::Executor(ExecutorProcessTransport::new( + started.process, + program_name, + )), + _process_group_guard: None, + }) + } + + fn process_api_argv(program: &OsString, args: &[OsString]) -> Result> { + let mut argv = Vec::with_capacity(args.len() + 1); + argv.push(Self::os_string_to_process_api_string( + program.clone(), + "command", + )?); + for arg in args { + argv.push(Self::os_string_to_process_api_string( + arg.clone(), + "argument", + )?); + } + Ok(argv) + } + + fn process_api_env(env: HashMap) -> Result> { + env.into_iter() + .map(|(key, value)| { + Ok(( + Self::os_string_to_process_api_string(key, "environment variable name")?, + Self::os_string_to_process_api_string(value, "environment variable value")?, + )) + }) + .collect() + } + + fn os_string_to_process_api_string(value: OsString, label: &str) -> Result { + value + .into_string() + .map_err(|_| anyhow!("{label} must be valid Unicode for remote MCP stdio")) + } + + fn remote_env_policy(remote_env_vars: &[String]) -> ExecEnvPolicy { + let include_only = if remote_env_vars.is_empty() { + Vec::new() + } else { + // `source = "remote"` means the value is read from the executor's + // environment, not copied from Codex. Start from `All` only so the + // named remote variable is available to the filter below; the + // effective child env is still limited by `include_only`. + crate::utils::DEFAULT_ENV_VARS + .iter() + .map(|name| (*name).to_string()) + .chain(remote_env_vars.iter().cloned()) + .collect() + }; + ExecEnvPolicy { + inherit: if remote_env_vars.is_empty() { + ShellEnvironmentPolicyInherit::Core + } else { + ShellEnvironmentPolicyInherit::All + }, + ignore_default_excludes: true, + exclude: Vec::new(), + r#set: HashMap::new(), + include_only, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_config::shell_environment; + use codex_config::types::EnvironmentVariablePattern; + use codex_config::types::ShellEnvironmentPolicy; + + #[test] + fn remote_env_policy_uses_core_env_without_remote_source_vars() { + let policy = ExecutorStdioServerLauncher::remote_env_policy(&[]); + + assert_eq!(policy.inherit, ShellEnvironmentPolicyInherit::Core); + assert!(policy.include_only.is_empty()); + } + + #[test] + fn remote_env_policy_includes_remote_source_vars_without_full_env() { + let policy = ExecutorStdioServerLauncher::remote_env_policy(&["REMOTE_TOKEN".to_string()]); + + assert_eq!(policy.inherit, ShellEnvironmentPolicyInherit::All); + assert!( + policy.include_only.contains(&"REMOTE_TOKEN".to_string()), + "remote source var should be included in executor env policy" + ); + assert!( + policy + .include_only + .contains(&crate::utils::DEFAULT_ENV_VARS[0].to_string()), + "remote default env vars should remain available" + ); + } + + #[test] + fn remote_env_policy_effectively_filters_unrequested_vars() { + let exec_policy = + ExecutorStdioServerLauncher::remote_env_policy(&["REMOTE_TOKEN".to_string()]); + let policy = ShellEnvironmentPolicy { + inherit: exec_policy.inherit, + ignore_default_excludes: exec_policy.ignore_default_excludes, + exclude: exec_policy + .exclude + .iter() + .map(|pattern| EnvironmentVariablePattern::new_case_insensitive(pattern)) + .collect(), + r#set: exec_policy.r#set, + include_only: exec_policy + .include_only + .iter() + .map(|pattern| EnvironmentVariablePattern::new_case_insensitive(pattern)) + .collect(), + use_profile: false, + }; + + let env = shell_environment::create_env_from_vars( + [ + ("PATH".to_string(), "/remote/bin".to_string()), + ("REMOTE_TOKEN".to_string(), "remote-secret".to_string()), + ( + "UNREQUESTED_SECRET".to_string(), + "must-not-pass".to_string(), + ), + ], + &policy, + /*thread_id*/ None, + ); + + assert_eq!(env.get("PATH").map(String::as_str), Some("/remote/bin")); + assert_eq!( + env.get("REMOTE_TOKEN").map(String::as_str), + Some("remote-secret") + ); + assert!(!env.contains_key("UNREQUESTED_SECRET")); + } +} diff --git a/codex-rs/rmcp-client/src/utils.rs b/codex-rs/rmcp-client/src/utils.rs index cf8d4cbb2..6176f071c 100644 --- a/codex-rs/rmcp-client/src/utils.rs +++ b/codex-rs/rmcp-client/src/utils.rs @@ -1,4 +1,6 @@ use anyhow::Result; +use anyhow::anyhow; +use codex_config::types::McpServerEnvVar; use reqwest::ClientBuilder; use reqwest::header::HeaderMap; use reqwest::header::HeaderName; @@ -9,17 +11,52 @@ use std::ffi::OsString; pub(crate) fn create_env_for_mcp_server( extra_env: Option>, - env_vars: &[String], -) -> HashMap { - DEFAULT_ENV_VARS + env_vars: &[McpServerEnvVar], +) -> Result> { + let additional_env_vars = local_stdio_env_var_names(env_vars)?; + let env = DEFAULT_ENV_VARS .iter() .copied() - .chain(env_vars.iter().map(String::as_str)) + .chain(additional_env_vars) .filter_map(|var| env::var_os(var).map(|value| (OsString::from(var), value))) .chain(extra_env.unwrap_or_default()) + .collect(); + Ok(env) +} + +pub(crate) fn create_env_overlay_for_remote_mcp_server( + extra_env: Option>, + env_vars: &[McpServerEnvVar], +) -> HashMap { + // Remote stdio should inherit PATH/HOME/etc. from the executor side, not + // from the orchestrator process. Only forward variables explicitly named + // by the MCP config plus literal env overrides from that config. + env_vars + .iter() + .filter(|var| !var.is_remote_source()) + .filter_map(|var| env::var_os(var.name()).map(|value| (OsString::from(var.name()), value))) + .chain(extra_env.unwrap_or_default()) .collect() } +pub(crate) fn remote_mcp_env_var_names(env_vars: &[McpServerEnvVar]) -> Vec { + env_vars + .iter() + .filter(|var| var.is_remote_source()) + .map(|var| var.name().to_string()) + .collect() +} + +fn local_stdio_env_var_names(env_vars: &[McpServerEnvVar]) -> Result> { + if let Some(remote_var) = env_vars.iter().find(|var| var.is_remote_source()) { + return Err(anyhow!( + "env_vars entry `{}` uses source `remote`, which requires remote MCP stdio", + remote_var.name() + )); + } + Ok(env_vars.iter().map(McpServerEnvVar::name)) +} + pub(crate) fn build_default_headers( http_headers: Option>, env_http_headers: Option>, @@ -182,7 +219,8 @@ mod tests { let env = create_env_for_mcp_server( Some(HashMap::from([(OsString::from("TZ"), expected.clone())])), &[], - ); + ) + .expect("local MCP env should build"); assert_eq!(env.get(OsStr::new("TZ")), Some(&expected)); } @@ -193,10 +231,92 @@ mod tests { let value = "from-env"; let expected = OsString::from(value); let _guard = EnvVarGuard::set(custom_var, value); - let env = create_env_for_mcp_server(/*extra_env*/ None, &[custom_var.to_string()]); + let env = create_env_for_mcp_server(/*extra_env*/ None, &[custom_var.into()]) + .expect("local MCP env should build"); assert_eq!(env.get(OsStr::new(custom_var)), Some(&expected)); } + #[test] + #[serial(extra_rmcp_env)] + fn create_remote_env_overlay_only_forwards_explicit_variables() { + let default_var = DEFAULT_ENV_VARS[0]; + let custom_var = "EXTRA_REMOTE_RMCP_ENV"; + let custom_value = OsString::from("from-env"); + let _default_guard = EnvVarGuard::set(default_var, "from-default"); + let _custom_guard = EnvVarGuard::set(custom_var, &custom_value); + + let env = + create_env_overlay_for_remote_mcp_server(/*extra_env*/ None, &[custom_var.into()]); + + assert_eq!( + env, + HashMap::from([(OsString::from(custom_var), custom_value)]) + ); + } + + #[test] + #[serial(extra_rmcp_env)] + fn create_remote_env_overlay_does_not_copy_remote_source_variables() { + let remote_var = "REMOTE_ONLY_RMCP_ENV"; + let local_var = "LOCAL_RMCP_ENV"; + let local_value = OsString::from("from-local-env"); + let _remote_guard = EnvVarGuard::set(remote_var, "should-not-be-copied"); + let _local_guard = EnvVarGuard::set(local_var, &local_value); + + let env = create_env_overlay_for_remote_mcp_server( + /*extra_env*/ None, + &[ + McpServerEnvVar::Config { + name: remote_var.to_string(), + source: Some("remote".to_string()), + }, + McpServerEnvVar::Config { + name: local_var.to_string(), + source: Some("local".to_string()), + }, + ], + ); + + assert_eq!( + env, + HashMap::from([(OsString::from(local_var), local_value)]) + ); + } + + #[test] + fn remote_mcp_env_var_names_returns_remote_source_names() { + let names = remote_mcp_env_var_names(&[ + "LEGACY".into(), + McpServerEnvVar::Config { + name: "LOCAL".to_string(), + source: Some("local".to_string()), + }, + McpServerEnvVar::Config { + name: "REMOTE".to_string(), + source: Some("remote".to_string()), + }, + ]); + + assert_eq!(names, vec!["REMOTE".to_string()]); + } + + #[test] + fn create_local_env_rejects_remote_source_variables() { + let err = create_env_for_mcp_server( + /*extra_env*/ None, + &[McpServerEnvVar::Config { + name: "REMOTE".to_string(), + source: Some("remote".to_string()), + }], + ) + .expect_err("remote source should require remote stdio"); + + assert!( + err.to_string().contains("requires remote MCP stdio"), + "unexpected error: {err}" + ); + } + #[cfg(unix)] #[test] #[serial(extra_rmcp_env)] @@ -207,7 +327,8 @@ mod tests { let expected = raw_path.to_os_string(); let _guard = EnvVarGuard::set("PATH", raw_path); - let env = create_env_for_mcp_server(/*extra_env*/ None, &[]); + let env = + create_env_for_mcp_server(/*extra_env*/ None, &[]).expect("local MCP env should build"); assert_eq!(env.get(OsStr::new("PATH")), Some(&expected)); } diff --git a/codex-rs/utils/cli/src/format_env_display.rs b/codex-rs/utils/cli/src/format_env_display.rs index dda0cad09..cee63bb32 100644 --- a/codex-rs/utils/cli/src/format_env_display.rs +++ b/codex-rs/utils/cli/src/format_env_display.rs @@ -1,6 +1,9 @@ use std::collections::HashMap; -pub fn format_env_display(env: Option<&HashMap>, env_vars: &[String]) -> String { +pub fn format_env_display>( + env: Option<&HashMap>, + env_vars: &[S], +) -> String { let mut parts: Vec = Vec::new(); if let Some(map) = env { @@ -10,7 +13,7 @@ pub fn format_env_display(env: Option<&HashMap>, env_vars: &[Str } if !env_vars.is_empty() { - parts.extend(env_vars.iter().map(|var| format!("{var}=*****"))); + parts.extend(env_vars.iter().map(|var| format!("{}=*****", var.as_ref()))); } if parts.is_empty() { @@ -26,10 +29,11 @@ mod tests { #[test] fn returns_dash_when_empty() { - assert_eq!(format_env_display(/*env*/ None, &[]), "-"); + let empty_vars: &[String] = &[]; + assert_eq!(format_env_display(/*env*/ None, empty_vars), "-"); let empty_map = HashMap::new(); - assert_eq!(format_env_display(Some(&empty_map), &[]), "-"); + assert_eq!(format_env_display(Some(&empty_map), empty_vars), "-"); } #[test] @@ -38,7 +42,10 @@ mod tests { env.insert("B".to_string(), "two".to_string()); env.insert("A".to_string(), "one".to_string()); - assert_eq!(format_env_display(Some(&env), &[]), "A=*****, B=*****"); + assert_eq!( + format_env_display(Some(&env), &[] as &[String]), + "A=*****, B=*****" + ); } #[test]