Resolve MCP server registrations through a catalog (#27634)

## Why

MCP servers currently come from user config, local plugins,
compatibility Apps synthesis, and host extensions. Those sources were
composed by mutating a shared map, leaving registration identity,
precedence, removal, and provenance implicit in assembly order.

Before adding executor-owned MCPs, Codex needs one durable resolution
boundary above `McpConnectionManager`. This PR introduces that boundary
while preserving current server configuration, policy, and runtime
behavior. Executor-scoped registrations and explicit policy layers
remain follow-ups.

## What changed

- Add typed `McpServerRegistration` inputs and an immutable
`ResolvedMcpCatalog` in `codex-mcp`.
- Retain each registration's complete `McpServerConfig`, including its
environment binding, while recording its source and provenance.
- Preserve the existing structural precedence between plugin, config,
compatibility, and ordered extension sources.
- Resolve equal-precedence actions by contribution order; provenance IDs
are used only for diagnostics and cannot affect the winner.
- Preserve extension removals and the existing name-scoped `enabled =
false` veto.
- Report same-tier conflicts with every contender and the final catalog
outcome, including whether the winning action registers or removes the
server.
- Require MCP contributors to provide a stable diagnostic identity.
- Derive materialized server maps and plugin ownership from the resolved
catalog.

`McpConnectionManager`, transport startup, tool calls, and resource
routing continue to consume the same effective `McpServerConfig` values.

## Scope

This PR does not add new MCP capabilities or change user-visible
behavior. It does not add executor plugin discovery, thread-scoped
registrations, dynamic refresh generations, or new user/managed policy
semantics.

## Verification

- Added focused catalog coverage for source precedence, complete
configuration preservation, disabled vetoes, plugin ownership,
contribution-order tie breaking, removal outcomes, and conflict
diagnostics.
- Extended hosted Apps coverage for ordered extension removal and
Apps-disabled hosts with and without the hosted extension installed.
- `cargo check -p codex-mcp --tests -p codex-extension-api -p
codex-core`

codex-rs/core/src/mcp.rs

@@ -10,11 +10,12 @@ use codex_login::CodexAuth;
 use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
 use codex_mcp::EffectiveMcpServer;
 use codex_mcp::McpConfig;
-use codex_mcp::ToolPluginProvenance;
+use codex_mcp::McpServerRegistration;
 use codex_mcp::codex_apps_mcp_server_config;
 use codex_mcp::configured_mcp_servers;
 use codex_mcp::effective_mcp_servers;
-use codex_mcp::tool_plugin_provenance as collect_tool_plugin_provenance;
+
+const LEGACY_CODEX_APPS_REGISTRATION_ID: &str = "legacy_codex_apps";
 
 #[derive(Clone)]
 pub struct McpManager {
@@ -45,32 +46,53 @@ impl McpManager {
     /// runtime-only extension overlays.
     pub async fn runtime_config(&self, config: &Config) -> McpConfig {
         let mut mcp_config = config.to_mcp_config(self.plugins_manager.as_ref()).await;
-        let disabled_server_names = mcp_config
-            .configured_mcp_servers
-            .iter()
-            .filter(|(_, server)| !server.enabled)
-            .map(|(name, _)| name.clone())
-            .collect::<Vec<_>>();
+        let mut catalog = mcp_config.mcp_server_catalog.to_builder();
         if mcp_config.apps_enabled {
-            mcp_config.configured_mcp_servers.insert(
+            catalog.register(McpServerRegistration::from_compatibility(
                 CODEX_APPS_MCP_SERVER_NAME.to_string(),
+                LEGACY_CODEX_APPS_REGISTRATION_ID,
                 codex_apps_mcp_server_config(
                     &mcp_config.chatgpt_base_url,
                     mcp_config.apps_mcp_product_sku.as_deref(),
                 ),
-            );
+            ));
         } else {
-            mcp_config
-                .configured_mcp_servers
-                .remove(CODEX_APPS_MCP_SERVER_NAME);
+            catalog.remove_compatibility(
+                CODEX_APPS_MCP_SERVER_NAME.to_string(),
+                LEGACY_CODEX_APPS_REGISTRATION_ID,
+            );
         }
-        let contributions = self.contributions(config).await;
-        Self::apply_to_configured_servers(&contributions, &mut mcp_config.configured_mcp_servers);
-        for name in disabled_server_names {
-            if let Some(server) = mcp_config.configured_mcp_servers.get_mut(&name) {
-                server.enabled = false;
+
+        let mut contribution_order = 0;
+        for contributor in self.extensions.mcp_server_contributors() {
+            for contribution in contributor.contribute(config).await {
+                match contribution {
+                    McpServerContribution::Set {
+                        name,
+                        config: server_config,
+                    } => catalog.register(McpServerRegistration::from_extension(
+                        name,
+                        contributor.id(),
+                        contribution_order,
+                        *server_config,
+                    )),
+                    McpServerContribution::Remove { name } => {
+                        catalog.remove_extension(name, contributor.id(), contribution_order)
+                    }
+                }
+                contribution_order += 1;
             }
         }
+        let catalog = catalog.build();
+        for conflict in catalog.conflicts() {
+            tracing::warn!(
+                server = conflict.name,
+                outcome = ?conflict.outcome,
+                contenders = ?conflict.contenders,
+                "conflicting MCP server actions; using resolved catalog outcome"
+            );
+        }
+        mcp_config.mcp_server_catalog = catalog;
         mcp_config
     }
 
@@ -95,34 +117,4 @@ impl McpManager {
         let mcp_config = self.runtime_config(config).await;
         effective_mcp_servers(&mcp_config, auth)
     }
-
-    /// Returns provenance for plugin-owned servers in the configured view.
-    pub async fn tool_plugin_provenance(&self, config: &Config) -> ToolPluginProvenance {
-        let mcp_config = config.to_mcp_config(self.plugins_manager.as_ref()).await;
-        collect_tool_plugin_provenance(&mcp_config)
-    }
-
-    async fn contributions(&self, config: &Config) -> Vec<McpServerContribution> {
-        let mut contributions = Vec::new();
-        for contributor in self.extensions.mcp_server_contributors() {
-            contributions.extend(contributor.contribute(config).await);
-        }
-        contributions
-    }
-
-    fn apply_to_configured_servers(
-        contributions: &[McpServerContribution],
-        servers: &mut HashMap<String, McpServerConfig>,
-    ) {
-        for contribution in contributions {
-            match contribution {
-                McpServerContribution::Set { name, config } => {
-                    servers.insert(name.clone(), config.as_ref().clone());
-                }
-                McpServerContribution::Remove { name } => {
-                    servers.remove(name);
-                }
-            }
-        }
-    }
 }