diff --git a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs index f23bec8e06..eff275dfc1 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs @@ -429,7 +429,7 @@ startup_timeout_sec = 10 } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn selected_executor_plugin_activates_after_it_becomes_ready() -> Result<()> { +async fn selected_executor_plugin_runtime_survives_thread_resume() -> Result<()> { let responses_server = responses::start_mock_server().await; let (apps_url, apps_server_handle) = start_apps_server_with_delays( vec![AppInfo { @@ -497,40 +497,8 @@ args = ["exec-server", "--listen", "stdio"] ), )?; - // The selected root exists, but its manifest arrives after the first model step. + // Environment contents are complete before selection and remain stable for the thread. let plugin = TempDir::new()?; - let mut app_server = TestAppServer::new(codex_home.path()).await?; - timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??; - let selected_thread = start_thread( - &mut app_server, - Some(vec![SelectedCapabilityRoot { - id: "executor-demo@1".to_string(), - location: CapabilityRootLocation::Environment { - environment_id: EXECUTOR_ID.to_string(), - path: PathUri::from_host_native_path(plugin.path())?, - }, - }]), - ) - .await?; - - let before_ready = responses::mount_sse_once( - &responses_server, - responses::sse(vec![ - responses::ev_response_created("before-ready"), - responses::ev_assistant_message("before-ready-message", "Waiting"), - responses::ev_completed("before-ready"), - ]), - ) - .await; - run_turn(&mut app_server, &selected_thread, "Check available tools").await?; - let namespace = format!("mcp__{DYNAMIC_MCP_SERVER_NAME}"); - assert!( - before_ready - .single_request() - .tool_by_name(&namespace, "echo") - .is_none() - ); - std::fs::create_dir_all(plugin.path().join(".codex-plugin"))?; std::fs::write( plugin.path().join(".codex-plugin/plugin.json"), @@ -556,6 +524,21 @@ args = ["exec-server", "--listen", "stdio"] }))?, )?; + let mut app_server = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??; + let selected_thread = start_thread( + &mut app_server, + Some(vec![SelectedCapabilityRoot { + id: "executor-demo@1".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: EXECUTOR_ID.to_string(), + path: PathUri::from_host_native_path(plugin.path())?, + }, + }]), + ) + .await?; + let namespace = format!("mcp__{DYNAMIC_MCP_SERVER_NAME}"); + let response_mock = responses::mount_sse_sequence( &responses_server, vec![ diff --git a/codex-rs/core/src/session/mcp_projection.rs b/codex-rs/core/src/session/mcp_projection.rs index 7094e76b91..143a779816 100644 --- a/codex-rs/core/src/session/mcp_projection.rs +++ b/codex-rs/core/src/session/mcp_projection.rs @@ -8,7 +8,6 @@ use super::TurnContext; use crate::config::Config; use crate::environment_selection::TurnEnvironmentSnapshot; use codex_config::McpServerConfig; -use codex_core_plugins::ExecutorPluginRuntime; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::ElicitationReviewerHandle; use codex_mcp::McpConfig; @@ -28,7 +27,6 @@ pub(super) enum McpRuntimeScope<'a> { struct ProjectedMcpConfig { bindings: Vec<(usize, ResolvedSelectedCapabilityRoot)>, - plugins: Vec<(usize, ExecutorPluginRuntime)>, config: McpConfig, runtime_context: McpRuntimeContext, } @@ -50,25 +48,7 @@ impl Session { } let bindings = selected_bindings(selected_roots, resolved_roots); let cached_runtime = cache.runtime_for_bindings(&bindings); - let mut plugins = cache.plugins_for_bindings(&bindings).unwrap_or_default(); - let mut discovered_plugin = false; - if cached_runtime.is_some() { - let unresolved = bindings - .iter() - .filter(|(order, _)| { - !plugins - .iter() - .any(|(plugin_order, _)| plugin_order == order) - }) - .cloned() - .collect::>(); - let discovered = project_executor_plugins(&unresolved).await; - discovered_plugin = !discovered.is_empty(); - plugins.extend(discovered); - plugins.sort_unstable_by_key(|(order, _)| *order); - } else { - plugins = project_executor_plugins(&bindings).await; - } + let plugins = cache.project_plugins(&bindings).await; let base = cache.base_runtime(); let mcp_config = Arc::new( @@ -83,8 +63,7 @@ impl Session { .collect::>(); let runtime_context = self.mcp_runtime_context(&turn_context.config, environments, &pinned_roots); - if !discovered_plugin - && let Some(runtime) = cached_runtime + if let Some(runtime) = cached_runtime && runtime.matches_projection(mcp_config.as_ref(), &runtime_context) { runtime @@ -109,7 +88,7 @@ impl Session { ) .await }; - cache.replace_selected_runtime(bindings, plugins, Arc::clone(&runtime)); + cache.replace_selected_runtime(bindings, Arc::clone(&runtime)); self.services .publish_existing_mcp_runtime(Arc::clone(&runtime)); runtime @@ -119,11 +98,16 @@ impl Session { &self, config: &Config, ) -> (McpConfig, McpRuntimeContext) { - let projection = self.project_mcp_config_inner(config).await; + let mut cache = self.services.selected_mcp_runtime.lock().await; + let projection = self.project_mcp_config_inner(config, &mut cache).await; (projection.config, projection.runtime_context) } - async fn project_mcp_config_inner(&self, config: &Config) -> ProjectedMcpConfig { + async fn project_mcp_config_inner( + &self, + config: &Config, + cache: &mut super::SelectedMcpRuntimeCache, + ) -> ProjectedMcpConfig { let environments = self.services.turn_environments.snapshot().await; let selected_roots = &self.services.selected_capability_roots; let resolved_roots = self @@ -136,7 +120,7 @@ impl Session { ) .await; let bindings = selected_bindings(selected_roots, &resolved_roots); - let plugins = project_executor_plugins(&bindings).await; + let plugins = cache.project_plugins(&bindings).await; let base_config = self.services.mcp_manager.runtime_config(config).await; let mcp_config = self .services @@ -145,7 +129,6 @@ impl Session { let runtime_context = self.mcp_runtime_context(config, &environments, &resolved_roots); ProjectedMcpConfig { bindings, - plugins, config: mcp_config, runtime_context, } @@ -156,7 +139,7 @@ impl Session { config: &Config, ) -> Arc { let mut cache = self.services.selected_mcp_runtime.lock().await; - let projection = self.project_mcp_config_inner(config).await; + let projection = self.project_mcp_config_inner(config, &mut cache).await; let current = cache .runtime_for_bindings(&projection.bindings) .unwrap_or_else(|| self.services.latest_mcp_runtime()); @@ -183,11 +166,7 @@ impl Session { if projection.bindings.is_empty() { cache.replace_base_and_invalidate_selected(Arc::clone(&runtime)); } else { - cache.replace_selected_runtime( - projection.bindings, - projection.plugins, - Arc::clone(&runtime), - ); + cache.replace_selected_runtime(projection.bindings, Arc::clone(&runtime)); } self.services .publish_existing_mcp_runtime(Arc::clone(&runtime)); @@ -309,23 +288,3 @@ fn selected_bindings( }) .collect() } - -async fn project_executor_plugins( - bindings: &[(usize, ResolvedSelectedCapabilityRoot)], -) -> Vec<(usize, ExecutorPluginRuntime)> { - let mut plugins = Vec::new(); - for (selection_order, root) in bindings { - match ExecutorPluginRuntime::project(root).await { - Ok(Some(runtime)) => plugins.push((*selection_order, runtime)), - Ok(None) => {} - Err(err) => { - tracing::warn!( - selected_root = root.selected_root().id, - error = %err, - "failed to project selected executor plugin runtime" - ); - } - } - } - plugins -} diff --git a/codex-rs/core/src/session/mcp_runtime.rs b/codex-rs/core/src/session/mcp_runtime.rs index a58d526c7b..263bf96dd9 100644 --- a/codex-rs/core/src/session/mcp_runtime.rs +++ b/codex-rs/core/src/session/mcp_runtime.rs @@ -3,36 +3,43 @@ use std::sync::Arc; use codex_core_plugins::ExecutorPluginRuntime; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::McpRuntimeSnapshot; +use codex_protocol::capabilities::SelectedCapabilityRoot; /// One live selected-plugin MCP runtime retained between model steps. /// -/// A cached runtime is a reuse candidate only for the same ordered selected roots and the same -/// process-local environment instances. The caller additionally compares the effective MCP config -/// and runtime context before reuse. Selected environment contents are treated as stable, so -/// manifest and MCP config file changes do not invalidate this cache. +/// Selected environment identity and contents are stable. Plugin manifests, MCP declarations, and +/// app declarations are therefore cached by the complete selected root for the session lifetime. +/// Missing or failed projections are not cached, so a deferred environment can recover later. +/// +/// A live runtime is a separate cache: it is reusable only for the same ordered selected roots and +/// the same process-local environment handles. The caller additionally compares the effective MCP +/// config and runtime context. Replacing a connection handle may rebuild live processes, but it +/// reuses the stable plugin projection and does not reread capability files. /// /// Within a live session, the selected runtime is invalidated in exactly two ways: /// /// 1. [`Self::replace_base_and_invalidate_selected`] installs a new base MCP runtime. /// 2. [`Self::replace_selected_runtime`] stores a newly projected runtime. This happens when the -/// bindings change, when a previously unavailable plugin appears, or when the effective config -/// or runtime context changes. An unavailable environment disappears from the binding list and -/// therefore follows this path; returning with a new environment instance rebuilds the live -/// runtime even when the stable environment ID is unchanged. +/// active bindings or effective runtime configuration change. /// /// In-flight [`McpRuntimeSnapshot`] values retain their manager until their model step finishes. #[derive(Default)] pub(crate) struct SelectedMcpRuntimeCache { base_runtime: Option>, + plugin_projections: Vec, runtime: Option, } struct CachedSelectedRuntime { bindings: Vec<(usize, ResolvedSelectedCapabilityRoot)>, - plugins: Vec<(usize, ExecutorPluginRuntime)>, runtime: Arc, } +struct CachedExecutorPluginProjection { + selected_root: SelectedCapabilityRoot, + plugin: ExecutorPluginRuntime, +} + impl SelectedMcpRuntimeCache { pub(crate) fn replace_base_and_invalidate_selected( &mut self, @@ -59,27 +66,51 @@ impl SelectedMcpRuntimeCache { .map(|cached| Arc::clone(&cached.runtime)) } - pub(crate) fn plugins_for_bindings( - &self, + pub(crate) async fn project_plugins( + &mut self, bindings: &[(usize, ResolvedSelectedCapabilityRoot)], - ) -> Option> { - self.runtime - .as_ref() - .filter(|cached| same_bindings(&cached.bindings, bindings)) - .map(|cached| cached.plugins.clone()) + ) -> Vec<(usize, ExecutorPluginRuntime)> { + let mut plugins = Vec::new(); + for (selection_order, root) in bindings { + let selected_root = root.selected_root(); + if let Some(plugin) = self + .plugin_projections + .iter() + .find(|cached| &cached.selected_root == selected_root) + .map(|cached| cached.plugin.clone()) + { + plugins.push((*selection_order, plugin)); + continue; + } + + match ExecutorPluginRuntime::project(root).await { + Ok(Some(plugin)) => { + self.plugin_projections + .push(CachedExecutorPluginProjection { + selected_root: selected_root.clone(), + plugin: plugin.clone(), + }); + plugins.push((*selection_order, plugin)); + } + Ok(None) => {} + Err(err) => { + tracing::warn!( + selected_root = selected_root.id, + error = %err, + "failed to project selected executor plugin runtime" + ); + } + } + } + plugins } pub(crate) fn replace_selected_runtime( &mut self, bindings: Vec<(usize, ResolvedSelectedCapabilityRoot)>, - plugins: Vec<(usize, ExecutorPluginRuntime)>, runtime: Arc, ) { - self.runtime = Some(CachedSelectedRuntime { - bindings, - plugins, - runtime, - }); + self.runtime = Some(CachedSelectedRuntime { bindings, runtime }); } } @@ -88,8 +119,8 @@ fn same_bindings( right: &[(usize, ResolvedSelectedCapabilityRoot)], ) -> bool { // Order is part of the key because later selected roots can be renamed when MCP server names - // collide. Arc identity is part of the key because live processes and connections belong to - // one exact environment instance, even when a replacement reuses the same stable ID. + // collide. Arc identity is only a live-connection key: stable plugin metadata is cached above + // by selected root and survives connection-handle replacement. left.len() == right.len() && left .iter()