openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-20 12:47:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cache selected plugin metadata by stable root

Browse Source
This commit is contained in:
jif-oai
2026-06-25 17:42:10 +01:00
parent 102ffbc704
commit 338cbd0bf9
3 changed files with 85 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
codex-rs/app-server/tests/suite/v2/executor_mcp.rs
View File

@@ -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![

67
codex-rs/core/src/session/mcp_projection.rs
View File

@@ -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::<Vec<_>>();
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::<Vec<_>>();
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<McpRuntimeSnapshot> {
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
}

79
codex-rs/core/src/session/mcp_runtime.rs
View File

@@ -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<Arc<McpRuntimeSnapshot>>,
plugin_projections: Vec<CachedExecutorPluginProjection>,
runtime: Option<CachedSelectedRuntime>,
}
struct CachedSelectedRuntime {
bindings: Vec<(usize, ResolvedSelectedCapabilityRoot)>,
plugins: Vec<(usize, ExecutorPluginRuntime)>,
runtime: Arc<McpRuntimeSnapshot>,
}
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<Vec<(usize, ExecutorPluginRuntime)>> {
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<McpRuntimeSnapshot>,
) {
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()