diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 33f0811a10..520004d7c3 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3381,6 +3381,7 @@ dependencies = [ "codex-api", "codex-config", "codex-context-fragments", + "codex-core-plugin-common", "codex-exec-server-protocol", "codex-file-system", "codex-guardian-context", diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 4d1da4f1ec..ac348c7439 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -78,9 +78,11 @@ pub use mcp::hosted_plugin_runtime_mcp_server_config; pub use mcp::tool_plugin_context; pub use plugin_config::PluginMcpConfigParseOutcome; pub use plugin_config::PluginMcpServerParseError; +pub use plugin_config::normalize_executor_plugin_mcp_declarations; pub use plugin_config::parse_agent_plugin_mcp_config; pub use plugin_config::parse_executor_plugin_mcp_config; pub use plugin_config::parse_plugin_mcp_config; +pub use plugin_config::parse_plugin_mcp_declarations; pub use mcp::McpServerStatusSnapshot; pub use mcp::McpSnapshotDetail; diff --git a/codex-rs/codex-mcp/src/plugin_config.rs b/codex-rs/codex-mcp/src/plugin_config.rs index 9a14022a8d..15ab0af28d 100644 --- a/codex-rs/codex-mcp/src/plugin_config.rs +++ b/codex-rs/codex-mcp/src/plugin_config.rs @@ -63,6 +63,13 @@ impl PluginMcpFile { } } +/// Parses wrapped or plain plugin MCP declarations without runtime normalization. +pub fn parse_plugin_mcp_declarations( + contents: &str, +) -> Result, serde_json::Error> { + serde_json::from_str::(contents).map(PluginMcpFile::into_mcp_servers) +} + /// Parses the two supported plugin MCP file shapes and normalizes each server. /// /// Native plugin HTTP servers share the regular MCP transport configuration; @@ -93,6 +100,21 @@ pub fn parse_executor_plugin_mcp_config( ) } +/// Normalizes already-parsed declarations for one executor installation. +pub fn normalize_executor_plugin_mcp_declarations( + plugin_root: &PathUri, + declarations: BTreeMap, + environment_id: &str, +) -> PluginMcpConfigParseOutcome { + normalize_plugin_mcp_declarations( + declarations, + PluginMcpSource::Environment { + root: plugin_root, + environment_id, + }, + ) +} + impl PluginMcpSource<'_> { fn display(self) -> String { match self { @@ -106,10 +128,17 @@ fn parse_plugin_mcp_config_from( contents: &str, source: PluginMcpSource<'_>, ) -> Result { - let parsed = serde_json::from_str::(contents)?; + let declarations = parse_plugin_mcp_declarations(contents)?; + Ok(normalize_plugin_mcp_declarations(declarations, source)) +} + +fn normalize_plugin_mcp_declarations( + declarations: BTreeMap, + source: PluginMcpSource<'_>, +) -> PluginMcpConfigParseOutcome { let mut outcome = PluginMcpConfigParseOutcome::default(); - for (name, config_value) in parsed.into_mcp_servers() { + for (name, config_value) in declarations { match normalize_plugin_mcp_server(config_value, source) { Ok(config) => { outcome.servers.insert(name, config); @@ -120,7 +149,7 @@ fn parse_plugin_mcp_config_from( } } - Ok(outcome) + outcome } fn normalize_plugin_mcp_server( diff --git a/codex-rs/codex-mcp/src/plugin_config_tests.rs b/codex-rs/codex-mcp/src/plugin_config_tests.rs index 3c57226709..23accf5c1a 100644 --- a/codex-rs/codex-mcp/src/plugin_config_tests.rs +++ b/codex-rs/codex-mcp/src/plugin_config_tests.rs @@ -1,5 +1,6 @@ use super::PluginMcpConfigParseOutcome; use super::PluginMcpServerParseError; +use super::normalize_executor_plugin_mcp_declarations; use super::parse_agent_plugin_mcp_config; use super::parse_executor_plugin_mcp_config; use super::parse_plugin_mcp_config; @@ -751,6 +752,18 @@ fn executor_environment_placement_resolves_foreign_uri_cwd() { ) .expect("parse plugin MCP config"); + assert_eq!( + normalize_executor_plugin_mcp_declarations( + &plugin_root, + BTreeMap::from([( + "demo".to_string(), + serde_json::json!({"command": "demo-mcp", "cwd": "scripts"}), + )]), + "executor-1", + ), + outcome + ); + assert_eq!( outcome, PluginMcpConfigParseOutcome { diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 5d512a9e2d..28d02bd59b 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -12,6 +12,10 @@ #[path = "rmcp_client/status.rs"] mod status; +#[cfg(test)] +#[path = "rmcp_client/provenance_tests.rs"] +mod provenance_tests; + use std::borrow::Cow; use std::collections::BTreeMap; use std::collections::HashMap; @@ -97,6 +101,8 @@ use tracing::warn; /// not use it. Its `cacheable: false` property disables sharing tool definitions across connections. const MCP_TOOL_CATALOG_CACHE_CAPABILITY: &str = "codex/tool-catalog-cache"; const MCP_TOOL_CATALOG_CACHEABLE_PROPERTY: &str = "cacheable"; +// Bound only the model-visible attribution note, not plugin membership or tool availability. +const MAX_PLUGIN_PROVENANCE_BYTES: usize = 1_024; pub(crate) const MCP_TOOLS_LIST_DURATION_METRIC: &str = "codex.mcp.tools.list.duration_ms"; pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str = "codex.mcp.tools.fetch_uncached.duration_ms"; @@ -766,18 +772,39 @@ fn add_plugin_provenance_to_tool(tool: &mut ToolInfo, plugin_names: &[String]) { return; } - let plugin_source_note = if plugin_names.len() == 1 { - format!("This tool is part of plugin `{}`.", plugin_names[0]) + let plural = if plugin_names.len() == 1 { "" } else { "s" }; + let mut plugin_source_note = format!("This tool is part of plugin{plural} "); + let mut shown = 0; + for (index, name) in plugin_names.iter().enumerate() { + let separator = if index == 0 { "" } else { ", " }; + let remaining = plugin_names.len() - index - 1; + let ending = if remaining == 0 { + ".".to_string() + } else { + format!(", and {remaining} more.") + }; + // Reserve the suffix and keep whole names, including multibyte UTF-8 characters. + if plugin_source_note.len() + separator.len() + name.len() + 2 + ending.len() + > MAX_PLUGIN_PROVENANCE_BYTES + { + break; + } + plugin_source_note.push_str(separator); + plugin_source_note.push('`'); + plugin_source_note.push_str(name); + plugin_source_note.push('`'); + shown += 1; + } + if shown == 0 { + plugin_source_note = format!( + "This tool is part of {} plugin{plural}.", + plugin_names.len() + ); + } else if shown < plugin_names.len() { + plugin_source_note.push_str(&format!(", and {} more.", plugin_names.len() - shown)); } else { - format!( - "This tool is part of plugins {}.", - plugin_names - .iter() - .map(|plugin_name| format!("`{plugin_name}`")) - .collect::>() - .join(", ") - ) - }; + plugin_source_note.push('.'); + } let description = tool .tool .description diff --git a/codex-rs/codex-mcp/src/rmcp_client/provenance_tests.rs b/codex-rs/codex-mcp/src/rmcp_client/provenance_tests.rs new file mode 100644 index 0000000000..78f9ebe1e7 --- /dev/null +++ b/codex-rs/codex-mcp/src/rmcp_client/provenance_tests.rs @@ -0,0 +1,84 @@ +//! Plugin attribution stays bounded without changing tool definitions or membership. + +use super::*; +use pretty_assertions::assert_eq; + +fn app_tool(description: &str) -> ToolInfo { + tool_info_from_listed_tool( + CODEX_APPS_MCP_SERVER_NAME, + /*is_codex_apps_mcp_server*/ true, + /*server_instructions*/ None, + ToolWithConnectorId { + tool: RmcpTool::new("search", description.to_string(), Arc::default()), + connector_id: Some("shared-connector".to_string()), + connector_name: None, + connector_description: None, + }, + ) +} + +#[test] +fn provenance_preserves_short_descriptions_and_membership() { + for (description, names, expected_description) in [ + ("Search", vec![], "Search"), + ("", vec!["Alpha"], "This tool is part of plugin `Alpha`."), + ( + "Search", + vec!["Alpha"], + "Search. This tool is part of plugin `Alpha`.", + ), + ( + "Search!", + vec!["Alpha", "Beta"], + "Search! This tool is part of plugins `Alpha`, `Beta`.", + ), + ] { + let names = names.into_iter().map(str::to_string).collect::>(); + let mut tool = app_tool(description); + let mut expected = tool.clone(); + expected.tool.description = Some(expected_description.to_string().into()); + expected.plugin_display_names = names.clone(); + add_plugin_provenance_to_tool(&mut tool, &names); + assert_eq!(tool, expected); + } +} + +#[test] +fn provenance_bounds_aggregate_names_and_retains_all_associations() { + let exact_name = + "a".repeat(MAX_PLUGIN_PROVENANCE_BYTES - "This tool is part of plugin ``.".len()); + for names in [ + vec![exact_name.clone()], + vec![format!("{exact_name}a")], + vec!["界".repeat(MAX_PLUGIN_PROVENANCE_BYTES)], + (0..512) + .map(|index| format!("{index:03}{}", "界".repeat(61))) + .collect(), + ] { + let mut tool = app_tool("Search."); + let original = tool.clone(); + add_plugin_provenance_to_tool(&mut tool, &names); + let note = tool + .tool + .description + .as_deref() + .unwrap() + .strip_prefix("Search. ") + .unwrap(); + assert!(note.len() <= MAX_PLUGIN_PROVENANCE_BYTES); + if names[0] == exact_name { + assert_eq!(note.len(), MAX_PLUGIN_PROVENANCE_BYTES); + } else if names.len() == 1 { + assert_eq!(note, "This tool is part of 1 plugin."); + } else { + let shown = note.matches('`').count() / 2; + assert!(shown > 0 && shown < names.len()); + assert!(note.starts_with(&format!("This tool is part of plugins `{}`", names[0]))); + assert!(note.ends_with(&format!(", and {} more.", names.len() - shown))); + } + let mut expected = original; + expected.tool.description = tool.tool.description.clone(); + expected.plugin_display_names = names; + assert_eq!(tool, expected); + } +} diff --git a/codex-rs/core-plugin-common/src/identity.rs b/codex-rs/core-plugin-common/src/identity.rs new file mode 100644 index 0000000000..b983b07c1e --- /dev/null +++ b/codex-rs/core-plugin-common/src/identity.rs @@ -0,0 +1,20 @@ +//! Source-neutral identity shared by plugin catalogs and runtime attribution. + +/// Stable key used to merge discoveries of the same logical plugin. +/// Identity is independent of whether the plugin is supplied by the cloud or an executor. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum PluginIdentity { + /// Stable ID assigned by plugin-service and shared by its materialized installations. + Remote { remote_plugin_id: String }, + /// Local config key (normally `name@marketplace`), or a legacy selected-root identity. + Local { plugin_id: String }, +} + +impl PluginIdentity { + pub fn as_str(&self) -> &str { + match self { + Self::Remote { remote_plugin_id } => remote_plugin_id, + Self::Local { plugin_id } => plugin_id, + } + } +} diff --git a/codex-rs/core-plugin-common/src/lib.rs b/codex-rs/core-plugin-common/src/lib.rs index 4f5c0bdea9..f411e56758 100644 --- a/codex-rs/core-plugin-common/src/lib.rs +++ b/codex-rs/core-plugin-common/src/lib.rs @@ -1,4 +1,7 @@ //! Shared plugin identities and read-only installed-package rules. +mod identity; pub mod installed; pub mod plugin_id; + +pub use identity::PluginIdentity; diff --git a/codex-rs/core-plugins/src/catalog.rs b/codex-rs/core-plugins/src/catalog.rs index 7e53d955fc..a5c844e28f 100644 --- a/codex-rs/core-plugins/src/catalog.rs +++ b/codex-rs/core-plugins/src/catalog.rs @@ -1,32 +1,63 @@ -//! Source-neutral plugin discovery types. +//! Source-neutral plugin catalogs, reconciliation, and source-aware enablement policy. +use codex_config::types::PluginConfig; +use codex_core_plugin_common::PluginIdentity; use codex_utils_path_uri::PathUri; +use serde_json::Value; use std::collections::BTreeMap; +use std::collections::HashMap; -/// One source's complete discovery snapshot; no entries means no plugins were found. +/// One source's discovery snapshot; only a complete empty catalog establishes absence. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct PluginCatalog { pub entries: Vec, pub warnings: Vec, + /// Listing omitted entries. + pub is_partial: bool, } /// Manifest metadata and every location that can supply the same logical plugin. #[derive(Clone, PartialEq, Eq)] pub struct PluginCatalogEntry { pub id: PluginIdentity, + pub enabled: bool, pub display_name: String, - pub version: Option, - /// Serialized `.mcp.json` server declarations keyed by their authored names. - pub mcp_servers: BTreeMap, + pub version: String, + /// Raw MCP server declarations keyed by their authored names. + pub mcp_servers: BTreeMap, pub connector_ids: Vec, pub locations: Vec, } +impl PluginCatalogEntry { + /// Disabled entries never project. Otherwise Cloud membership supplies enablement; + /// executor-only entries require every installation's local plugin ID in the supplied + /// config, where omitted enablement retains the configured plugin's default. + /// A remote identity alone does not establish Cloud membership. + pub fn is_enabled(&self, configured_plugins: &HashMap) -> bool { + if !self.enabled { + return false; + } + let is_cloud_plugin = self + .locations + .iter() + .any(|location| matches!(location, PluginSourceLocation::Cloud { .. })); + is_cloud_plugin + || self.locations.iter().all(|location| match location { + PluginSourceLocation::Executor { plugin_id, .. } => configured_plugins + .get(plugin_id) + .is_some_and(|plugin| plugin.enabled.unwrap_or(true)), + PluginSourceLocation::Cloud { .. } => true, + }) + } +} + impl std::fmt::Debug for PluginCatalogEntry { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter .debug_struct("PluginCatalogEntry") .field("id", &self.id) + .field("enabled", &self.enabled) .field("display_name", &self.display_name) .field("version", &self.version) // Declaration values can contain secrets, so only log their keys. @@ -40,27 +71,8 @@ impl std::fmt::Debug for PluginCatalogEntry { } } -/// Stable key used to merge discoveries of the same logical plugin. -/// Identity is independent of whether the plugin is supplied by the cloud or an executor. -#[derive(Clone, Debug, PartialEq, Eq, Hash)] -pub enum PluginIdentity { - /// Stable ID assigned by plugin-service and shared by its materialized installations. - Remote { remote_plugin_id: String }, - /// Local `name@marketplace` config key for a plugin without a remote ID. - Local { plugin_id: String }, -} - -impl PluginIdentity { - pub fn as_str(&self) -> &str { - match self { - Self::Remote { remote_plugin_id } => remote_plugin_id, - Self::Local { plugin_id } => plugin_id, - } - } -} - /// A source that can supply a plugin; cloud URIs are opaque, not local paths. -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Hash)] pub enum PluginSourceLocation { Cloud { resource_uri: String, @@ -74,3 +86,54 @@ pub enum PluginSourceLocation { root: PathUri, }, } + +/// Uses Cloud membership and metadata, retaining version-matched installations in order. +/// Unavailable Cloud data preserves executor enablement. With a Cloud catalog, unmatched remote +/// plugins are retained but disabled. Local plugins retain their existing enablement. +pub fn merge_catalogs( + executor: Option<&PluginCatalog>, + cloud: Option<&PluginCatalog>, +) -> PluginCatalog { + let Some(executor) = executor else { + return cloud.cloned().unwrap_or_default(); + }; + let Some(cloud) = cloud else { + return executor.clone(); + }; + let mut merged = cloud.clone(); + merged.is_partial |= executor.is_partial; + merged.warnings.extend(executor.warnings.iter().cloned()); + for installation in &executor.entries { + if matches!(&installation.id, PluginIdentity::Local { .. }) { + merged.entries.push(installation.clone()); + continue; + } + // Appended executor entries are not Cloud definitions and cannot authorize each other. + let Some(definition) = merged.entries[..cloud.entries.len()] + .iter_mut() + .find(|definition| definition.id == installation.id) + else { + let mut disabled = installation.clone(); + disabled.enabled = false; + merged.entries.push(disabled); + continue; + }; + if definition.version != installation.version { + merged.warnings.push(format!( + "plugin `{}` has different cloud and executor versions; retaining the cloud definition", + definition.id.as_str(), + )); + continue; + } + for location in &installation.locations { + if !definition.locations.contains(location) { + definition.locations.push(location.clone()); + } + } + } + merged +} + +#[cfg(test)] +#[path = "catalog_tests.rs"] +mod tests; diff --git a/codex-rs/core-plugins/src/catalog_tests.rs b/codex-rs/core-plugins/src/catalog_tests.rs new file mode 100644 index 0000000000..7a68869c9f --- /dev/null +++ b/codex-rs/core-plugins/src/catalog_tests.rs @@ -0,0 +1,164 @@ +//! Tests Cloud catalog authority, version matching, and installation reconciliation. + +use super::PluginCatalog; +use super::PluginCatalogEntry; +use super::PluginIdentity; +use super::PluginSourceLocation; +use super::merge_catalogs; +use codex_utils_path_uri::PathUri; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; + +fn cloud_plugin(remote_id: &str) -> PluginCatalogEntry { + PluginCatalogEntry { + enabled: true, + id: PluginIdentity::Remote { + remote_plugin_id: remote_id.to_string(), + }, + display_name: "Example".to_string(), + version: "1.0.0".to_string(), + mcp_servers: BTreeMap::from([( + "example".to_string(), + serde_json::json!({"command":"node","args":["server.js"]}), + )]), + connector_ids: vec!["connector_example".to_string()], + locations: vec![PluginSourceLocation::Cloud { + resource_uri: format!("plugin://{remote_id}"), + bundle_uri: None, + }], + } +} + +fn executor_plugin(remote_id: &str, environment_id: &str) -> anyhow::Result { + let mut entry = cloud_plugin(remote_id); + entry.locations = vec![PluginSourceLocation::Executor { + environment_id: environment_id.to_string(), + plugin_id: "example@marketplace".to_string(), + root: PathUri::parse("file:///plugins/example/1.0.0")?, + }]; + Ok(entry) +} + +#[test] +fn merged_catalog_prefers_cloud_and_matches_executor_versions() -> anyhow::Result<()> { + let definition = cloud_plugin("plugin_a"); + let mut first = executor_plugin("plugin_a", "executor-a")?; + first.display_name = "Installed display name".to_string(); + first.mcp_servers.clear(); + first.connector_ids.clear(); + let second = executor_plugin("plugin_a", "executor-b")?; + let mut incompatible = executor_plugin("plugin_a", "executor-c")?; + incompatible.version = "2.0.0".to_string(); + let mut expected = definition.clone(); + expected.locations.extend(first.locations.clone()); + expected.locations.extend(second.locations.clone()); + + let executor = PluginCatalog { + entries: vec![first.clone(), first, second, incompatible], + warnings: vec!["executor warning".to_string()], + ..Default::default() + }; + let cloud = PluginCatalog { + entries: vec![definition], + warnings: vec!["cloud warning".to_string()], + ..Default::default() + }; + assert_eq!( + merge_catalogs(Some(&executor), Some(&cloud)), + PluginCatalog { + entries: vec![expected], + warnings: vec![ + "cloud warning".to_string(), + "executor warning".to_string(), + "plugin `plugin_a` has different cloud and executor versions; retaining the cloud definition" + .to_string(), + ], + ..Default::default() + } + ); + Ok(()) +} + +#[test] +fn cloud_catalog_controls_remote_plugins_but_preserves_local_plugins() -> anyhow::Result<()> { + let installed = executor_plugin("plugin_a", "executor-a")?; + let mut local = executor_plugin("plugin_a", "executor-b")?; + local.id = PluginIdentity::Local { + plugin_id: "example@marketplace".to_string(), + }; + let executor = PluginCatalog { + entries: vec![installed.clone(), local.clone()], + ..Default::default() + }; + let local_only = PluginCatalog { + entries: vec![local.clone()], + ..Default::default() + }; + // An available Cloud catalog controls remote entries; local entries are unaffected. + let mut other_installation = executor_plugin("plugin_a", "executor-c")?; + other_installation.version = "2.0.0".to_string(); + let unmatched = PluginCatalog { + entries: vec![installed.clone(), other_installation.clone(), local.clone()], + ..Default::default() + }; + let mut disabled = installed.clone(); + disabled.enabled = false; + other_installation.enabled = false; + let disabled_entries = vec![disabled, other_installation, local.clone()]; + assert_eq!(merge_catalogs(Some(&unmatched), /*cloud*/ None), unmatched); + let explicitly_disabled = PluginCatalog { + entries: disabled_entries.clone(), + ..Default::default() + }; + assert_eq!( + merge_catalogs(Some(&explicitly_disabled), /*cloud*/ None), + explicitly_disabled, + ); + for cloud in [ + Some(PluginCatalog::default()), + Some(PluginCatalog { + is_partial: true, + warnings: vec!["partial cloud listing".to_string()], + ..Default::default() + }), + ] { + assert_eq!( + merge_catalogs(Some(&unmatched), cloud.as_ref()), + PluginCatalog { + entries: disabled_entries.clone(), + is_partial: cloud.as_ref().is_some_and(|catalog| catalog.is_partial), + warnings: cloud.map(|catalog| catalog.warnings).unwrap_or_default(), + }, + ); + } + + let definition = cloud_plugin("plugin_a"); + let cloud = PluginCatalog { + entries: vec![definition.clone()], + ..Default::default() + }; + assert_eq!(merge_catalogs(/*executor*/ None, Some(&cloud)), cloud); + assert_eq!( + merge_catalogs(/*executor*/ None, /*cloud*/ None), + PluginCatalog::default(), + ); + let mut authorized = definition.clone(); + authorized.locations.extend(installed.locations); + assert_eq!( + merge_catalogs(Some(&executor), Some(&cloud)), + PluginCatalog { + entries: vec![authorized, local.clone()], + ..Default::default() + } + ); + + // Removing an installation preserves its Cloud definition without the executor route. + assert_eq!( + merge_catalogs(Some(&local_only), Some(&cloud)), + PluginCatalog { + entries: vec![definition, local], + ..Default::default() + } + ); + Ok(()) +} diff --git a/codex-rs/core-plugins/src/executor_catalog.rs b/codex-rs/core-plugins/src/executor_catalog.rs new file mode 100644 index 0000000000..355e2bc596 --- /dev/null +++ b/codex-rs/core-plugins/src/executor_catalog.rs @@ -0,0 +1,97 @@ +//! Converts V2 discovery contents to inert plugin catalogs without additional I/O. + +use crate::PluginCatalog; +use crate::PluginCatalogEntry; +use crate::PluginIdentity; +use crate::PluginSourceLocation; +use codex_connectors::parse_plugin_app_config; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; +use codex_exec_server::ExecutorPlugin; +use codex_mcp::parse_plugin_mcp_declarations; + +/// Projects V2 inventories into a plugin catalog without additional filesystem access. +pub fn catalog_from_snapshots( + snapshots: &[ExecutorCapabilityDiscoveryV2Snapshot], +) -> PluginCatalog { + let mut catalog = PluginCatalog::default(); + for snapshot in snapshots { + let inventory = match &snapshot.result { + Ok(inventory) => inventory, + Err(error) => { + catalog.warnings.push(format!( + "could not discover plugins in environment `{}`; skipping its plugin inventory: {error}", + snapshot.environment_id + )); + continue; + } + }; + catalog.warnings.extend(inventory.warnings.iter().cloned()); + for plugin in &inventory.plugins { + let (entry, warnings) = from_discovered_plugin(&snapshot.environment_id, plugin); + catalog.entries.push(entry); + catalog.warnings.extend(warnings); + } + } + catalog +} + +fn from_discovered_plugin( + environment_id: &str, + plugin: &ExecutorPlugin, +) -> (PluginCatalogEntry, Vec) { + let plugin_id = &plugin.id; + let mut warnings = Vec::new(); + let mcp_servers = match plugin + .files + .mcp_config + .as_ref() + .map(|file| parse_plugin_mcp_declarations(&file.contents)) + { + Some(Ok(servers)) => servers, + Some(Err(error)) => { + warnings.push(format!("plugin `{plugin_id}` MCP configuration: {error}")); + Default::default() + } + None => Default::default(), + }; + let connector_ids = match plugin + .files + .apps_config + .as_ref() + .map(|file| parse_plugin_app_config(&file.contents)) + { + Some(Ok(apps)) => apps.into_iter().map(|app| app.connector_id.0).collect(), + Some(Err(error)) => { + warnings.push(format!("plugin `{plugin_id}` app configuration: {error}")); + Vec::new() + } + None => Vec::new(), + }; + ( + PluginCatalogEntry { + enabled: true, + id: match &plugin.remote_plugin_id { + Some(remote_plugin_id) => PluginIdentity::Remote { + remote_plugin_id: remote_plugin_id.clone(), + }, + None => PluginIdentity::Local { + plugin_id: plugin.id.clone(), + }, + }, + display_name: plugin.display_name.clone(), + version: plugin.version.clone(), + mcp_servers, + connector_ids, + locations: vec![PluginSourceLocation::Executor { + environment_id: environment_id.to_string(), + plugin_id: plugin_id.to_string(), + root: plugin.root.clone(), + }], + }, + warnings, + ) +} + +#[cfg(test)] +#[path = "executor_catalog_tests.rs"] +mod tests; diff --git a/codex-rs/core-plugins/src/executor_catalog_tests.rs b/codex-rs/core-plugins/src/executor_catalog_tests.rs new file mode 100644 index 0000000000..83c6e3d837 --- /dev/null +++ b/codex-rs/core-plugins/src/executor_catalog_tests.rs @@ -0,0 +1,73 @@ +//! Tests conversion of DiscoveryV2 payloads into inert plugin catalogs. + +use super::catalog_from_snapshots; +use crate::PluginCatalog; +use crate::PluginCatalogEntry; +use crate::PluginIdentity; +use crate::PluginSourceLocation; +use codex_exec_server::CapabilityTextFile; +use codex_exec_server::DiscoverV2CapabilitiesResponse; +use codex_exec_server::DiscoveredPluginFiles; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; +use codex_exec_server::ExecutorPlugin; +use codex_utils_path_uri::PathUri; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use std::sync::Arc; + +#[test] +fn discovered_mcp_contents_populate_catalog() -> anyhow::Result<()> { + let root = PathUri::parse("file:///plugins/example/1.0.0")?; + let manifest_path = root.join(".codex-plugin/plugin.json")?; + let declarations = serde_json::json!({"example": {"command": "node"}}); + let snapshot = ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: "executor-a".to_string(), + result: Ok(Arc::new(DiscoverV2CapabilitiesResponse { + plugins: vec![ExecutorPlugin { + id: "example@marketplace".to_string(), + display_name: "Example".to_string(), + remote_plugin_id: Some("plugin_a".to_string()), + version: "1.0.0".to_string(), + files: DiscoveredPluginFiles { + manifest: CapabilityTextFile { + path: manifest_path.clone(), + contents: r#"{"name":"example"}"#.to_string(), + }, + mcp_config: Some(CapabilityTextFile { + path: manifest_path, + contents: declarations.to_string(), + }), + apps_config: None, + }, + root: root.clone(), + skills: Vec::new(), + }], + ..Default::default() + })), + }; + assert_eq!( + catalog_from_snapshots(&[snapshot]), + PluginCatalog { + entries: vec![PluginCatalogEntry { + enabled: true, + id: PluginIdentity::Remote { + remote_plugin_id: "plugin_a".to_string(), + }, + display_name: "Example".to_string(), + version: "1.0.0".to_string(), + mcp_servers: BTreeMap::from([( + "example".to_string(), + serde_json::json!({"command": "node"}), + )]), + connector_ids: Vec::new(), + locations: vec![PluginSourceLocation::Executor { + environment_id: "executor-a".to_string(), + plugin_id: "example@marketplace".to_string(), + root, + }], + }], + ..Default::default() + }, + ); + Ok(()) +} diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index b542f867c0..9cc617a548 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -4,6 +4,7 @@ mod catalog; mod command_migration; mod discoverable; mod error_subtype; +mod executor_catalog; mod executor_hooks; mod executor_provider; mod git_policy; @@ -61,8 +62,9 @@ pub use artifact_operation::ArtifactOperation; pub use artifact_operation::recognize_artifact_operation; pub use catalog::PluginCatalog; pub use catalog::PluginCatalogEntry; -pub use catalog::PluginIdentity; pub use catalog::PluginSourceLocation; +pub use catalog::merge_catalogs; +pub use codex_core_plugin_common::PluginIdentity; pub use command_migration::CommandDescriptionMode; pub use command_migration::CommandMigrationProfile; pub use command_migration::RewriteProfile as CommandRewriteProfile; @@ -71,6 +73,7 @@ pub use command_migration::import_commands_with_profile; pub use command_migration::missing_command_names_with_profile; pub use discoverable::ToolSuggestDiscoverablePlugin; pub use discoverable::ToolSuggestPluginDiscoveryInput; +pub use executor_catalog::catalog_from_snapshots; pub use executor_hooks::executor_plugin_hook_sources; pub use executor_provider::ExecutorPluginProvider; pub use executor_provider::ExecutorPluginProviderError; @@ -126,3 +129,5 @@ pub use script_attribution::PluginCommandAttribution; pub use script_attribution::PluginMeasurementTarget; pub use script_attribution::TrustedPluginRoots; pub use script_attribution::command_script_arguments; + +pub use marketplace_policy::configured_plugins_from_stack; diff --git a/codex-rs/core-plugins/src/marketplace_policy.rs b/codex-rs/core-plugins/src/marketplace_policy.rs index 46879d2a85..64b0c38e89 100644 --- a/codex-rs/core-plugins/src/marketplace_policy.rs +++ b/codex-rs/core-plugins/src/marketplace_policy.rs @@ -295,7 +295,7 @@ fn allowed_configured_marketplace_names_with_policy( .collect() } -pub(crate) fn configured_plugins_from_stack( +pub fn configured_plugins_from_stack( config_layer_stack: &ConfigLayerStack, codex_home: &Path, ) -> HashMap { diff --git a/codex-rs/core-plugins/src/provider.rs b/codex-rs/core-plugins/src/provider.rs index 3ebabcc961..39106e8d4c 100644 --- a/codex-rs/core-plugins/src/provider.rs +++ b/codex-rs/core-plugins/src/provider.rs @@ -38,7 +38,8 @@ pub type PluginProviderFuture<'a, T> = /// Cloud providers implement `list`; executor listing is not yet supported. /// Package access uses [`crate::ExecutorPluginProvider::resolve_bound`] directly. pub trait PluginProvider: Send + Sync { - /// Returns a complete snapshot; on error, callers retain the previous snapshot. + /// Returns a snapshot; on error, callers retain the previous snapshot. + /// Providers set `is_partial` when listing omits entries; warnings are diagnostic only. fn list(&self, _query: PluginListQuery) -> PluginProviderFuture<'_, PluginCatalog> { Box::pin(async { Ok(PluginCatalog::default()) }) } diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index c28e7b86f7..ccc96cc488 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; use crate::config::Config; @@ -7,6 +8,7 @@ use codex_config::McpServerConfig; use codex_config::types::PluginsConfigToml; use codex_connectors::ConnectorRuntimeManager; use codex_connectors::PluginConnectorSource; +use codex_core_plugins::PluginIdentity as CatalogPluginIdentity; use codex_core_plugins::PluginsManager; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; @@ -202,32 +204,54 @@ impl McpManager { let roots = context .ready_selected_capability_roots() .unwrap_or_default(); - let mut selected = SelectedPluginSnapshot::default(); - if roots.is_empty() { + let mut selected = SelectedPluginSnapshot { + disabled_plugin_ids: disabled_plugin_ids.to_vec(), + ..Default::default() + }; + let v2 = context.executor_capability_discovery_v2().is_some(); + let available_roots = available_plugin_roots(context); + if roots.is_empty() && !v2 { return selected; } for contributor in self.extensions.mcp_server_contributors() { for SelectedPlugin { selected_root_id, plugin_id, + identity, + enabled, .. } in contributor.selected_plugins(context, plugins_config).await { - if !roots.iter().any(|root| root.id == selected_root_id) { + if !available_roots.contains(&selected_root_id) { continue; } - if !context.config().features.enabled(Feature::Plugins) + if !enabled + || !context.config().features.enabled(Feature::Plugins) || !plugins_config.allows_plugin(&plugin_id) || disabled_plugin_ids.contains(&plugin_id) + || disabled_plugin_ids.iter().any(|id| id == identity.as_str()) { selected.disabled_plugin_roots.push(selected_root_id); + selected.disabled_plugin_ids.push(plugin_id); } else { selected.plugins.push(SelectedPluginIdentity { selected_root_id: Some(selected_root_id), - plugin_id, + plugin_id: identity, }); } } + if context.config().features.enabled(Feature::Plugins) { + for plugin_id in contributor.hosted_plugin_ids(context) { + if !disabled_plugin_ids.contains(&plugin_id) { + selected.plugins.push(SelectedPluginIdentity { + selected_root_id: None, + plugin_id: CatalogPluginIdentity::Remote { + remote_plugin_id: plugin_id, + }, + }); + } + } + } } selected } @@ -245,6 +269,7 @@ impl McpManager { let mut selected_plugin_registrations = Vec::new(); let mut selected_plugins = Vec::new(); let mut disabled_plugin_roots = Vec::new(); + let mut suppressed_plugin_ids = disabled_plugin_ids.to_vec(); let mut overlays = Vec::new(); // A contributor can emit multiple ordered actions, so order each action globally rather // than enumerating contributors. @@ -256,22 +281,27 @@ impl McpManager { for SelectedPlugin { selected_root_id, plugin_id, + identity, + enabled, mcp, } in contributor.selected_plugins(context, plugins_config).await { - if !plugins_config.allows_plugin(&plugin_id) { + if !enabled || !plugins_config.allows_plugin(&plugin_id) { disabled_plugin_roots.push(selected_root_id); + suppressed_plugin_ids.push(plugin_id); continue; } let selection_order = plugin_selection_order; plugin_selection_order += 1; - let disabled = disabled_plugin_ids.contains(&plugin_id); + let disabled = disabled_plugin_ids.contains(&plugin_id) + || disabled_plugin_ids.iter().any(|id| id == identity.as_str()); if !config.features.enabled(Feature::Plugins) || disabled { disabled_plugin_roots.push(selected_root_id); + suppressed_plugin_ids.push(plugin_id.clone()); } else { selected_plugins.push(SelectedPluginIdentity { selected_root_id: Some(selected_root_id), - plugin_id: plugin_id.clone(), + plugin_id: identity, }); } let contribution = mcp.await; @@ -355,7 +385,9 @@ impl McpManager { if !disabled_plugin_ids.contains(&plugin_id) { selected_plugins.push(SelectedPluginIdentity { selected_root_id: None, - plugin_id: plugin_id.clone(), + plugin_id: CatalogPluginIdentity::Remote { + remote_plugin_id: plugin_id.clone(), + }, }); } if !connector_ids.is_empty() { @@ -456,6 +488,7 @@ impl McpManager { selected_plugins: SelectedPluginSnapshot { plugins: selected_plugins, disabled_plugin_roots, + disabled_plugin_ids: suppressed_plugin_ids, }, } } @@ -482,3 +515,26 @@ impl McpManager { effective_mcp_servers(&mcp_config, auth) } } + +fn available_plugin_roots(context: McpServerContributionContext<'_, Config>) -> HashSet { + let mut roots = context + .ready_selected_capability_roots() + .unwrap_or_default() + .iter() + .map(|root| root.id.clone()) + .collect::>(); + for snapshot in context + .executor_capability_discovery_v2() + .unwrap_or_default() + { + if let Ok(inventory) = &snapshot.result { + roots.extend( + inventory + .plugins + .iter() + .map(|plugin| format!("{}:{}", snapshot.environment_id, plugin.id)), + ); + } + } + roots +} diff --git a/codex-rs/core/src/session/active_plugin_inventory_tests.rs b/codex-rs/core/src/session/active_plugin_inventory_tests.rs index 918ad88271..4c1433689e 100644 --- a/codex-rs/core/src/session/active_plugin_inventory_tests.rs +++ b/codex-rs/core/src/session/active_plugin_inventory_tests.rs @@ -79,7 +79,9 @@ async fn active_plugin_ids_select_and_validate_by_source() { .into_iter() .map(|plugin_id| SelectedPluginIdentity { selected_root_id: Some("capability-root".to_string()), - plugin_id: plugin_id.to_string(), + plugin_id: codex_core_plugins::PluginIdentity::Local { + plugin_id: plugin_id.to_string(), + }, }) .collect::>(); turn_context.active_host_plugin_identities = Some(identities); @@ -125,7 +127,9 @@ async fn active_plugin_ids_enforce_bounds_after_selection_and_deduplication() { let id = format!("{}@local", "p".repeat(length - 6)); let selected_plugin = SelectedPluginIdentity { selected_root_id: Some("capability-root".to_string()), - plugin_id: id.clone(), + plugin_id: codex_core_plugins::PluginIdentity::Local { + plugin_id: id.clone(), + }, }; turn_context.active_host_plugin_identities = Some(Vec::new()); let selected_plugins = SelectedPluginSnapshot { @@ -154,7 +158,9 @@ async fn active_plugin_ids_enforce_bounds_after_selection_and_deduplication() { .iter() .map(|id| SelectedPluginIdentity { selected_root_id: Some("capability-root".to_string()), - plugin_id: id.clone(), + plugin_id: codex_core_plugins::PluginIdentity::Local { + plugin_id: id.clone(), + }, }) .collect::>(); turn_context.active_host_plugin_identities = Some(identities); diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index cf4bc7b056..1c2f97766a 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -597,12 +597,25 @@ impl TurnContext { const MAX_TELEMETRY_PLUGIN_ID_BYTES: usize = 128; let mut identities = self.active_host_plugin_identities.clone()?; - // Selected roots provide a package key, not a remote identity for that object. + // Selected catalogs preserve local package keys or canonical remote identities. if let Some(selected) = selected { - identities.extend(selected.plugins.iter().map(|plugin| PluginIdentity { - plugin_id: plugin.plugin_id.clone(), - remote_plugin_id: None, - })); + identities.extend( + selected + .plugins + .iter() + .map(|plugin| match &plugin.plugin_id { + codex_core_plugins::PluginIdentity::Local { plugin_id } => PluginIdentity { + plugin_id: plugin_id.clone(), + remote_plugin_id: None, + }, + codex_core_plugins::PluginIdentity::Remote { remote_plugin_id } => { + PluginIdentity { + plugin_id: String::new(), + remote_plugin_id: Some(remote_plugin_id.clone()), + } + } + }), + ); } let mut ids = Vec::with_capacity(identities.len()); for identity in identities { diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 0800effb8a..533dfa44d3 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1675,6 +1675,10 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() vec![codex_extension_api::SelectedPlugin { selected_root_id: selected_root_id.clone(), plugin_id: plugin_id.clone(), + identity: codex_core_plugins::PluginIdentity::Local { + plugin_id: plugin_id.clone(), + }, + enabled: true, mcp: Box::pin(async move { mcp_loaded.fetch_add(1, Ordering::Relaxed); codex_extension_api::SelectedPluginContribution { diff --git a/codex-rs/core/tests/suite/executor_discovery_v2.rs b/codex-rs/core/tests/suite/executor_discovery_v2.rs index 55393cb72d..7d3ce5e77b 100644 --- a/codex-rs/core/tests/suite/executor_discovery_v2.rs +++ b/codex-rs/core/tests/suite/executor_discovery_v2.rs @@ -4,14 +4,19 @@ use anyhow::Context; use anyhow::Result; use codex_core::config::Config; use codex_core::windows_sandbox::WindowsSandboxLevelExt; +use codex_core_plugins::PluginIdentity; use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::HttpRequestBodyDeltaNotification; use codex_exec_server::HttpRequestParams; +use codex_extension_api::ContextContributor; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; use codex_extension_api::McpServerContributor; +use codex_extension_api::SelectedPluginSnapshot; +use codex_extension_api::WorldStateContributionInput; +use codex_extension_api::WorldStateSectionContribution; use codex_features::Feature; use codex_protocol::config_types::WindowsSandboxLevel; use core_test_support::responses; @@ -33,6 +38,7 @@ use tokio_util::task::AbortOnDropHandle; struct DiscoveryRecorder { snapshots: Arc>>, + selected_plugins: Arc>>, } impl McpServerContributor for DiscoveryRecorder { @@ -42,12 +48,32 @@ impl McpServerContributor for DiscoveryRecorder { fn contribute<'a>( &'a self, - context: McpServerContributionContext<'a, Config>, + _context: McpServerContributionContext<'a, Config>, ) -> ExtensionFuture<'a, Vec> { + Box::pin(async { + vec![McpServerContribution::HostedPluginConnectors { + plugin_id: "plugin_sentinel".to_string(), + plugin_display_name: "Inert sentinel".to_string(), + connector_ids: Vec::new(), + }] + }) + } + + fn hosted_plugin_ids(&self, _context: McpServerContributionContext<'_, Config>) -> Vec { + vec!["plugin_sentinel".to_string()] + } + + fn selected_plugins<'a>( + &'a self, + context: McpServerContributionContext<'a, Config>, + _plugins_config: &'a codex_config::types::PluginsConfigToml, + ) -> ExtensionFuture<'a, Vec>> { Box::pin(async move { + let mut package_roots = vec!["unavailable-root".to_string()]; let selected_roots = context .ready_selected_capability_roots() .unwrap_or_default(); + package_roots.extend(selected_roots.iter().map(|root| root.id.clone())); if !selected_roots.is_empty() { let legacy = context.executor_capability_discovery(); assert_eq!( @@ -60,6 +86,49 @@ impl McpServerContributor for DiscoveryRecorder { } if let Some(snapshots) = context.executor_capability_discovery_v2() { self.snapshots.lock().await.extend_from_slice(snapshots); + for snapshot in snapshots { + if let Ok(inventory) = &snapshot.result { + package_roots.extend( + inventory + .plugins + .iter() + .map(|plugin| format!("{}:{}", snapshot.environment_id, plugin.id)), + ); + } + } + } + package_roots + .into_iter() + .map(|selected_root_id| codex_extension_api::SelectedPlugin { + selected_root_id, + identity: PluginIdentity::Local { + plugin_id: "inert-sentinel@market".to_string(), + }, + plugin_id: "inert-sentinel@market".to_string(), + enabled: true, + mcp: Box::pin(async { + codex_extension_api::SelectedPluginContribution { + plugin_display_name: "Inert sentinel".to_string(), + source_environment_id: codex_exec_server::REMOTE_ENVIRONMENT_ID + .to_string(), + servers: Vec::new(), + connector_ids: Vec::new(), + } + }), + }) + .collect() + }) + } +} + +impl ContextContributor for DiscoveryRecorder { + fn contribute_world_state<'a>( + &'a self, + input: WorldStateContributionInput<'a>, + ) -> ExtensionFuture<'a, Vec> { + Box::pin(async move { + if let Some(snapshot) = input.step_store.get::() { + *self.selected_plugins.lock().await = Some(snapshot.as_ref().clone()); } Vec::new() }) @@ -274,16 +343,20 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> { let executor = Executor::start(supported, DiscoveryFailure::None).await?; let model = responses::start_mock_server().await; let observed = Arc::new(Mutex::new(Vec::new())); + let selected_plugins = Arc::new(Mutex::new(None)); let mut extensions = ExtensionRegistryBuilder::new(); let recorder = Arc::new(DiscoveryRecorder { snapshots: Arc::clone(&observed), + selected_plugins: Arc::clone(&selected_plugins), }); + extensions.prompt_contributor(recorder.clone()); extensions.mcp_server_contributor(recorder); let test = test_codex() .with_extensions(Arc::new(extensions.build())) .with_exec_server_url(executor.url.clone()) .with_config(move |config| { config.update_plan_enabled = true; + assert!(config.features.enable(Feature::Plugins).is_ok()); assert!( config .features @@ -345,19 +418,24 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> { exec_policy: None, mcp_policy: None, network_policy: None, - selected_capability_roots: vec![codex_protocol::capabilities::SelectedCapabilityRoot { - id: "legacy-root".to_string(), - location: codex_protocol::capabilities::CapabilityRootLocation::Environment { - environment_id: selection.environment_id.clone(), - path: cwd.clone(), - }, - }], + selected_capability_roots: if acquires_v2 { + Vec::new() + } else { + vec![codex_protocol::capabilities::SelectedCapabilityRoot { + id: "legacy-root".to_string(), + location: codex_protocol::capabilities::CapabilityRootLocation::Environment { + environment_id: selection.environment_id.clone(), + path: cwd.clone(), + }, + }] + }, }); // Only the successful lifecycle case needs two turns and multiple model steps. let turns = if acquires_v2 { 2 } else { 1 }; let mut previous_inventory = None; for turn in 0..turns { observed.lock().await.clear(); + *selected_plugins.lock().await = None; let before_turn = calls().await; let mut bodies = Vec::new(); if acquires_v2 { @@ -381,6 +459,37 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> { } test.submit_turn_with_environments("Say hello.", Some(vec![selection.clone()])) .await?; + let executor_root = if acquires_v2 { + format!("{}:inert-sentinel@market", selection.environment_id) + } else { + "legacy-root".to_string() + }; + let local_identity = PluginIdentity::Local { + plugin_id: "inert-sentinel@market".to_string(), + }; + // Both paths publish accepted Cloud packages and available installations only. + let expected_plugins = vec![ + (Some(executor_root), local_identity), + ( + None, + PluginIdentity::Remote { + remote_plugin_id: "plugin_sentinel".to_string(), + }, + ), + ]; + { + let selected_plugins = selected_plugins.lock().await; + let snapshot = selected_plugins.as_ref().context("turn plugin snapshot")?; + assert_eq!( + snapshot + .plugins + .iter() + .map(|plugin| (plugin.selected_root_id.clone(), plugin.plugin_id.clone())) + .collect::>(), + expected_plugins, + ); + assert_eq!(snapshot.disabled_plugin_ids, Vec::::new()); + } let model_requests = responses .iter() .flat_map(core_test_support::responses::ResponseMock::requests) @@ -426,11 +535,13 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> { } } let requests = executor.requests.lock().await.clone(); - assert!( - requests - .iter() - .any(|request| request["method"] == "capabilityRoots/discoverV1") - ); + if !acquires_v2 { + assert!( + requests + .iter() + .any(|request| request["method"] == "capabilityRoots/discoverV1") + ); + } let discovery = requests .iter() .filter(|request| request["method"] == "capabilities/discoverV2") diff --git a/codex-rs/ext/extension-api/Cargo.toml b/codex-rs/ext/extension-api/Cargo.toml index 4b487226eb..2c2bd6d384 100644 --- a/codex-rs/ext/extension-api/Cargo.toml +++ b/codex-rs/ext/extension-api/Cargo.toml @@ -14,6 +14,7 @@ doctest = false workspace = true [dependencies] +codex-core-plugin-common = { workspace = true } codex-api = { workspace = true } futures = { workspace = true } codex-guardian-context = { workspace = true } diff --git a/codex-rs/ext/extension-api/src/contributors.rs b/codex-rs/ext/extension-api/src/contributors.rs index 44160e2b17..10947856fc 100644 --- a/codex-rs/ext/extension-api/src/contributors.rs +++ b/codex-rs/ext/extension-api/src/contributors.rs @@ -90,6 +90,11 @@ pub trait McpServerContributor: Send + Sync { context: McpServerContributionContext<'a, C>, ) -> ExtensionFuture<'a, Vec>; + /// Reads canonical hosted-plugin IDs from published state without discovery or MCP I/O. + fn hosted_plugin_ids(&self, _context: McpServerContributionContext<'_, C>) -> Vec { + Vec::new() + } + /// Declares executor plugins, including those without MCP servers or connectors. Each /// declaration identifies the plugin once; its deferred MCP data cannot change that identity. /// Return plugins in precedence order: across contributors in registration order, the first diff --git a/codex-rs/ext/extension-api/src/contributors/mcp.rs b/codex-rs/ext/extension-api/src/contributors/mcp.rs index bec685a921..fc8df636a9 100644 --- a/codex-rs/ext/extension-api/src/contributors/mcp.rs +++ b/codex-rs/ext/extension-api/src/contributors/mcp.rs @@ -1,4 +1,5 @@ use codex_config::McpServerConfig; +use codex_core_plugin_common::PluginIdentity; use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; use codex_exec_server_protocol::ExecutorCapabilityDiscoveryV2Snapshot; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -151,7 +152,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { self.ready_selected_capability_roots } - /// Returns the V2 inventories supplied by the host, when available. + /// `Some` selects V2, including an empty inventory list; `None` selects legacy discovery. pub fn executor_capability_discovery_v2( &self, ) -> Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]> { @@ -170,6 +171,7 @@ pub struct SelectedPluginSnapshot { pub plugins: Vec, /// Selected plugin roots suppressed by the effective plugin feature policy. pub disabled_plugin_roots: Vec, + pub disabled_plugin_ids: Vec, } /// The configured identity of a selected plugin and, when present, the root owning its skills. @@ -177,7 +179,7 @@ pub struct SelectedPluginSnapshot { pub struct SelectedPluginIdentity { /// Hosted plugins have no executor root and cannot own skills from an executor folder. pub selected_root_id: Option, - pub plugin_id: String, + pub plugin_id: PluginIdentity, } /// An executor plugin and its deferred MCP data. Callers that only need the identity can drop @@ -186,6 +188,8 @@ pub struct SelectedPluginIdentity { pub struct SelectedPlugin<'a> { pub selected_root_id: String, pub plugin_id: String, + pub identity: PluginIdentity, + pub enabled: bool, pub mcp: ExtensionFuture<'a, SelectedPluginContribution>, } diff --git a/codex-rs/ext/mcp/src/cloud_plugin.rs b/codex-rs/ext/mcp/src/cloud_plugin.rs index 9cabe53f4b..0968ea1863 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin.rs @@ -21,8 +21,8 @@ use std::collections::HashSet; use std::sync::Arc; const MAX_PROJECTED_CLOUD_PLUGINS: usize = 100; -const MAX_PLUGIN_DISPLAY_NAME_CHARS: usize = 64; -const MAX_CONNECTOR_IDS_PER_PLUGIN: usize = 64; +pub(crate) const MAX_PLUGIN_DISPLAY_NAME_CHARS: usize = 64; +pub(crate) const MAX_CONNECTOR_IDS_PER_PLUGIN: usize = 64; impl ThreadLifecycleContributor for PluginContributor { fn on_thread_start<'a>( @@ -123,6 +123,7 @@ pub(crate) fn hosted_plugin_connectors(catalog: &PluginCatalog) -> Vec PluginCatalog { PluginCatalog { entries: vec![PluginCatalogEntry { + enabled: true, id: PluginIdentity::Remote { remote_plugin_id: "plugin_remote".to_string(), }, display_name: "Cloud Demo".to_string(), - version: Some("1.0.0".to_string()), + version: "1.0.0".to_string(), mcp_servers: Default::default(), connector_ids: vec!["calendar".to_string()], locations: vec![codex_core_plugins::PluginSourceLocation::Cloud { @@ -343,7 +349,7 @@ fn cloud_catalog() -> PluginCatalog { bundle_uri: None, }], }], - warnings: Vec::new(), + ..Default::default() } } @@ -384,6 +390,15 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { }; gmail_catalog.entries[0].display_name = "Gmail Demo".to_string(); gmail_catalog.entries[0].connector_ids = vec!["gmail".to_string()]; + let mut disabled_catalog = cloud_catalog(); + disabled_catalog.entries[0].enabled = false; + let mut partial_catalog = gmail_catalog.clone(); + partial_catalog.is_partial = true; + let partial_empty_catalog = PluginCatalog { + is_partial: true, + warnings: vec!["resource listing was truncated".to_string()], + ..Default::default() + }; let provider = Arc::new(CloudProvider::default()); let step_packages = Arc::new(StepPackages::default()); let mut extensions = ExtensionRegistryBuilder::new(); @@ -421,6 +436,30 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { let calendar = ("calendar", "list_events", "Cloud Demo"); let gmail = ("gmail", "search_email", "Gmail Demo"); for (turn, reply, expected_plugin, new_account, tool_call) in [ + ( + "initial-partial", + Ok(partial_catalog.clone()), + Some("plugin_gmail"), + None, + None, + ), + ( + "partial-replaces-partial", + Ok(PluginCatalog { + is_partial: true, + ..cloud_catalog() + }), + Some("plugin_remote"), + None, + Some(calendar), + ), + ( + "partial-empty-replaces-partial", + Ok(partial_empty_catalog.clone()), + None, + None, + None, + ), ( "empty", Ok(PluginCatalog::default()), @@ -428,6 +467,13 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { None, Some(calendar), ), + ( + "partial-after-complete-empty", + Ok(partial_catalog.clone()), + None, + None, + None, + ), ( "discovered", Ok(cloud_catalog()), @@ -435,6 +481,20 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { None, Some(calendar), ), + ( + "partial-refresh", + Ok(partial_catalog.clone()), + Some("plugin_remote"), + None, + Some(calendar), + ), + ( + "partial-empty-refresh", + Ok(partial_empty_catalog), + Some("plugin_remote"), + None, + None, + ), ( "same-account-failure", Err("temporarily unavailable".to_string()), @@ -442,6 +502,30 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { None, None, ), + ( + "complete-with-warning-disables-plugin", + Ok(PluginCatalog { + warnings: vec!["optional bundle hint ignored".to_string()], + ..disabled_catalog + }), + None, + None, + None, + ), + ( + "partial-after-disable", + Ok(partial_catalog.clone()), + None, + None, + None, + ), + ( + "re-enabled", + Ok(cloud_catalog()), + Some("plugin_remote"), + None, + None, + ), ("removed", Ok(PluginCatalog::default()), None, None, None), ( "restored", @@ -457,6 +541,13 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { Some("account-b"), None, ), + ( + "new-account-partial", + Ok(partial_catalog), + Some("plugin_gmail"), + None, + Some(gmail), + ), ( "new-account-discovered", Ok(gmail_catalog), diff --git a/codex-rs/ext/mcp/src/plugin.rs b/codex-rs/ext/mcp/src/plugin.rs index e9ef333e92..bb6db0f5ac 100644 --- a/codex-rs/ext/mcp/src/plugin.rs +++ b/codex-rs/ext/mcp/src/plugin.rs @@ -2,8 +2,12 @@ use codex_config::types::PluginsConfigToml; use codex_connectors_extension::PluginAppProvider; use codex_core::config::Config; use codex_core_plugins::ExecutorPluginProviderError; +use codex_core_plugins::PluginIdentity; use codex_core_plugins::PluginRootOwnership; +use codex_core_plugins::PluginSourceLocation; +use codex_core_plugins::configured_plugins_from_stack; use codex_core_plugins::loader::apply_configured_plugin_mcp_server_policies; +use codex_core_plugins::merge_catalogs; use codex_exec_server::CapabilityRootDiscovery; use codex_exec_server::FileSystemSandboxContext; use codex_extension_api::ExtensionFuture; @@ -210,6 +214,36 @@ impl McpServerContributor for PluginContributor { }) } + fn hosted_plugin_ids(&self, context: McpServerContributionContext<'_, Config>) -> Vec { + if self.providers.cloud.is_none() || !context.config().features.enabled(Feature::Plugins) { + return Vec::new(); + } + let Some(state) = context + .thread_store() + .and_then(codex_extension_api::ExtensionData::get::) + else { + return Vec::new(); + }; + state + .contributor_state() + .cloud_catalog() + .map(|catalog| { + hosted_plugin_connectors(catalog) + .into_iter() + .filter_map(|contribution| { + if let McpServerContribution::HostedPluginConnectors { plugin_id, .. } = + contribution + { + Some(plugin_id) + } else { + None + } + }) + .collect() + }) + .unwrap_or_default() + } + fn selected_plugins<'a>( &'a self, context: McpServerContributionContext<'a, Config>, @@ -228,6 +262,69 @@ impl McpServerContributor for PluginContributor { .ready_selected_capability_roots() .unwrap_or_default(); let mut plugins = Vec::new(); + if let Some(snapshots) = context.executor_capability_discovery_v2() { + let merged = { + let mut cache = state.contributor_state(); + cache.update_executor_catalog(snapshots); + let cloud = (self.providers.cloud.is_some() + && context.config().features.enabled(Feature::Plugins)) + .then(|| cache.cloud_catalog()) + .flatten(); + merge_catalogs(cache.executor_catalog.as_ref(), cloud) + }; + for warning in &merged.warnings { + tracing::warn!(%warning, "plugin catalog merge reported a warning"); + } + let configured = configured_plugins_from_stack( + &context.config().config_layer_stack, + context.config().codex_home.as_path(), + ); + for entry in merged.entries { + for location in &entry.locations { + let PluginSourceLocation::Executor { + environment_id, + plugin_id, + .. + } = location + else { + continue; + }; + let selected_root_id = format!("{environment_id}:{plugin_id}"); + let root_id = selected_root_id.clone(); + let plugin_id = plugin_id.clone(); + let metadata_plugin_id = plugin_id.clone(); + let environment_id = environment_id.clone(); + let captured = entry.clone(); + plugins.push(SelectedPlugin { + selected_root_id, + plugin_id, + identity: entry.id.clone(), + enabled: entry.is_enabled(&configured), + mcp: Box::pin(async move { + let metadata = discovery::v2_metadata_from_catalog(&captured) + .into_iter() + .find(|(id, _)| id == &root_id) + .map(|(_, metadata)| metadata) + .unwrap_or_default(); + let contribution = SelectedPluginContribution { + plugin_display_name: metadata.plugin_display_name, + source_environment_id: environment_id, + servers: metadata.servers, + connector_ids: metadata.connector_ids, + }; + project_metadata( + context.config(), + plugins_config, + &metadata_plugin_id, + contribution, + ) + }), + }); + } + } + return plugins; + } + state.contributor_state().update_executor_catalog(&[]); if let Some(snapshot) = context.executor_capability_discovery() { for root in snapshot.roots() { @@ -270,6 +367,10 @@ impl McpServerContributor for PluginContributor { plugins.push(SelectedPlugin { selected_root_id: root.selected_root.id.clone(), plugin_id: root.selected_root.id.clone(), + identity: PluginIdentity::Local { + plugin_id: root.selected_root.id.clone(), + }, + enabled: true, mcp: Box::pin(async move { let metadata = discovery::metadata_from_discovery( &root.selected_root, @@ -309,6 +410,10 @@ impl McpServerContributor for PluginContributor { plugins.push(SelectedPlugin { selected_root_id: selected_root.id.clone(), plugin_id: selected_root.id.clone(), + identity: PluginIdentity::Local { + plugin_id: selected_root.id.clone(), + }, + enabled: true, mcp: Box::pin(async move { project_metadata( context.config(), @@ -336,6 +441,10 @@ fn denied_plugin(selected_root: &SelectedCapabilityRoot) -> SelectedPlugin<'stat SelectedPlugin { selected_root_id: selected_root.id.clone(), plugin_id: selected_root.id.clone(), + identity: PluginIdentity::Local { + plugin_id: selected_root.id.clone(), + }, + enabled: false, mcp: Box::pin(async move { metadata }), } } diff --git a/codex-rs/ext/mcp/src/plugin/discovery.rs b/codex-rs/ext/mcp/src/plugin/discovery.rs index bdd6f585a1..e3d77b9497 100644 --- a/codex-rs/ext/mcp/src/plugin/discovery.rs +++ b/codex-rs/ext/mcp/src/plugin/discovery.rs @@ -1,13 +1,21 @@ +use crate::cloud_plugin::MAX_CONNECTOR_IDS_PER_PLUGIN; +use crate::cloud_plugin::MAX_PLUGIN_DISPLAY_NAME_CHARS; +use crate::plugin_contributor_state::SelectedPluginMetadata; use codex_connectors::parse_plugin_app_config; +use codex_core_plugins::PluginCatalogEntry; +use codex_core_plugins::PluginSourceLocation; use codex_core_plugins::manifest::UriPluginManifest; use codex_core_plugins::manifest::parse_plugin_manifest_uri; use codex_exec_server::CapabilityRootDiscovery; use codex_exec_server::DiscoveredPluginFiles; use codex_extension_api::SelectedPluginContribution; +use codex_mcp::PluginMcpConfigParseOutcome; +use codex_mcp::normalize_executor_plugin_mcp_declarations; use codex_mcp::parse_executor_plugin_mcp_config; use codex_plugin::manifest::PluginManifestMcpServers; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; +use std::collections::HashSet; pub(super) fn manifest_from_discovery<'a>( selected_root: &SelectedCapabilityRoot, @@ -134,6 +142,71 @@ pub(super) fn metadata_from_discovery( } } +pub(super) fn v2_metadata_from_catalog( + entry: &PluginCatalogEntry, +) -> Vec<(String, SelectedPluginMetadata)> { + let plugin_display_name: String = entry + .display_name + .chars() + .take(MAX_PLUGIN_DISPLAY_NAME_CHARS) + .collect(); + let mut seen_connector_ids = HashSet::new(); + let connector_ids: Vec = entry + .connector_ids + .iter() + .filter(|id| seen_connector_ids.insert(id.as_str())) + .take(MAX_CONNECTOR_IDS_PER_PLUGIN) + .cloned() + .collect(); + let installations = entry + .locations + .iter() + .filter_map(|location| match location { + PluginSourceLocation::Executor { + environment_id, + plugin_id, + root, + } => Some((environment_id, plugin_id, root)), + PluginSourceLocation::Cloud { .. } => None, + }); + let mut projections = Vec::new(); + for (index, (environment_id, plugin_id, root)) in installations.enumerate() { + let root_id = format!("{environment_id}:{plugin_id}"); + // Only the first retained installation supplies MCPs; the others retain package attribution. + let servers = if index == 0 { + let parsed = normalize_executor_plugin_mcp_declarations( + root, + entry.mcp_servers.clone(), + environment_id, + ); + servers_from_outcome(&root_id, parsed) + } else { + Vec::new() + }; + let metadata = SelectedPluginMetadata { + plugin_display_name: plugin_display_name.clone(), + servers, + connector_ids: connector_ids.clone(), + }; + projections.push((root_id, metadata)); + } + if entry + .locations + .iter() + .any(|location| matches!(location, PluginSourceLocation::Cloud { .. })) + { + projections.push(( + format!("cloud:{}", entry.id.as_str()), + SelectedPluginMetadata { + plugin_display_name, + servers: Vec::new(), + connector_ids, + }, + )); + } + projections +} + fn parse_mcp_servers( selected_root: &SelectedCapabilityRoot, plugin_root: &codex_utils_path_uri::PathUri, @@ -151,9 +224,16 @@ fn parse_mcp_servers( return Vec::new(); } }; + servers_from_outcome(&selected_root.id, parsed) +} + +fn servers_from_outcome( + selected_root_id: &str, + parsed: PluginMcpConfigParseOutcome, +) -> Vec<(String, codex_config::McpServerConfig)> { for error in parsed.errors { tracing::warn!( - selected_root = selected_root.id, + selected_root = selected_root_id, server = error.name, error = error.message, "ignoring invalid exec-server-discovered MCP server" @@ -161,3 +241,7 @@ fn parse_mcp_servers( } parsed.servers.into_iter().collect() } + +#[cfg(test)] +#[path = "discovery_tests.rs"] +mod tests; diff --git a/codex-rs/ext/mcp/src/plugin/discovery_tests.rs b/codex-rs/ext/mcp/src/plugin/discovery_tests.rs new file mode 100644 index 0000000000..1e4af55939 --- /dev/null +++ b/codex-rs/ext/mcp/src/plugin/discovery_tests.rs @@ -0,0 +1,134 @@ +//! Verifies merged MCPs use the first retained installation while preserving all package identities. + +use super::*; +use codex_config::McpServerTransportConfig; +use codex_config::types::PluginsConfigToml; +use codex_core::config::ConfigBuilder; +use codex_core_plugins::PluginIdentity; +use codex_extension_api::SelectedPluginContribution; +use codex_utils_path_uri::PathUri; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn merged_mcp_declarations_require_an_executor_location() -> anyhow::Result<()> { + let codex_home = tempfile::tempdir()?; + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(codex_home.path().to_path_buf())) + .cli_overrides(vec![("features.plugins".to_string(), true.into())]) + .build() + .await?; + let mut entry = PluginCatalogEntry { + enabled: true, + id: PluginIdentity::Remote { + remote_plugin_id: "plugin_example".to_string(), + }, + display_name: "Example".to_string(), + version: "1".to_string(), + mcp_servers: std::collections::BTreeMap::from([( + "example".to_string(), + serde_json::json!({"command":"node","args":["server.js"],"cwd":"mcp"}), + )]), + connector_ids: vec!["connector_example".to_string()], + locations: vec![PluginSourceLocation::Cloud { + resource_uri: "plugin://plugin_example".to_string(), + bundle_uri: None, + }], + }; + let projections = v2_metadata_from_catalog(&entry); + let [(root_id, metadata)] = projections.as_slice() else { + panic!("one cloud projection expected"); + }; + assert_eq!(root_id, "cloud:plugin_example"); + assert!(metadata.servers.is_empty()); + assert_eq!(metadata.connector_ids, vec!["connector_example"]); + + // Attribution keeps the installation key; server configs distinguish the executors. + let installations = [ + ("executor-b", "example@acme", "/other/example/1"), + ("executor-a", "example@acme", "/plugins/example/1"), + ]; + // Retained catalog order, not lexical environment order, chooses the executor. + for (environment_id, plugin_id, root) in installations { + entry.locations.push(PluginSourceLocation::Executor { + environment_id: environment_id.to_string(), + plugin_id: plugin_id.to_string(), + root: PathUri::parse(&format!("file://{root}"))?, + }); + } + let projections = v2_metadata_from_catalog(&entry); + assert_eq!(projections.len(), installations.len() + 1); + for (index, ((root_id, metadata), (environment_id, plugin_id, root))) in + projections.iter().zip(installations).enumerate() + { + assert_eq!(root_id, &format!("{environment_id}:{plugin_id}")); + let contribution = super::super::project_metadata( + &config, + &PluginsConfigToml::default(), + plugin_id, + SelectedPluginContribution { + plugin_display_name: metadata.plugin_display_name.clone(), + source_environment_id: environment_id.to_string(), + servers: metadata.servers.clone(), + connector_ids: metadata.connector_ids.clone(), + }, + ); + assert_eq!(contribution.connector_ids, entry.connector_ids); + if index != 0 { + assert!(contribution.servers.is_empty()); + continue; + } + let [(name, mcp_config)] = contribution.servers.as_slice() else { + panic!("one MCP contribution expected"); + }; + assert_eq!(mcp_config.environment_id, environment_id); + assert_eq!(name, "example"); + let McpServerTransportConfig::Stdio { command, cwd, .. } = &mcp_config.transport else { + panic!("stdio expected"); + }; + assert_eq!(command, "node"); + assert_eq!( + cwd.as_ref() + .map(codex_utils_path_uri::LegacyAppPathString::as_str), + Some(format!("{root}/mcp").as_str()) + ); + } + + let (root_id, metadata) = &projections[installations.len()]; + assert_eq!(root_id, "cloud:plugin_example"); + assert!(metadata.servers.is_empty()); + assert_eq!(metadata.connector_ids, vec!["connector_example"]); + + // Bound projected metadata without mutating the source catalog or losing installations. + entry.display_name = "界".repeat(MAX_PLUGIN_DISPLAY_NAME_CHARS + 1); + entry.connector_ids = (0..=MAX_CONNECTOR_IDS_PER_PLUGIN) + .flat_map(|index| vec![format!("connector_{index}"); 2]) + .collect(); + let expected_connectors = (0..MAX_CONNECTOR_IDS_PER_PLUGIN) + .map(|index| format!("connector_{index}")) + .collect::>(); + entry.mcp_servers.clear(); + let projections = v2_metadata_from_catalog(&entry); + assert_eq!(projections.len(), installations.len() + 1); + assert!( + projections + .iter() + .all(|(_, metadata)| metadata.servers.is_empty()) + ); + for (_, metadata) in projections { + assert_eq!( + metadata.plugin_display_name, + "界".repeat(MAX_PLUGIN_DISPLAY_NAME_CHARS) + ); + assert_eq!(metadata.connector_ids, expected_connectors); + } + assert_eq!( + entry.display_name.chars().count(), + MAX_PLUGIN_DISPLAY_NAME_CHARS + 1 + ); + assert_eq!( + entry.connector_ids.len(), + (MAX_CONNECTOR_IDS_PER_PLUGIN + 1) * 2 + ); + Ok(()) +} diff --git a/codex-rs/ext/mcp/src/plugin_contributor_state.rs b/codex-rs/ext/mcp/src/plugin_contributor_state.rs index 3086151b48..c14802037d 100644 --- a/codex-rs/ext/mcp/src/plugin_contributor_state.rs +++ b/codex-rs/ext/mcp/src/plugin_contributor_state.rs @@ -2,12 +2,15 @@ use codex_core_plugins::PluginCatalog; use codex_core_plugins::PluginRootOwnership; +use codex_core_plugins::catalog_from_snapshots; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::FileSystemSandboxContext; use codex_extension_api::SelectedPluginContribution; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpResourceClient; use codex_mcp::McpResourceClientAuthKey; use codex_protocol::capabilities::SelectedCapabilityRoot; +use std::collections::HashMap; use std::sync::Mutex; use std::sync::MutexGuard; @@ -35,6 +38,8 @@ impl PluginsThreadState { pub(crate) struct PluginContributorState { pub(crate) plugins_enabled: bool, pub(crate) cloud_generation: Option, + pub(crate) executor_catalog: Option, + executor_snapshots: Vec, pub(crate) executor_cache: Vec, } @@ -67,8 +72,36 @@ impl CloudPluginGeneration { } impl PluginContributorState { + /// Reuses unchanged discovery and preserves surviving installations' order on refresh. + pub(crate) fn update_executor_catalog( + &mut self, + snapshots: &[ExecutorCapabilityDiscoveryV2Snapshot], + ) { + if self.executor_catalog.is_some() && self.executor_snapshots == snapshots { + return; + } + let mut catalog = catalog_from_snapshots(snapshots); + if let Some(previous) = &self.executor_catalog { + // Preserve installation order to keep MCPs on the same executor while it remains available. + let mut previous_ranks = HashMap::with_capacity(previous.entries.len()); + for (rank, entry) in previous.entries.iter().enumerate() { + previous_ranks + .entry((&entry.id, entry.locations.as_slice())) + .or_insert(rank); + } + catalog.entries.sort_by_key(|entry| { + previous_ranks + .get(&(&entry.id, entry.locations.as_slice())) + .copied() + .unwrap_or(usize::MAX) + }); + } + self.executor_catalog = Some(catalog); + self.executor_snapshots = snapshots.to_vec(); + } + /// Returns the cloud catalog only while it belongs to the live auth generation. - fn cloud_catalog(&self) -> Option<&PluginCatalog> { + pub(crate) fn cloud_catalog(&self) -> Option<&PluginCatalog> { self.cloud_generation .as_ref() .filter(|generation| generation.is_current()) @@ -105,6 +138,14 @@ impl PluginContributorState { else { return false; }; + if catalog.is_partial + && generation + .catalog + .as_ref() + .is_some_and(|previous| !previous.is_partial) + { + return true; + } generation.catalog = Some(catalog); true } @@ -122,3 +163,15 @@ pub(crate) enum CachedPluginMetadata { Unloaded, Loaded(Option), } + +/// Bounded declarations projected from merged plugin metadata. +#[derive(Clone, Default)] +pub(crate) struct SelectedPluginMetadata { + pub(crate) plugin_display_name: String, + pub(crate) servers: Vec<(String, codex_config::McpServerConfig)>, + pub(crate) connector_ids: Vec, +} + +#[cfg(test)] +#[path = "plugin_contributor_state_tests.rs"] +mod tests; diff --git a/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs b/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs new file mode 100644 index 0000000000..ed4ed918c9 --- /dev/null +++ b/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs @@ -0,0 +1,463 @@ +//! Tests thread-owned discovery caching, installation ordering, and contribution projection. + +use super::*; +use crate::PluginProviders; +use crate::plugin_contributor::PluginContributor; +use codex_core::config::ConfigBuilder; +use codex_core_plugins::ExecutorPluginProvider; +use codex_core_plugins::PluginCatalog; +use codex_core_plugins::PluginCatalogEntry; +use codex_core_plugins::PluginIdentity; +use codex_core_plugins::PluginProvider; +use codex_core_plugins::PluginSourceLocation; +use codex_exec_server::CapabilityTextFile; +use codex_exec_server::DiscoverV2CapabilitiesResponse; +use codex_exec_server::DiscoveredPluginFiles; +use codex_exec_server::EnvironmentManager; +use codex_exec_server::ExecutorCapabilityDiscoveryCache; +use codex_exec_server::ExecutorPlugin; +use codex_exec_server::LOCAL_ENVIRONMENT_ID; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionDataInit; +use codex_extension_api::McpServerContribution; +use codex_extension_api::McpServerContributionContext; +use codex_extension_api::McpServerContributor; +use codex_features::Feature; +use codex_protocol::capabilities::CapabilityRootLocation; +use codex_utils_path_uri::PathUri; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use std::sync::Arc; + +async fn contribution_labels( + contributor: &PluginContributor, + context: McpServerContributionContext<'_, codex_core::config::Config>, +) -> Vec { + let mut labels = Vec::new(); + let policy = codex_config::types::PluginsConfigToml::default(); + for plugin in contributor.selected_plugins(context, &policy).await { + if !plugin.enabled { + labels.push(format!("disabled:{}", plugin.plugin_id)); + continue; + } + let contribution = plugin.mcp.await; + labels.extend( + contribution + .servers + .into_iter() + .map(|(name, _)| format!("mcp:{name}")), + ); + labels.push(plugin.selected_root_id); + } + for contribution in contributor.contribute(context).await { + if let McpServerContribution::HostedPluginConnectors { plugin_id, .. } = contribution { + labels.push(format!("cloud:{plugin_id}")); + } else { + panic!("unexpected contribution: {contribution:?}"); + } + } + labels +} + +fn cloud_plugin(remote_id: &str) -> PluginCatalogEntry { + PluginCatalogEntry { + enabled: true, + id: PluginIdentity::Remote { + remote_plugin_id: remote_id.to_string(), + }, + display_name: "Example".to_string(), + version: "1.0.0".to_string(), + mcp_servers: BTreeMap::from([( + "example".to_string(), + serde_json::json!({"command":"node","args":["server.js"]}), + )]), + connector_ids: vec!["connector_example".to_string()], + locations: vec![PluginSourceLocation::Cloud { + resource_uri: format!("plugin://{remote_id}"), + bundle_uri: None, + }], + } +} + +fn executor_snapshot( + environment_id: &str, +) -> anyhow::Result { + let root = PathUri::parse("file:///plugins/example/1.0.0")?; + Ok(ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: environment_id.to_string(), + result: Ok(Arc::new(DiscoverV2CapabilitiesResponse { + plugins: vec![ExecutorPlugin { + id: "example@marketplace".to_string(), + display_name: "Example".to_string(), + remote_plugin_id: Some("plugin_a".to_string()), + version: "1.0.0".to_string(), + files: DiscoveredPluginFiles { + manifest: CapabilityTextFile { + path: root.join(".codex-plugin/plugin.json")?, + contents: r#"{"name":"example"}"#.to_string(), + }, + mcp_config: None, + apps_config: None, + }, + root, + skills: Vec::new(), + }], + ..Default::default() + })), + }) +} + +#[test] +fn executor_refresh_preserves_order_and_removes_stale_entries() -> anyhow::Result<()> { + let mut state = PluginContributorState::default(); + // Reuse equal snapshots, including equal responses allocated by a later discovery. + let b = executor_snapshot("executor-b")?; + let c = executor_snapshot("executor-c")?; + state.update_executor_catalog(&[b.clone(), c.clone()]); + state.update_executor_catalog(&[executor_snapshot("executor-b")?, c.clone()]); + assert!(Arc::ptr_eq( + state.executor_snapshots[0].result.as_ref().unwrap(), + b.result.as_ref().unwrap(), + )); + + // Keep B ahead of C even when discovery reverses them and adds an earlier-named A. + let a = executor_snapshot("executor-a")?; + let mut updated_b = b; + Arc::make_mut(updated_b.result.as_mut().unwrap()).plugins[0].display_name = + "Updated".to_string(); + state.update_executor_catalog(&[a.clone(), c.clone(), updated_b.clone()]); + assert_eq!( + state.executor_catalog.as_ref().unwrap(), + &catalog_from_snapshots(&[updated_b.clone(), c.clone(), a.clone()]), + ); + // Removing B promotes C; B returning later appends it rather than displacing C. + state.update_executor_catalog(&[a.clone(), c.clone()]); + state.update_executor_catalog(&[updated_b.clone(), a.clone(), c.clone()]); + assert_eq!( + state.executor_catalog.as_ref().unwrap(), + &catalog_from_snapshots(&[c.clone(), a.clone(), updated_b]), + ); + // A failed scan removes that environment's entries, and recovery rebuilds them. + let failed = ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: c.environment_id.clone(), + result: Err("discovery failed".to_string()), + }; + state.update_executor_catalog(&[failed.clone(), a.clone()]); + assert_eq!( + state.executor_catalog.as_ref().unwrap(), + &catalog_from_snapshots(&[failed, a.clone()]), + ); + state.update_executor_catalog(&[c.clone(), a.clone()]); + assert_eq!( + state.executor_catalog.as_ref().unwrap(), + &catalog_from_snapshots(&[a, c]), + ); + Ok(()) +} + +#[tokio::test] +async fn projection_respects_discovery_mode_and_preserves_enabled_plugins() -> anyhow::Result<()> { + struct CloudProvider; + impl PluginProvider for CloudProvider {} + + let home = tempfile::tempdir()?; + let root = home.path().join("legacy"); + std::fs::create_dir_all(root.join(".codex-plugin"))?; + std::fs::write( + root.join(".codex-plugin/plugin.json"), + r#"{"name":"legacy","mcpServers":{"legacy":{"command":"legacy-command"}}}"#, + )?; + let mut config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .fallback_cwd(Some(home.path().to_path_buf())) + .build() + .await?; + config.features.enable(Feature::Plugins)?; + let environments = Arc::new(EnvironmentManager::default_for_tests()); + let roots = vec![SelectedCapabilityRoot { + id: "legacy-root".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: LOCAL_ENVIRONMENT_ID.to_string(), + path: PathUri::from_host_native_path(&root)?, + }, + }]; + let v1 = ExecutorCapabilityDiscoveryCache::new(Arc::clone(&environments)) + .snapshot(&roots, &Default::default()) + .await; + let mut contributor = PluginContributor { + providers: PluginProviders::new(Arc::new(ExecutorPluginProvider::new(environments))) + .with_cloud_provider(Arc::new(CloudProvider)), + }; + let thread = ExtensionData::new("projection-test"); + let init = ExtensionDataInit::default(); + let state = thread.get_or_init(PluginsThreadState::default); + { + let mut state = state.contributor_state(); + let key = state.prepare_cloud_generation(/*mcp_resources*/ None); + assert!(state.publish_cloud_catalog( + key.as_ref(), + PluginCatalog { + entries: vec![cloud_plugin("plugin_a")], + ..Default::default() + } + )); + } + // A different environment ensures per-root filtering cannot hide a legacy fallback. + let mut valid = [executor_snapshot("executor-a")?]; + Arc::make_mut(valid[0].result.as_mut().unwrap()).plugins[0] + .files + .mcp_config = Some(CapabilityTextFile { + path: PathUri::parse("file:///plugins/example/1.0.0/.mcp.json")?, + contents: r#"{"executor-only":{"command":"node"}}"#.to_string(), + }); + let mut local = [ + executor_snapshot("executor-a")?, + executor_snapshot("executor-b")?, + ]; + for snapshot in &mut local { + Arc::make_mut(snapshot.result.as_mut().unwrap()).plugins[0].remote_plugin_id = None; + } + let empty = [ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: "executor-a".to_string(), + result: Ok(Arc::new(DiscoverV2CapabilitiesResponse::default())), + }]; + let failed = [ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: "executor-a".to_string(), + result: Err("discovery failed".to_string()), + }]; + for (case, v2_enabled, snapshots, expected) in [ + ( + "snapshot selects V2 independently of the acquisition flag", + false, + Some(valid.as_slice()), + vec![ + "mcp:example", + "executor-a:example@marketplace", + "cloud:plugin_a", + ], + ), + ( + "legacy", + false, + None, + vec!["mcp:legacy", "legacy-root", "cloud:plugin_a"], + ), + ( + "no V2 snapshot", + true, + None, + vec!["mcp:legacy", "legacy-root", "cloud:plugin_a"], + ), + ( + "empty snapshots", + true, + Some([].as_slice()), + vec!["cloud:plugin_a"], + ), + ( + "empty inventory", + true, + Some(empty.as_slice()), + vec!["cloud:plugin_a"], + ), + ( + "failed", + true, + Some(failed.as_slice()), + vec!["cloud:plugin_a"], + ), + ( + "valid", + true, + Some(valid.as_slice()), + vec![ + "mcp:example", + "executor-a:example@marketplace", + "cloud:plugin_a", + ], + ), + ( + "unconfigured local", + true, + Some(local.as_slice()), + vec![ + "disabled:example@marketplace", + "disabled:example@marketplace", + "cloud:plugin_a", + ], + ), + ] { + config + .features + .set_enabled(Feature::ExecutorCapabilityDiscoveryV2, v2_enabled)?; + for legacy_snapshot in [None, Some(&v1)] { + state.contributor_state().update_executor_catalog(&valid); + let contributions = contribution_labels( + &contributor, + McpServerContributionContext::for_step( + &config, + &init, + &thread, + "test", + &roots, + legacy_snapshot, + snapshots, + ), + ) + .await; + assert_eq!( + contributions, + expected, + "{case}, V2={v2_enabled}, V1={}", + legacy_snapshot.is_some() + ); + let expected_catalog = catalog_from_snapshots(snapshots.unwrap_or_default()); + assert_eq!( + state.contributor_state().executor_catalog.as_ref(), + Some(&expected_catalog), + "{case}" + ); + } + } + std::fs::write( + home.path().join("config.toml"), + "[plugins.\"example@marketplace\"]\nenabled = true\n", + )?; + let mut config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .fallback_cwd(Some(home.path().to_path_buf())) + .build() + .await?; + config.features.enable(Feature::Plugins)?; + config + .features + .enable(Feature::ExecutorCapabilityDiscoveryV2)?; + + // Cloud counterparts control enablement without deleting the executor installation. + let mut disabled_cloud = cloud_plugin("plugin_a"); + disabled_cloud.enabled = false; + for (case, cloud_enabled, catalog, enabled) in [ + ("no cloud provider", false, None, true), + ("no cloud catalog", true, None, true), + ("absent", true, Some(PluginCatalog::default()), false), + ( + "partial absent", + true, + Some(PluginCatalog { + is_partial: true, + warnings: vec!["partial cloud listing".to_string()], + ..Default::default() + }), + false, + ), + ( + "disabled", + true, + Some(PluginCatalog { + entries: vec![disabled_cloud], + ..Default::default() + }), + false, + ), + ( + "re-enabled", + true, + Some(PluginCatalog { + entries: vec![cloud_plugin("plugin_a")], + ..Default::default() + }), + true, + ), + ( + "partial counterpart", + true, + Some(PluginCatalog { + entries: vec![cloud_plugin("plugin_a")], + is_partial: true, + warnings: vec!["partial cloud listing".to_string()], + }), + true, + ), + ] { + let cloud_available = catalog.is_some(); + contributor.providers.cloud = if cloud_enabled { + Some(Arc::new(CloudProvider)) + } else { + None + }; + { + let mut cache = state.contributor_state(); + cache.cloud_generation = None; + let key = cache.prepare_cloud_generation(/*mcp_resources*/ None); + if let Some(catalog) = catalog { + assert!(cache.publish_cloud_catalog(key.as_ref(), catalog)); + } + } + let contributions = contribution_labels( + &contributor, + McpServerContributionContext::for_step( + &config, + &init, + &thread, + case, + &roots, + /*executor_capability_discovery*/ None, + Some(&valid), + ), + ) + .await; + let expected = if !cloud_enabled || !cloud_available { + vec!["mcp:executor-only", "executor-a:example@marketplace"] + } else if enabled { + vec![ + "mcp:example", + "executor-a:example@marketplace", + "cloud:plugin_a", + ] + } else { + vec!["disabled:example@marketplace"] + }; + assert_eq!(contributions, expected, "{case}"); + assert_eq!( + state.contributor_state().executor_catalog, + Some(catalog_from_snapshots(&valid)), + "{case}" + ); + } + + // A full Cloud catalog must not crowd out enabled executor-local plugins. + let cloud_plugin_count = 100; + { + let mut state = state.contributor_state(); + let key = state.prepare_cloud_generation(/*mcp_resources*/ None); + let entries = (0..cloud_plugin_count) + .map(|index| cloud_plugin(&format!("plugin_{index}"))) + .collect::>(); + assert!(state.publish_cloud_catalog( + key.as_ref(), + PluginCatalog { + entries, + ..Default::default() + } + )); + } + let contributions = contribution_labels( + &contributor, + McpServerContributionContext::for_step( + &config, + &init, + &thread, + "merged", + &roots, + /*executor_capability_discovery*/ None, + Some(&local[..1]), + ), + ) + .await; + assert_eq!(contributions.len(), cloud_plugin_count + 1); + assert_eq!(contributions[0], "executor-a:example@marketplace"); + for (index, contribution) in contributions[1..].iter().enumerate() { + assert_eq!(contribution, &format!("cloud:plugin_{index}")); + } + Ok(()) +} diff --git a/codex-rs/ext/mcp/tests/plugin_mcp.rs b/codex-rs/ext/mcp/tests/plugin_mcp.rs index 247abdf9fd..9123fe43eb 100644 --- a/codex-rs/ext/mcp/tests/plugin_mcp.rs +++ b/codex-rs/ext/mcp/tests/plugin_mcp.rs @@ -662,10 +662,6 @@ async fn raw_selected_plugin_contributions( None }; - let v2 = [codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot { - environment_id: LOCAL_ENVIRONMENT_ID.to_string(), - result: Err("V2 must not change existing contributions".to_string()), - }]; let selected = registry.mcp_server_contributors()[0] .selected_plugins( McpServerContributionContext::for_step( @@ -675,7 +671,7 @@ async fn raw_selected_plugin_contributions( "test_originator", &selected_capability_roots, executor_capability_discovery.as_ref(), - Some(&v2), + /*executor_capability_discovery_v2*/ None, ), &config.plugins, ) @@ -752,19 +748,20 @@ async fn run_cloud_plugin_projection_scenarios() -> anyhow::Result<()> { } else if cloud_available { Ok(PluginCatalog { entries: vec![PluginCatalogEntry { + enabled: true, id: PluginIdentity::Remote { remote_plugin_id: "plugin_notes".into(), }, display_name: "Cloud Notes".into(), - version: Some("1".into()), - mcp_servers: [("notes".into(), declaration.to_string())].into(), + version: "1".into(), + mcp_servers: [("notes".into(), declaration.clone())].into(), connector_ids: vec!["calendar".into()], locations: vec![PluginSourceLocation::Cloud { resource_uri: "plugin://plugin_notes".into(), bundle_uri: None, }], }], - warnings: Vec::new(), + ..Default::default() }) } else { Err("plugin-service unavailable".into()) diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index d2bab6dc63..632e09e6d4 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -78,7 +78,7 @@ pub(crate) fn attribute_executor_plugins( .iter() .find(|plugin| plugin.selected_root_id.as_ref() == Some(&skill.authority.id)) { - skill.plugin_id = Some(plugin.plugin_id.clone()); + skill.plugin_id = Some(plugin.plugin_id.as_str().to_string()); skill.analytics_scope = Some(SkillScope::User); } }