Reconcile V2 executor plugins with cloud catalogs (#55301)

## What changed - Use V2 discovery inventories for plugin MCP contributions when supplied, including empty or failed inventories; retain legacy discovery when no V2 snapshot is supplied. - Merge plugins by canonical identity, using cloud metadata and enablement for remote plugins and attaching executor installations only when versions match. Preserve configured executor-local plugins and executor enablement when cloud data is unavailable. - Keep installation order stable across refreshes and run merged MCP declarations on the first retained executor installation while preserving attribution for the others. - Carry local and remote identities through selected-plugin snapshots, skill attribution, and active-plugin telemetry. Track disabled plugin IDs alongside disabled roots. - Preserve a complete cloud catalog when a partial refresh arrives, and cap tool-description plugin attribution at 1,024 bytes while retaining all plugin associations. ## Testing Add coverage for catalog parsing and reconciliation, enablement and discovery-mode selection, installation refresh ordering, executor-specific MCP normalization, selected-plugin snapshots, and bounded UTF-8 attribution text. GitOrigin-RevId: a8ad352de6a2b8dc4bfd1ff6a5c62ee8ed1d1664
2026-10-20 17:38:45 +00:00 · 2026-10-19 15:15:17 +00:00
parent 1277bcfddc
commit f7a135ec61
32 changed files with 1804 additions and 90 deletions
@@ -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<PluginCatalogEntry>,
    pub warnings: Vec<String>,
+    /// 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<String>,
-    /// Serialized `.mcp.json` server declarations keyed by their authored names.
-    pub mcp_servers: BTreeMap<String, String>,
+    pub version: String,
+    /// Raw MCP server declarations keyed by their authored names.
+    pub mcp_servers: BTreeMap<String, Value>,
    pub connector_ids: Vec<String>,
    pub locations: Vec<PluginSourceLocation>,
 }

+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<String, PluginConfig>) -> 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;
@@ -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<PluginCatalogEntry> {
+    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(())
+}
@@ -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<String>) {
+    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;
@@ -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(())
+}
@@ -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;
@@ -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<String, PluginConfig> {
@@ -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()) })
    }