mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
Discover stdio MCP servers from selected executor plugins (#27870)
## Why **In short:** this PR discovers MCP registrations by reading a selected plugin's `.mcp.json` on its executor. #27884 then resolves those registrations in the shared catalog. `thread/start.selectedCapabilityRoots` can select a plugin root owned by an executor, and Codex can resolve that package through the executor filesystem. MCP declarations inside the selected plugin are still ignored. This PR adds the source-specific discovery layer on top of the selected-plugin catalog boundary in #27884: ```text selected capability root | v resolve the plugin through its executor filesystem | v read and normalize its MCP config through the same filesystem | v contribute stdio registrations bound to that environment ID ``` The existing MCP launcher and connection manager remain unchanged. MCP config parsing is shared with local plugins through #27863. ## What changed - Added an executor plugin MCP provider in the MCP extension. - Retained only the exact filesystem capability used for package resolution and reused it for the selected plugin's MCP config, with no host-filesystem fallback or unrelated process/HTTP authority. - Read either the manifest-declared MCP config or the default `.mcp.json`; a missing default file means the plugin has no MCP servers. - Accepted stdio servers only for this first vertical. Executor-owned HTTP declarations are skipped with a warning until their placement semantics are defined. - Normalized stdio registrations with the owning environment's stable logical ID and plugin-root working directory. - Resolved environment-variable names on the owning executor and rejected explicit local forwarding for non-local plugins. - Froze discovered declarations once per active thread runtime, then applied current managed plugin and MCP requirements when contributing them. - Carried the selected root ID, display name, and selection order into the catalog contribution defined by #27884. ## Behavior and scope There is intentionally no production behavior change yet. This PR provides the executor provider and contribution boundary, but app-server does not install it in this change. Existing local plugin MCP loading is unchanged, and no MCP process is launched by this PR alone. ## Assumptions - The selected root ID is the plugin policy identity; the manifest display name is presentation metadata. - An environment ID is a stable logical authority. Reconnection or replacement under the same ID does not change ownership. - Selected plugin packages and their manifests are trusted inputs. - The selected package and MCP discovery snapshot remain frozen for the active thread runtime. ## Follow-up The next PR installs this contributor in app-server and adds an end-to-end test proving that a selected plugin MCP tool launches on its owning executor, can be called by the model, survives an explicit MCP refresh, and is invisible when its root was not selected. Resume, fork, environment removal or ID changes, dynamic catalog reload, and executor-owned HTTP MCP placement remain separate lifecycle decisions. ## Verification Focused tests cover executor-only filesystem reads, missing and malformed config, stdio filtering and normalization, managed requirements, package attribution, and selection order. CI owns execution of the test suite.
This commit is contained in:
jif
GitHub
Generated
+8
@@ -3268,13 +3268,21 @@ dependencies = [
|
||||
"codex-config",
|
||||
"codex-core",
|
||||
"codex-core-plugins",
|
||||
"codex-exec-server",
|
||||
"codex-extension-api",
|
||||
"codex-features",
|
||||
"codex-login",
|
||||
"codex-mcp",
|
||||
"codex-plugin",
|
||||
"codex-protocol",
|
||||
"codex-utils-absolute-path",
|
||||
"codex-utils-path-uri",
|
||||
"pretty_assertions",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -45,3 +45,4 @@ pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketpl
|
||||
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
|
||||
pub use provider::ExecutorPluginProvider;
|
||||
pub use provider::ExecutorPluginProviderError;
|
||||
pub use provider::ResolvedExecutorPlugin;
|
||||
|
||||
@@ -662,14 +662,7 @@ async fn load_plugin(
|
||||
restriction_product,
|
||||
skill_config_rules,
|
||||
} => {
|
||||
loaded_plugin.manifest_name = manifest
|
||||
.interface
|
||||
.as_ref()
|
||||
.and_then(|interface| interface.display_name.as_deref())
|
||||
.map(str::trim)
|
||||
.filter(|display_name| !display_name.is_empty())
|
||||
.map(str::to_string)
|
||||
.or_else(|| Some(manifest.name.clone()));
|
||||
loaded_plugin.manifest_name = Some(manifest.display_name().to_string());
|
||||
loaded_plugin.manifest_description = manifest.description.clone();
|
||||
loaded_plugin.skill_roots = plugin_skill_roots(&plugin_root, manifest_paths);
|
||||
let resolved_skills = load_plugin_skills(
|
||||
|
||||
@@ -77,6 +77,25 @@ pub struct ExecutorPluginProvider {
|
||||
environment_manager: Arc<EnvironmentManager>,
|
||||
}
|
||||
|
||||
/// A resolved plugin paired with the concrete filesystem used to read it.
|
||||
#[derive(Clone)]
|
||||
pub struct ResolvedExecutorPlugin {
|
||||
plugin: ResolvedPlugin,
|
||||
file_system: Arc<dyn ExecutorFileSystem>,
|
||||
}
|
||||
|
||||
impl ResolvedExecutorPlugin {
|
||||
/// Returns the source-neutral plugin descriptor.
|
||||
pub fn plugin(&self) -> &ResolvedPlugin {
|
||||
&self.plugin
|
||||
}
|
||||
|
||||
/// Returns the concrete filesystem that resolved the descriptor.
|
||||
pub fn file_system(&self) -> &dyn ExecutorFileSystem {
|
||||
self.file_system.as_ref()
|
||||
}
|
||||
}
|
||||
|
||||
impl ExecutorPluginProvider {
|
||||
/// Creates a provider backed by the active execution environments.
|
||||
pub fn new(environment_manager: Arc<EnvironmentManager>) -> Self {
|
||||
@@ -84,15 +103,12 @@ impl ExecutorPluginProvider {
|
||||
environment_manager,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PluginProvider for ExecutorPluginProvider {
|
||||
type Error = ExecutorPluginProviderError;
|
||||
|
||||
async fn resolve(
|
||||
/// Resolves a plugin and retains the exact filesystem used for package access.
|
||||
pub async fn resolve_bound(
|
||||
&self,
|
||||
selected_root: &SelectedCapabilityRoot,
|
||||
) -> Result<Option<ResolvedPlugin>, Self::Error> {
|
||||
) -> Result<Option<ResolvedExecutorPlugin>, ExecutorPluginProviderError> {
|
||||
let root_id = &selected_root.id;
|
||||
let plugin_root = selected_plugin_root(selected_root)?;
|
||||
let CapabilityRootLocation::Environment { environment_id, .. } = &selected_root.location;
|
||||
@@ -104,8 +120,25 @@ impl PluginProvider for ExecutorPluginProvider {
|
||||
environment_id: environment_id.clone(),
|
||||
})?;
|
||||
let file_system = environment.get_filesystem();
|
||||
let plugin = resolve_plugin_root(selected_root, plugin_root, file_system.as_ref()).await?;
|
||||
|
||||
resolve_plugin_root(selected_root, plugin_root, file_system.as_ref()).await
|
||||
Ok(plugin.map(|plugin| ResolvedExecutorPlugin {
|
||||
plugin,
|
||||
file_system,
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
impl PluginProvider for ExecutorPluginProvider {
|
||||
type Error = ExecutorPluginProviderError;
|
||||
|
||||
async fn resolve(
|
||||
&self,
|
||||
selected_root: &SelectedCapabilityRoot,
|
||||
) -> Result<Option<ResolvedPlugin>, Self::Error> {
|
||||
self.resolve_bound(selected_root)
|
||||
.await
|
||||
.map(|plugin| plugin.map(|plugin| plugin.plugin))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ version.workspace = true
|
||||
[lib]
|
||||
name = "codex_mcp_extension"
|
||||
path = "src/lib.rs"
|
||||
test = false
|
||||
doctest = false
|
||||
|
||||
[lints]
|
||||
@@ -15,13 +14,22 @@ workspace = true
|
||||
|
||||
[dependencies]
|
||||
codex-core = { workspace = true }
|
||||
codex-core-plugins = { workspace = true }
|
||||
codex-config = { workspace = true }
|
||||
codex-exec-server = { workspace = true }
|
||||
codex-extension-api = { workspace = true }
|
||||
codex-features = { workspace = true }
|
||||
codex-mcp = { workspace = true }
|
||||
codex-plugin = { workspace = true }
|
||||
codex-protocol = { workspace = true }
|
||||
codex-utils-absolute-path = { workspace = true }
|
||||
codex-utils-path-uri = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
tokio = { workspace = true, features = ["sync"] }
|
||||
|
||||
[dev-dependencies]
|
||||
codex-config = { workspace = true }
|
||||
codex-core-plugins = { workspace = true }
|
||||
codex-login = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
tempfile = { workspace = true }
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
use codex_core::config::Config;
|
||||
use codex_core_plugins::ExecutorPluginProvider;
|
||||
use codex_exec_server::EnvironmentManager;
|
||||
use codex_extension_api::ExtensionDataInit;
|
||||
use codex_extension_api::ExtensionFuture;
|
||||
use codex_extension_api::McpServerContribution;
|
||||
use codex_extension_api::McpServerContributionContext;
|
||||
use codex_extension_api::McpServerContributor;
|
||||
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::OnceCell;
|
||||
|
||||
use self::provider::ExecutorPluginMcpProvider;
|
||||
|
||||
mod provider;
|
||||
|
||||
/// Frozen MCP declarations for one selected package.
|
||||
///
|
||||
/// Each server config retains the stable logical environment ID. Reconnection may replace the
|
||||
/// concrete environment instance without changing that authority.
|
||||
#[derive(Clone)]
|
||||
struct SelectedPluginMcpServers {
|
||||
plugin_id: String,
|
||||
plugin_display_name: String,
|
||||
selection_order: usize,
|
||||
servers: Vec<(String, codex_config::McpServerConfig)>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(crate) struct SelectedExecutorPluginMcpState {
|
||||
snapshot: OnceCell<Vec<SelectedPluginMcpServers>>,
|
||||
}
|
||||
|
||||
pub(crate) fn seed_thread_state(thread_init: &mut ExtensionDataInit) {
|
||||
thread_init.insert(SelectedExecutorPluginMcpState::default());
|
||||
}
|
||||
|
||||
pub(crate) struct SelectedExecutorPluginMcpContributor {
|
||||
plugin_provider: ExecutorPluginProvider,
|
||||
mcp_provider: ExecutorPluginMcpProvider,
|
||||
}
|
||||
|
||||
impl SelectedExecutorPluginMcpContributor {
|
||||
pub(crate) fn new(environment_manager: Arc<EnvironmentManager>) -> Self {
|
||||
Self {
|
||||
plugin_provider: ExecutorPluginProvider::new(Arc::clone(&environment_manager)),
|
||||
mcp_provider: ExecutorPluginMcpProvider,
|
||||
}
|
||||
}
|
||||
|
||||
async fn resolve_snapshot(
|
||||
&self,
|
||||
selected_roots: &[SelectedCapabilityRoot],
|
||||
) -> Vec<SelectedPluginMcpServers> {
|
||||
let mut snapshot = Vec::new();
|
||||
|
||||
for (selection_order, selected_root) in selected_roots.iter().enumerate() {
|
||||
let plugin = match self.plugin_provider.resolve_bound(selected_root).await {
|
||||
Ok(Some(plugin)) => plugin,
|
||||
Ok(None) => continue,
|
||||
Err(err) => {
|
||||
tracing::warn!(
|
||||
selected_root = selected_root.id,
|
||||
error = %err,
|
||||
"failed to resolve selected executor plugin for MCP discovery"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match self.mcp_provider.load(&plugin).await {
|
||||
Ok(servers) => snapshot.push(SelectedPluginMcpServers {
|
||||
plugin_id: plugin.plugin().selected_root_id().to_string(),
|
||||
plugin_display_name: plugin.plugin().manifest().display_name().to_string(),
|
||||
selection_order,
|
||||
servers,
|
||||
}),
|
||||
Err(err) => {
|
||||
tracing::warn!(
|
||||
selected_root = selected_root.id,
|
||||
error = %err,
|
||||
"failed to load selected executor plugin MCP servers"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
snapshot
|
||||
}
|
||||
}
|
||||
|
||||
impl McpServerContributor<Config> for SelectedExecutorPluginMcpContributor {
|
||||
fn id(&self) -> &'static str {
|
||||
"selected_executor_plugin_mcp"
|
||||
}
|
||||
|
||||
fn contribute<'a>(
|
||||
&'a self,
|
||||
context: McpServerContributionContext<'a, Config>,
|
||||
) -> ExtensionFuture<'a, Vec<McpServerContribution>> {
|
||||
Box::pin(async move {
|
||||
let Some(thread_init) = context.thread_init() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Some(selected_roots) = thread_init.get::<Vec<SelectedCapabilityRoot>>() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Some(state) = thread_init.get::<SelectedExecutorPluginMcpState>() else {
|
||||
tracing::warn!("selected executor plugin MCP state was not initialized");
|
||||
return Vec::new();
|
||||
};
|
||||
let snapshot = state
|
||||
.snapshot
|
||||
.get_or_init(|| self.resolve_snapshot(selected_roots.as_ref()))
|
||||
.await;
|
||||
let mut contributions = Vec::new();
|
||||
|
||||
for plugin in snapshot {
|
||||
let mut servers = plugin.servers.iter().cloned().collect::<HashMap<_, _>>();
|
||||
context
|
||||
.config()
|
||||
.apply_plugin_mcp_server_requirements(&plugin.plugin_id, &mut servers);
|
||||
let mut servers = servers.into_iter().collect::<Vec<_>>();
|
||||
servers.sort_unstable_by(|left, right| left.0.cmp(&right.0));
|
||||
contributions.extend(servers.into_iter().map(|(name, config)| {
|
||||
McpServerContribution::SelectedPlugin {
|
||||
name,
|
||||
plugin_id: plugin.plugin_id.clone(),
|
||||
plugin_display_name: plugin.plugin_display_name.clone(),
|
||||
selection_order: plugin.selection_order,
|
||||
config: Box::new(config),
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
contributions
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
use codex_config::McpServerConfig;
|
||||
use codex_config::McpServerTransportConfig;
|
||||
use codex_core_plugins::ResolvedExecutorPlugin;
|
||||
use codex_exec_server::ExecutorFileSystem;
|
||||
use codex_mcp::PluginMcpServerPlacement;
|
||||
use codex_mcp::parse_plugin_mcp_config;
|
||||
use codex_plugin::PluginResourceLocator;
|
||||
use codex_plugin::ResolvedPlugin;
|
||||
use codex_plugin::ResolvedPluginLocation;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use codex_utils_path_uri::PathUri;
|
||||
use std::io;
|
||||
use thiserror::Error;
|
||||
|
||||
const DEFAULT_MCP_CONFIG_FILE: &str = ".mcp.json";
|
||||
|
||||
/// Loads MCP declarations from resolved plugins through their owning executor.
|
||||
#[derive(Clone, Copy, Debug, Default)]
|
||||
pub(super) struct ExecutorPluginMcpProvider;
|
||||
|
||||
/// Failure to load an executor plugin's MCP declarations.
|
||||
#[derive(Debug, Error)]
|
||||
pub(super) enum ExecutorPluginMcpProviderError {
|
||||
#[error("failed to read MCP config for selected plugin `{plugin_id}` at `{path}`: {source}")]
|
||||
ReadConfig {
|
||||
plugin_id: String,
|
||||
path: AbsolutePathBuf,
|
||||
#[source]
|
||||
source: io::Error,
|
||||
},
|
||||
#[error("failed to parse MCP config for selected plugin `{plugin_id}` at `{path}`: {source}")]
|
||||
ParseConfig {
|
||||
plugin_id: String,
|
||||
path: AbsolutePathBuf,
|
||||
#[source]
|
||||
source: serde_json::Error,
|
||||
},
|
||||
}
|
||||
|
||||
impl ExecutorPluginMcpProvider {
|
||||
/// Returns stdio servers declared by `plugin`, bound to its environment.
|
||||
pub(super) async fn load(
|
||||
&self,
|
||||
plugin: &ResolvedExecutorPlugin,
|
||||
) -> Result<Vec<(String, McpServerConfig)>, ExecutorPluginMcpProviderError> {
|
||||
let ResolvedPluginLocation::Environment { root, .. } = plugin.plugin().location();
|
||||
|
||||
load_from_file_system(plugin.plugin(), root, plugin.file_system()).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_from_file_system(
|
||||
plugin: &ResolvedPlugin,
|
||||
plugin_root: &AbsolutePathBuf,
|
||||
file_system: &dyn ExecutorFileSystem,
|
||||
) -> Result<Vec<(String, McpServerConfig)>, ExecutorPluginMcpProviderError> {
|
||||
let ResolvedPluginLocation::Environment { environment_id, .. } = plugin.location();
|
||||
let plugin_id = plugin.selected_root_id();
|
||||
let (config_path, is_default) = match plugin.manifest().paths.mcp_servers.as_ref() {
|
||||
Some(PluginResourceLocator::Environment { path, .. }) => (path.clone(), false),
|
||||
None => (plugin_root.join(DEFAULT_MCP_CONFIG_FILE), true),
|
||||
};
|
||||
let config_uri = PathUri::from_abs_path(&config_path);
|
||||
|
||||
let contents = match file_system
|
||||
.read_file_text(&config_uri, /*sandbox*/ None)
|
||||
.await
|
||||
{
|
||||
Ok(contents) => contents,
|
||||
Err(source) if is_default && source.kind() == io::ErrorKind::NotFound => {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
Err(source) => {
|
||||
return Err(ExecutorPluginMcpProviderError::ReadConfig {
|
||||
plugin_id: plugin_id.to_string(),
|
||||
path: config_path.clone(),
|
||||
source,
|
||||
});
|
||||
}
|
||||
};
|
||||
let parsed = parse_plugin_mcp_config(
|
||||
plugin_root.as_path(),
|
||||
&contents,
|
||||
PluginMcpServerPlacement::Environment { environment_id },
|
||||
)
|
||||
.map_err(|source| ExecutorPluginMcpProviderError::ParseConfig {
|
||||
plugin_id: plugin_id.to_string(),
|
||||
path: config_path,
|
||||
source,
|
||||
})?;
|
||||
|
||||
for error in parsed.errors {
|
||||
tracing::warn!(
|
||||
plugin = plugin_id,
|
||||
server = error.name,
|
||||
error = error.message,
|
||||
"ignoring invalid executor plugin MCP server"
|
||||
);
|
||||
}
|
||||
|
||||
Ok(parsed
|
||||
.servers
|
||||
.into_iter()
|
||||
.filter_map(|(name, config)| match &config.transport {
|
||||
McpServerTransportConfig::Stdio { .. } => Some((name, config)),
|
||||
McpServerTransportConfig::StreamableHttp { .. } => {
|
||||
tracing::warn!(
|
||||
plugin = plugin_id,
|
||||
server = name,
|
||||
"ignoring HTTP MCP server from executor plugin"
|
||||
);
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "provider_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,266 @@
|
||||
use super::DEFAULT_MCP_CONFIG_FILE;
|
||||
use super::ExecutorPluginMcpProviderError;
|
||||
use super::load_from_file_system;
|
||||
use codex_config::McpServerConfig;
|
||||
use codex_config::McpServerTransportConfig;
|
||||
use codex_exec_server::CopyOptions;
|
||||
use codex_exec_server::CreateDirectoryOptions;
|
||||
use codex_exec_server::ExecutorFileSystem;
|
||||
use codex_exec_server::ExecutorFileSystemFuture;
|
||||
use codex_exec_server::FileMetadata;
|
||||
use codex_exec_server::FileSystemResult;
|
||||
use codex_exec_server::FileSystemSandboxContext;
|
||||
use codex_exec_server::ReadDirectoryEntry;
|
||||
use codex_exec_server::RemoveOptions;
|
||||
use codex_plugin::ResolvedPlugin;
|
||||
use codex_plugin::manifest::PluginManifest;
|
||||
use codex_plugin::manifest::PluginManifestPaths;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use codex_utils_path_uri::PathUri;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::collections::HashMap;
|
||||
use std::io;
|
||||
use std::sync::Mutex;
|
||||
|
||||
const MCP_CONFIG_CONTENTS: &str = r#"{
|
||||
"mcpServers": {
|
||||
"demo": {"command": "demo-mcp", "environment_id": "local"},
|
||||
"hosted": {"url": "https://example.com/mcp"}
|
||||
}
|
||||
}"#;
|
||||
|
||||
struct SyntheticExecutorFileSystem {
|
||||
config_path: AbsolutePathBuf,
|
||||
config_contents: Option<&'static str>,
|
||||
reads: Mutex<Vec<AbsolutePathBuf>>,
|
||||
}
|
||||
|
||||
impl SyntheticExecutorFileSystem {
|
||||
fn unsupported<T>() -> FileSystemResult<T> {
|
||||
Err(io::Error::new(
|
||||
io::ErrorKind::Unsupported,
|
||||
"operation is not used by executor MCP provider tests",
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
impl ExecutorFileSystem for SyntheticExecutorFileSystem {
|
||||
fn canonicalize<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, PathUri> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn read_file<'a>(
|
||||
&'a self,
|
||||
path: &'a PathUri,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, Vec<u8>> {
|
||||
Box::pin(async move {
|
||||
let path = path.to_abs_path()?;
|
||||
self.reads
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.push(path.clone());
|
||||
if path != self.config_path {
|
||||
return Err(io::Error::new(io::ErrorKind::NotFound, "not found"));
|
||||
}
|
||||
self.config_contents
|
||||
.map(|contents| contents.as_bytes().to_vec())
|
||||
.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "not found"))
|
||||
})
|
||||
}
|
||||
|
||||
fn write_file<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_contents: Vec<u8>,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, ()> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn create_directory<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_options: CreateDirectoryOptions,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, ()> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn get_metadata<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, FileMetadata> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn read_directory<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, Vec<ReadDirectoryEntry>> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn remove<'a>(
|
||||
&'a self,
|
||||
_path: &'a PathUri,
|
||||
_options: RemoveOptions,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, ()> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
|
||||
fn copy<'a>(
|
||||
&'a self,
|
||||
_source_path: &'a PathUri,
|
||||
_destination_path: &'a PathUri,
|
||||
_options: CopyOptions,
|
||||
_sandbox: Option<&'a FileSystemSandboxContext>,
|
||||
) -> ExecutorFileSystemFuture<'a, ()> {
|
||||
Box::pin(async { Self::unsupported() })
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reads_declared_config_only_through_executor_file_system() {
|
||||
let temp_dir = tempfile::tempdir().expect("tempdir");
|
||||
let plugin_root =
|
||||
AbsolutePathBuf::from_absolute_path_checked(temp_dir.path().join("executor-only-plugin"))
|
||||
.expect("absolute plugin root");
|
||||
assert!(!plugin_root.as_path().exists());
|
||||
let config_path = plugin_root.join("config/mcp.json");
|
||||
let plugin = resolved_plugin(&plugin_root, Some(config_path.clone()));
|
||||
let file_system = SyntheticExecutorFileSystem {
|
||||
config_path: config_path.clone(),
|
||||
config_contents: Some(MCP_CONFIG_CONTENTS),
|
||||
reads: Mutex::new(Vec::new()),
|
||||
};
|
||||
|
||||
let servers = load_from_file_system(&plugin, &plugin_root, &file_system)
|
||||
.await
|
||||
.expect("load executor MCP config");
|
||||
|
||||
assert_eq!(
|
||||
servers,
|
||||
vec![(
|
||||
"demo".to_string(),
|
||||
McpServerConfig {
|
||||
transport: McpServerTransportConfig::Stdio {
|
||||
command: "demo-mcp".to_string(),
|
||||
args: Vec::new(),
|
||||
env: None,
|
||||
env_vars: Vec::new(),
|
||||
cwd: Some(plugin_root.to_path_buf()),
|
||||
},
|
||||
environment_id: "executor-test".to_string(),
|
||||
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: None,
|
||||
oauth_resource: None,
|
||||
tools: HashMap::new(),
|
||||
},
|
||||
)]
|
||||
);
|
||||
assert_eq!(reads(&file_system), vec![config_path]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_default_config_is_empty() {
|
||||
let temp_dir = tempfile::tempdir().expect("tempdir");
|
||||
let plugin_root = AbsolutePathBuf::from_absolute_path_checked(temp_dir.path().join("plugin"))
|
||||
.expect("absolute plugin root");
|
||||
let config_path = plugin_root.join(DEFAULT_MCP_CONFIG_FILE);
|
||||
let plugin = resolved_plugin(&plugin_root, /*mcp_servers*/ None);
|
||||
let file_system = SyntheticExecutorFileSystem {
|
||||
config_path: config_path.clone(),
|
||||
config_contents: None,
|
||||
reads: Mutex::new(Vec::new()),
|
||||
};
|
||||
|
||||
let servers = load_from_file_system(&plugin, &plugin_root, &file_system)
|
||||
.await
|
||||
.expect("missing default config should be ignored");
|
||||
|
||||
assert_eq!(servers, Vec::new());
|
||||
assert_eq!(reads(&file_system), vec![config_path]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn malformed_declared_config_is_an_error() {
|
||||
let temp_dir = tempfile::tempdir().expect("tempdir");
|
||||
let plugin_root = AbsolutePathBuf::from_absolute_path_checked(temp_dir.path().join("plugin"))
|
||||
.expect("absolute plugin root");
|
||||
let config_path = plugin_root.join("mcp.json");
|
||||
let plugin = resolved_plugin(&plugin_root, Some(config_path.clone()));
|
||||
let file_system = SyntheticExecutorFileSystem {
|
||||
config_path: config_path.clone(),
|
||||
config_contents: Some("{not-json"),
|
||||
reads: Mutex::new(Vec::new()),
|
||||
};
|
||||
|
||||
let err = load_from_file_system(&plugin, &plugin_root, &file_system)
|
||||
.await
|
||||
.expect_err("malformed declared config should fail");
|
||||
|
||||
let ExecutorPluginMcpProviderError::ParseConfig {
|
||||
plugin_id,
|
||||
path,
|
||||
source: _,
|
||||
} = err
|
||||
else {
|
||||
panic!("expected parse error");
|
||||
};
|
||||
assert_eq!(
|
||||
(plugin_id, path),
|
||||
("selected-root".to_string(), config_path.clone())
|
||||
);
|
||||
assert_eq!(reads(&file_system), vec![config_path]);
|
||||
}
|
||||
|
||||
fn resolved_plugin(
|
||||
plugin_root: &AbsolutePathBuf,
|
||||
mcp_servers: Option<AbsolutePathBuf>,
|
||||
) -> ResolvedPlugin {
|
||||
ResolvedPlugin::from_environment(
|
||||
"selected-root".to_string(),
|
||||
"executor-test".to_string(),
|
||||
plugin_root.clone(),
|
||||
plugin_root.join(".codex-plugin/plugin.json"),
|
||||
PluginManifest {
|
||||
name: "demo-plugin".to_string(),
|
||||
version: None,
|
||||
description: None,
|
||||
keywords: Vec::new(),
|
||||
paths: PluginManifestPaths {
|
||||
skills: None,
|
||||
mcp_servers,
|
||||
apps: None,
|
||||
hooks: None,
|
||||
},
|
||||
interface: None,
|
||||
},
|
||||
)
|
||||
.expect("valid plugin descriptor")
|
||||
}
|
||||
|
||||
fn reads(file_system: &SyntheticExecutorFileSystem) -> Vec<AbsolutePathBuf> {
|
||||
file_system
|
||||
.reads
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.clone()
|
||||
}
|
||||
@@ -7,6 +7,8 @@ use codex_extension_api::McpServerContributor;
|
||||
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
|
||||
use codex_mcp::hosted_plugin_runtime_mcp_server_config;
|
||||
|
||||
mod executor_plugin;
|
||||
|
||||
struct HostedPluginRuntimeExtension;
|
||||
|
||||
impl McpServerContributor<Config> for HostedPluginRuntimeExtension {
|
||||
@@ -39,3 +41,20 @@ impl McpServerContributor<Config> for HostedPluginRuntimeExtension {
|
||||
pub fn install(builder: &mut ExtensionRegistryBuilder<Config>) {
|
||||
builder.mcp_server_contributor(std::sync::Arc::new(HostedPluginRuntimeExtension));
|
||||
}
|
||||
|
||||
/// Installs discovery for MCP servers declared by thread-selected executor plugins.
|
||||
pub fn install_executor_plugins(
|
||||
builder: &mut ExtensionRegistryBuilder<Config>,
|
||||
environment_manager: std::sync::Arc<codex_exec_server::EnvironmentManager>,
|
||||
) {
|
||||
builder.mcp_server_contributor(std::sync::Arc::new(
|
||||
executor_plugin::SelectedExecutorPluginMcpContributor::new(environment_manager),
|
||||
));
|
||||
}
|
||||
|
||||
/// Seeds the per-thread snapshot used by selected executor plugin MCP discovery.
|
||||
pub fn initialize_executor_plugin_thread_data(
|
||||
thread_init: &mut codex_extension_api::ExtensionDataInit,
|
||||
) {
|
||||
executor_plugin::seed_thread_state(thread_init);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
use codex_config::test_support::CloudConfigBundleFixture;
|
||||
use codex_core::config::Config;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_exec_server::EnvironmentManager;
|
||||
use codex_exec_server::LOCAL_ENVIRONMENT_ID;
|
||||
use codex_extension_api::ExtensionDataInit;
|
||||
use codex_extension_api::ExtensionRegistryBuilder;
|
||||
use codex_extension_api::McpServerContribution;
|
||||
use codex_extension_api::McpServerContributionContext;
|
||||
use codex_protocol::capabilities::CapabilityRootLocation;
|
||||
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::sync::Arc;
|
||||
|
||||
type TestResult = Result<(), Box<dyn std::error::Error>>;
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
struct ContributionSummary {
|
||||
name: String,
|
||||
plugin_id: String,
|
||||
plugin_display_name: String,
|
||||
selection_order: usize,
|
||||
enabled: bool,
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn selected_plugin_servers_use_managed_requirements_for_the_selected_root_id() -> TestResult {
|
||||
let codex_home = tempfile::tempdir()?;
|
||||
let plugin_root = tempfile::tempdir()?;
|
||||
std::fs::create_dir_all(plugin_root.path().join(".codex-plugin"))?;
|
||||
std::fs::write(
|
||||
plugin_root.path().join(".codex-plugin/plugin.json"),
|
||||
r#"{"name":"different-manifest-name","interface":{"displayName":"Selected Demo"}}"#,
|
||||
)?;
|
||||
std::fs::write(
|
||||
plugin_root.path().join(".mcp.json"),
|
||||
r#"{
|
||||
"mcpServers": {
|
||||
"allowed": {"command":"allowed-command"},
|
||||
"mismatched": {"command":"wrong-command"},
|
||||
"unlisted": {"command":"unlisted-command"}
|
||||
}
|
||||
}"#,
|
||||
)?;
|
||||
let config = ConfigBuilder::default()
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
||||
.cloud_config_bundle(
|
||||
CloudConfigBundleFixture::loader_with_enterprise_requirement(
|
||||
r#"
|
||||
[plugins."selected-root".mcp_servers.allowed.identity]
|
||||
command = "allowed-command"
|
||||
|
||||
[plugins."selected-root".mcp_servers.mismatched.identity]
|
||||
command = "expected-command"
|
||||
"#,
|
||||
),
|
||||
)
|
||||
.build()
|
||||
.await?;
|
||||
|
||||
let contributions = selected_plugin_contributions(&config, plugin_root.path()).await;
|
||||
|
||||
assert_eq!(
|
||||
contributions,
|
||||
vec![
|
||||
ContributionSummary {
|
||||
name: "allowed".to_string(),
|
||||
plugin_id: "selected-root".to_string(),
|
||||
plugin_display_name: "Selected Demo".to_string(),
|
||||
selection_order: 0,
|
||||
enabled: true,
|
||||
},
|
||||
ContributionSummary {
|
||||
name: "mismatched".to_string(),
|
||||
plugin_id: "selected-root".to_string(),
|
||||
plugin_display_name: "Selected Demo".to_string(),
|
||||
selection_order: 0,
|
||||
enabled: false,
|
||||
},
|
||||
ContributionSummary {
|
||||
name: "unlisted".to_string(),
|
||||
plugin_id: "selected-root".to_string(),
|
||||
plugin_display_name: "Selected Demo".to_string(),
|
||||
selection_order: 0,
|
||||
enabled: false,
|
||||
},
|
||||
]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn selected_plugin_contributions(
|
||||
config: &Config,
|
||||
plugin_root: &std::path::Path,
|
||||
) -> Vec<ContributionSummary> {
|
||||
let mut builder = ExtensionRegistryBuilder::new();
|
||||
codex_mcp_extension::install_executor_plugins(
|
||||
&mut builder,
|
||||
Arc::new(EnvironmentManager::default_for_tests()),
|
||||
);
|
||||
let registry = builder.build();
|
||||
let mut thread_init = ExtensionDataInit::new();
|
||||
thread_init.insert(vec![SelectedCapabilityRoot {
|
||||
id: "selected-root".to_string(),
|
||||
location: CapabilityRootLocation::Environment {
|
||||
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
|
||||
path: plugin_root.to_string_lossy().into_owned(),
|
||||
},
|
||||
}]);
|
||||
codex_mcp_extension::initialize_executor_plugin_thread_data(&mut thread_init);
|
||||
|
||||
registry.mcp_server_contributors()[0]
|
||||
.contribute(McpServerContributionContext::for_thread(
|
||||
config,
|
||||
&thread_init,
|
||||
))
|
||||
.await
|
||||
.into_iter()
|
||||
.map(|contribution| match contribution {
|
||||
McpServerContribution::SelectedPlugin {
|
||||
name,
|
||||
plugin_id,
|
||||
plugin_display_name,
|
||||
selection_order,
|
||||
config,
|
||||
} => ContributionSummary {
|
||||
name,
|
||||
plugin_id,
|
||||
plugin_display_name,
|
||||
selection_order,
|
||||
enabled: config.enabled,
|
||||
},
|
||||
McpServerContribution::Set { .. } | McpServerContribution::Remove { .. } => {
|
||||
panic!("expected selected plugin contribution")
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -71,6 +71,16 @@ impl<Resource> Default for PluginManifestInterface<Resource> {
|
||||
}
|
||||
|
||||
impl<Resource> PluginManifest<Resource> {
|
||||
/// Returns the model- and UI-facing package name, falling back to the manifest name.
|
||||
pub fn display_name(&self) -> &str {
|
||||
self.interface
|
||||
.as_ref()
|
||||
.and_then(|interface| interface.display_name.as_deref())
|
||||
.map(str::trim)
|
||||
.filter(|display_name| !display_name.is_empty())
|
||||
.unwrap_or(&self.name)
|
||||
}
|
||||
|
||||
pub(crate) fn try_map_resources<Mapped, Error>(
|
||||
self,
|
||||
mut map: impl FnMut(Resource) -> Result<Mapped, Error>,
|
||||
|
||||
Reference in New Issue
Block a user