diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 18e14757a6..7c5928cc26 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -13,6 +13,7 @@ use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; +use codex_extension_api::SelectedPlugin; use codex_extension_api::SelectedPluginIdentity; use codex_extension_api::SelectedPluginSnapshot; use codex_features::Feature; @@ -189,7 +190,6 @@ impl McpManager { disabled_plugin_ids: &[String], ) -> McpRuntimeProjection { let config = context.config(); - let mut selected_plugin_available = false; let mut selected_plugin_connector_sources = Vec::new(); let mut selected_plugin_registrations = Vec::new(); let mut selected_plugins = Vec::new(); @@ -198,7 +198,57 @@ impl McpManager { // A contributor can emit multiple ordered actions, so order each action globally rather // than enumerating contributors. let mut contribution_order = 0; + let mut plugin_selection_order = 0; for contributor in self.extensions.mcp_server_contributors() { + // Finish executor plugins before asking for hosted contributions, which may change + // accounts while an executor is still loading. + for SelectedPlugin { + selected_root_id, + plugin_id, + mcp, + } in contributor.selected_plugins(context).await + { + let selection_order = plugin_selection_order; + plugin_selection_order += 1; + let disabled = disabled_plugin_ids.contains(&plugin_id); + if !config.features.enabled(Feature::Plugins) || disabled { + disabled_plugin_roots.push(selected_root_id); + } else { + selected_plugins.push(SelectedPluginIdentity { + selected_root_id: Some(selected_root_id), + plugin_id: plugin_id.clone(), + }); + } + let contribution = mcp.await; + if !disabled { + for (name, server) in contribution.servers { + selected_plugin_registrations.push( + McpServerRegistration::from_selected_plugin( + name, + McpPluginAttribution::new( + plugin_id.clone(), + contribution.plugin_display_name.clone(), + ), + selection_order, + server, + ), + ); + } + } + // Disabled plugins still tell Apps which connectors to hide. + if config.features.enabled(Feature::Plugins) + && !contribution.connector_ids.is_empty() + { + selected_plugin_connector_sources.push( + PluginConnectorSource::from_connector_ids( + plugin_id, + contribution.plugin_display_name, + contribution.connector_ids.into_iter().map(AppConnectorId), + ), + ); + } + } + for contribution in contributor.contribute(context).await { match contribution { McpServerContribution::Set { name, config } => { @@ -240,50 +290,27 @@ impl McpManager { } overlays.push(OrderedMcpOverlay::Set(Box::new(registration))); } - McpServerContribution::SelectedPlugin { ref plugin_id, .. } - if disabled_plugin_ids.contains(plugin_id) => {} - McpServerContribution::SelectedPlugin { - name, - plugin_id, - plugin_display_name, - selection_order, - config, - } => selected_plugin_registrations.push( - McpServerRegistration::from_selected_plugin( - name, - McpPluginAttribution::new(plugin_id, plugin_display_name), - selection_order, - *config, - ), - ), - McpServerContribution::SelectedPluginPackage { - selected_root_id, .. - } if !config.features.enabled(Feature::Plugins) => { - disabled_plugin_roots.push(selected_root_id); - } - McpServerContribution::SelectedPluginPackage { - selected_root_id, + McpServerContribution::HostedPluginConnectors { plugin_id, plugin_display_name, connector_ids, } => { - if disabled_plugin_ids.contains(&plugin_id) { - disabled_plugin_roots.push(selected_root_id); - } else { - selected_plugin_available = true; - selected_plugins.push(SelectedPluginIdentity { - selected_root_id, - plugin_id: plugin_id.clone(), - }); - } - if !connector_ids.is_empty() { - selected_plugin_connector_sources.push( - PluginConnectorSource::from_connector_ids( - plugin_id, - plugin_display_name, - connector_ids.into_iter().map(AppConnectorId), - ), - ); + if config.features.enabled(Feature::Plugins) { + if !disabled_plugin_ids.contains(&plugin_id) { + selected_plugins.push(SelectedPluginIdentity { + selected_root_id: None, + plugin_id: plugin_id.clone(), + }); + } + if !connector_ids.is_empty() { + selected_plugin_connector_sources.push( + PluginConnectorSource::from_connector_ids( + plugin_id, + plugin_display_name, + connector_ids.into_iter().map(AppConnectorId), + ), + ); + } } } McpServerContribution::Remove { name } => { @@ -316,7 +343,7 @@ impl McpManager { }; let loaded_plugins = loaded_plugins.without_plugins(disabled_plugin_ids); let plugins_available = - selected_plugin_available || !loaded_plugins.capability_summaries().is_empty(); + !selected_plugins.is_empty() || !loaded_plugins.capability_summaries().is_empty(); let mut mcp_config = config .to_mcp_config_with_loaded_plugins(&loaded_plugins, selected_plugin_registrations); let mut catalog = mcp_config.mcp_server_catalog.to_builder(); 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 33497f8549..0d69fd1675 100644 --- a/codex-rs/core/src/session/active_plugin_inventory_tests.rs +++ b/codex-rs/core/src/session/active_plugin_inventory_tests.rs @@ -78,7 +78,7 @@ async fn active_plugin_ids_select_and_validate_by_source() { let selected_plugins = selected_ids .into_iter() .map(|plugin_id| SelectedPluginIdentity { - selected_root_id: "capability-root".to_string(), + selected_root_id: Some("capability-root".to_string()), plugin_id: plugin_id.to_string(), }) .collect::>(); @@ -127,7 +127,7 @@ 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: "capability-root".to_string(), + selected_root_id: Some("capability-root".to_string()), plugin_id: id.clone(), }; turn_context.active_host_plugin_identities = Some(Vec::new()); @@ -156,7 +156,7 @@ async fn active_plugin_ids_enforce_bounds_after_selection_and_deduplication() { let selected_plugins = ids[count / 2 - 1..] .iter() .map(|id| SelectedPluginIdentity { - selected_root_id: "capability-root".to_string(), + selected_root_id: Some("capability-root".to_string()), plugin_id: id.clone(), }) .collect::>(); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 7e43cdb7f6..b568942429 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3905,7 +3905,7 @@ impl Session { selected_plugins.plugins.retain(|plugin| { ready_selected_capability_roots .iter() - .any(|root| root.id == plugin.selected_root_id) + .any(|root| plugin.selected_root_id.as_ref() == Some(&root.id)) }); extension_data.insert(selected_plugins.clone()); let tool_router = turn::built_tools( diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 6f8b3b9abf..49a8ab8650 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1589,8 +1589,16 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() fn contribute<'a>( &'a self, - context: codex_extension_api::McpServerContributionContext<'a, Config>, + _context: codex_extension_api::McpServerContributionContext<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, Vec> + { + Box::pin(async { Vec::new() }) + } + + fn selected_plugins<'a>( + &'a self, + context: codex_extension_api::McpServerContributionContext<'a, Config>, + ) -> codex_extension_api::ExtensionFuture<'a, Vec>> { Box::pin(async move { let thread_init = context @@ -1619,22 +1627,18 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() &selected_root.location; server.environment_id = environment_id.clone(); server.enabled = false; - let plugin_id = selected_root.id; - vec![ - codex_extension_api::McpServerContribution::SelectedPluginPackage { - selected_root_id: plugin_id.clone(), - plugin_id: plugin_id.clone(), - plugin_display_name: plugin_id.clone(), - connector_ids: vec![], - }, - codex_extension_api::McpServerContribution::SelectedPlugin { - name: plugin_id.clone(), - plugin_display_name: plugin_id.clone(), - plugin_id, - selection_order: 0, - config: Box::new(server), - }, - ] + let plugin_id = format!("plugin-{}", selected_root.id); + vec![codex_extension_api::SelectedPlugin { + selected_root_id: selected_root.id.clone(), + plugin_id: plugin_id.clone(), + mcp: Box::pin(async move { + codex_extension_api::SelectedPluginContribution { + plugin_display_name: plugin_id, + connector_ids: vec![format!("{}-connector", selected_root.id)], + servers: vec![(selected_root.id, server)], + } + }), + }] }) } } @@ -1661,7 +1665,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() }); let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new(); extensions.thread_lifecycle_contributor(recorder.clone()); - extensions.mcp_server_contributor(recorder); + extensions.mcp_server_contributor(recorder.clone()); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( @@ -1810,7 +1814,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() .get("originator"), Some(&"codex_work_desktop".to_string()) ); - for disabled_plugin_ids in [vec!["selected-a".to_string()], vec![]] { + for disabled_plugin_ids in [vec!["plugin-selected-a".to_string()], vec![]] { let projection = first_session .services .mcp_manager @@ -1833,12 +1837,24 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() .await; assert_eq!( projection.selected_plugins.disabled_plugin_roots, - disabled_plugin_ids + if disabled_plugin_ids.is_empty() { + vec![] + } else { + vec!["selected-a".to_string()] + } ); assert_eq!( selected_servers(&projection.config).contains_key("selected-a"), disabled_plugin_ids.is_empty() ); + assert_eq!( + projection + .config + .connector_snapshot + .disabled_connector_ids() + .contains("selected-a-connector"), + !disabled_plugin_ids.is_empty() + ); assert_eq!( projection.selected_plugins.plugins.len(), usize::from(disabled_plugin_ids.is_empty()) diff --git a/codex-rs/ext/extension-api/src/contributors.rs b/codex-rs/ext/extension-api/src/contributors.rs index 10508fc1b2..81711a81de 100644 --- a/codex-rs/ext/extension-api/src/contributors.rs +++ b/codex-rs/ext/extension-api/src/contributors.rs @@ -29,6 +29,8 @@ pub use approval_review::SynchronousApprovalReviewer; pub use context::TurnContextContributionInput; pub use mcp::McpServerContribution; pub use mcp::McpServerContributionContext; +pub use mcp::SelectedPlugin; +pub use mcp::SelectedPluginContribution; pub use mcp::SelectedPluginIdentity; pub use mcp::SelectedPluginSnapshot; pub use prompt::PromptFragment; @@ -69,14 +71,15 @@ pub type ExtensionFuture<'a, T> = Pin + Send + 'a>>; /// Extension contribution that resolves runtime MCP servers from host config. /// -/// Contributors run in registration order. Later contributions for the same +/// Contributors run in registration order. Later ordinary server contributions for the same /// name replace earlier ones. Implementations must contribute only names they /// own and must apply any source-specific policy before returning a server. /// Thread-scoped resolution exposes the host-seeded thread inputs; global /// resolution exposes none and must not imply a local fallback. Thread inputs /// are frozen for the runtime and do not include lifecycle-contributor state. -/// Auto-discovered plugin servers are resolved by the plugin manager. A -/// thread-selected plugin contribution must carry its own package provenance. +/// Auto-discovered plugin servers are resolved by the plugin manager. Declare +/// executor plugins separately so the host attributes their MCP servers +/// and connectors to the same identity as their other capabilities. pub trait McpServerContributor: Send + Sync { /// Stable identity used for registration provenance and conflict diagnostics. fn id(&self) -> &'static str; @@ -85,6 +88,17 @@ pub trait McpServerContributor: Send + Sync { &'a self, context: McpServerContributionContext<'a, C>, ) -> ExtensionFuture<'a, Vec>; + + /// 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 + /// plugin wins if several provide the same server. + fn selected_plugins<'a>( + &'a self, + _context: McpServerContributionContext<'a, C>, + ) -> ExtensionFuture<'a, Vec>> { + Box::pin(async { Vec::new() }) + } } /// Extension contribution that adds prompt fragments during prompt assembly. diff --git a/codex-rs/ext/extension-api/src/contributors/mcp.rs b/codex-rs/ext/extension-api/src/contributors/mcp.rs index 69f8c6cdbb..ba1631ceb9 100644 --- a/codex-rs/ext/extension-api/src/contributors/mcp.rs +++ b/codex-rs/ext/extension-api/src/contributors/mcp.rs @@ -5,6 +5,7 @@ use codex_protocol::protocol::SessionSource; use crate::ExtensionData; use crate::ExtensionDataInit; +use crate::ExtensionFuture; /// Input supplied while resolving MCP server contributions. /// @@ -127,7 +128,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { } } -/// Validated plugin identities projected for the current set of selected roots. +/// Selected plugin identities and executor roots whose skills must be hidden. #[derive(Clone, Debug, Default)] pub struct SelectedPluginSnapshot { pub plugins: Vec, @@ -135,11 +136,29 @@ pub struct SelectedPluginSnapshot { pub disabled_plugin_roots: Vec, } -/// The configured identity of a plugin resolved from one selected root. +/// The configured identity of a selected plugin and, when present, the root owning its skills. #[derive(Clone, Debug)] 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, +} + +/// An executor plugin and its deferred MCP data. Callers that only need the identity can drop +/// `mcp` without loading server or connector configuration. A plugin still owns its skills when +/// it has no MCP data. +pub struct SelectedPlugin<'a> { pub selected_root_id: String, pub plugin_id: String, + pub mcp: ExtensionFuture<'a, SelectedPluginContribution>, +} + +/// MCP data attributed by the host to the plugin that declared it. +#[derive(Clone)] +pub struct SelectedPluginContribution { + pub plugin_display_name: String, + pub connector_ids: Vec, + pub servers: Vec<(String, McpServerConfig)>, } /// One extension-owned overlay for the runtime MCP server configuration. @@ -164,17 +183,9 @@ pub enum McpServerContribution { /// Overrides the HTTP protocol mode, or uses the hosted Apps default when absent. protocol_mode: Option, }, - /// Registers a server declared by a plugin selected for this thread. - SelectedPlugin { - name: String, - plugin_id: String, - plugin_display_name: String, - selection_order: usize, - config: Box, - }, - /// Records a plugin selected for this thread and any connector IDs it declares. - SelectedPluginPackage { - selected_root_id: String, + /// Attributes Apps connectors to an account-hosted plugin. Plugins from executor folders must + /// use `McpServerContributor::selected_plugins`, even if they only provide Apps connectors. + HostedPluginConnectors { plugin_id: String, plugin_display_name: String, connector_ids: Vec, diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index ef2f3886a2..ebac8d1363 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -61,6 +61,8 @@ pub use contributors::PreviousWorldStateSection; pub use contributors::PromptFragment; pub use contributors::PromptSlot; pub use contributors::RenderedWorldStateFragment; +pub use contributors::SelectedPlugin; +pub use contributors::SelectedPluginContribution; pub use contributors::SelectedPluginIdentity; pub use contributors::SelectedPluginSnapshot; pub use contributors::SkillInvocationContributor; diff --git a/codex-rs/ext/mcp/src/cloud_plugin.rs b/codex-rs/ext/mcp/src/cloud_plugin.rs index 55cd2b1bbc..2d10b403d2 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin.rs @@ -3,13 +3,13 @@ use crate::PluginListQuery; use crate::PluginsThreadState; use crate::plugin_contributor::PluginContributor; -use crate::plugin_contributor_state::SelectedPluginMetadata; use codex_core::config::Config; use codex_core_plugins::PluginCatalog; use codex_core_plugins::PluginIdentity; use codex_extension_api::ConfigContributor; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionFuture; +use codex_extension_api::McpServerContribution; use codex_extension_api::ThreadLifecycleContributor; use codex_extension_api::ThreadStartInput; use codex_extension_api::TurnLifecycleContributor; @@ -118,41 +118,29 @@ impl TurnLifecycleContributor for PluginContributor { } /// Cloud packages contribute Apps attribution, never executor MCP declarations. -pub(crate) struct CloudPluginMetadata { - pub(crate) selection_order: usize, - pub(crate) selected_root_id: String, - pub(crate) metadata: SelectedPluginMetadata, -} - -pub(crate) fn catalog_to_metadata(catalog: &PluginCatalog) -> Vec { +pub(crate) fn hosted_plugin_connectors(catalog: &PluginCatalog) -> Vec { catalog .entries .iter() - .enumerate() - .filter_map(|(selection_order, entry)| { + .filter_map(|entry| { let PluginIdentity::Remote { remote_plugin_id } = &entry.id else { return None; }; let mut seen_connector_ids = HashSet::new(); - Some(CloudPluginMetadata { - selection_order, - selected_root_id: format!("cloud:{remote_plugin_id}"), - metadata: SelectedPluginMetadata { - plugin_id: remote_plugin_id.clone(), - plugin_display_name: entry - .display_name - .chars() - .take(MAX_PLUGIN_DISPLAY_NAME_CHARS) - .collect(), - servers: Vec::new(), - connector_ids: entry - .connector_ids - .iter() - .filter(|connector_id| seen_connector_ids.insert((*connector_id).clone())) - .take(MAX_CONNECTOR_IDS_PER_PLUGIN) - .cloned() - .collect(), - }, + Some(McpServerContribution::HostedPluginConnectors { + plugin_id: remote_plugin_id.clone(), + plugin_display_name: entry + .display_name + .chars() + .take(MAX_PLUGIN_DISPLAY_NAME_CHARS) + .collect(), + connector_ids: entry + .connector_ids + .iter() + .filter(|connector_id| seen_connector_ids.insert((*connector_id).clone())) + .take(MAX_CONNECTOR_IDS_PER_PLUGIN) + .cloned() + .collect(), }) }) .take(MAX_PROJECTED_CLOUD_PLUGINS) diff --git a/codex-rs/ext/mcp/src/cloud_plugin_tests.rs b/codex-rs/ext/mcp/src/cloud_plugin_tests.rs index 8ac0f36b27..20c9502207 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin_tests.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin_tests.rs @@ -15,12 +15,17 @@ use codex_core_plugins::PluginIdentity; use codex_exec_server::EnvironmentManager; use codex_extension_api::ContextContributor; use codex_extension_api::ExtensionRegistryBuilder; +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_protocol::capabilities::CapabilityRootLocation; +use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; +use codex_utils_path_uri::PathUri; use core_test_support::apps_test_server::AppsTestServer; use core_test_support::apps_test_server::recorded_apps_tool_calls; use core_test_support::responses; @@ -178,6 +183,66 @@ async fn disabling_plugins_clears_cloud_catalog_and_skips_discovery() -> anyhow: Ok(()) } +#[tokio::test] +async fn account_change_clears_hosted_plugins_before_waiting_for_executor() -> anyhow::Result<()> { + struct UnreadyExecutor; + impl codex_exec_server::NoiseRendezvousConnectProvider for UnreadyExecutor { + fn connect_bundle( + &self, + _: codex_exec_server::NoiseChannelPublicKey, + ) -> futures::future::BoxFuture< + '_, + Result< + codex_exec_server::NoiseRendezvousConnectBundle, + codex_exec_server::ExecServerError, + >, + > { + Box::pin(futures::future::pending()) + } + } + + let home = tempfile::tempdir()?; + let mut config = codex_core::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 manager = Arc::new(EnvironmentManager::default_for_tests()); + manager + .materialize_pending_noise_environment("unready".to_string(), Arc::new(UnreadyExecutor))?; + let contributor = PluginContributor { + providers: PluginProviders::new(Arc::new(ExecutorPluginProvider::new(manager))) + .with_cloud_provider(Arc::new(CloudProvider::default())), + }; + let thread = ExtensionData::new("thread"); + let state = thread.get_or_init(PluginsThreadState::default); + let old_auth = state + .contributor_state() + .prepare_cloud_generation(/*mcp_resources*/ None); + assert!( + state + .contributor_state() + .publish_cloud_catalog(old_auth.as_ref(), cloud_catalog()) + ); + let roots = [SelectedCapabilityRoot { + id: "plugin".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: "unready".to_string(), + path: PathUri::parse("file:///plugins/demo")?, + }, + }]; + let init = codex_extension_api::ExtensionDataInit::default(); + let context = McpServerContributionContext::for_step( + &config, &init, &thread, "test", &roots, /*executor_capability_discovery*/ None, + ) + .with_auth_changed(/*auth_changed*/ true); + let mut loading = std::pin::pin!(contributor.selected_plugins(context)); + assert!(futures::poll!(&mut loading).is_pending()); + assert_eq!(state.cloud_catalog(), None); + Ok(()) +} + struct CloudProvider { reply: Mutex>, calls: AtomicUsize, @@ -187,8 +252,10 @@ struct CloudProvider { resume_discovery: Notify, } +type ObservedPackages = Vec<(String, Option)>; + #[derive(Default)] -struct StepPackages(Mutex>>); +struct StepPackages(Mutex>); impl ContextContributor for StepPackages { fn contribute_world_state<'a>( @@ -206,7 +273,7 @@ impl ContextContributor for StepPackages { snapshot .plugins .iter() - .map(|plugin| plugin.selected_root_id.clone()) + .map(|plugin| (plugin.plugin_id.clone(), plugin.selected_root_id.clone())) .collect() }); Vec::new() @@ -362,14 +429,14 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { ( "discovered", Ok(cloud_catalog()), - Some("cloud:plugin_remote"), + Some("plugin_remote"), None, Some(calendar), ), ( "same-account-failure", Err("temporarily unavailable".to_string()), - Some("cloud:plugin_remote"), + Some("plugin_remote"), None, None, ), @@ -377,7 +444,7 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { ( "restored", Ok(cloud_catalog()), - Some("cloud:plugin_remote"), + Some("plugin_remote"), None, None, ), @@ -391,7 +458,7 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { ( "new-account-discovered", Ok(gmail_catalog), - Some("cloud:plugin_gmail"), + Some("plugin_gmail"), None, Some(gmail), ), @@ -570,7 +637,12 @@ async fn run_cloud_catalog_refresh_lifecycle() -> anyhow::Result<()> { .0 .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), - Some(expected_plugin.into_iter().map(str::to_string).collect()), + Some( + expected_plugin + .into_iter() + .map(|id| (id.to_owned(), None)) + .collect() + ), "turn {turn}" ); assert_eq!( diff --git a/codex-rs/ext/mcp/src/plugin.rs b/codex-rs/ext/mcp/src/plugin.rs index 21aac3743d..d3ba5d4e52 100644 --- a/codex-rs/ext/mcp/src/plugin.rs +++ b/codex-rs/ext/mcp/src/plugin.rs @@ -7,17 +7,19 @@ use codex_extension_api::ExtensionFuture; use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; use codex_extension_api::McpServerContributor; +use codex_extension_api::SelectedPlugin; +use codex_extension_api::SelectedPluginContribution; use codex_features::Feature; use codex_protocol::capabilities::SelectedCapabilityRoot; use std::collections::HashMap; +use std::sync::Arc; +use std::sync::OnceLock; use self::provider::PluginMcpProvider; use crate::PluginsThreadState; -use crate::cloud_plugin::CloudPluginMetadata; -use crate::cloud_plugin::catalog_to_metadata; +use crate::cloud_plugin::hosted_plugin_connectors; use crate::plugin_contributor::PluginContributor; use crate::plugin_contributor_state::CachedSelectedRoot; -use crate::plugin_contributor_state::SelectedPluginMetadata; mod discovery; mod provider; @@ -33,7 +35,7 @@ impl PluginContributor { &self, state: &PluginsThreadState, selected_root: &SelectedCapabilityRoot, - ) -> Option { + ) -> Option { if let Some(cached) = state .contributor_state() .executor_cache @@ -83,8 +85,7 @@ impl PluginContributor { .into_iter() .map(|declaration| declaration.connector_id.0) .collect(); - Some(SelectedPluginMetadata { - plugin_id: plugin.plugin().selected_root_id().to_string(), + Some(SelectedPluginContribution { plugin_display_name: plugin.plugin().manifest().display_name().to_string(), servers, connector_ids, @@ -118,23 +119,44 @@ impl McpServerContributor for PluginContributor { let Some(thread_store) = context.thread_store() else { return Vec::new(); }; - // Cloud projection is gated independently; executor projection remains unchanged. let cloud_plugins_enabled = self.providers.cloud.is_some() && context.config().features.enabled(Feature::Plugins); let state = thread_store.get_or_init(PluginsThreadState::default); - if !cloud_plugins_enabled || context.auth_changed() { - // Clear stale cloud metadata before projecting a replacement runtime. + if !cloud_plugins_enabled { + state.contributor_state().cloud_generation = None; + return Vec::new(); + } + // Core reads hosted connectors after finishing the executor registrations. + state + .cloud_catalog() + .as_ref() + .map(hosted_plugin_connectors) + .unwrap_or_default() + }) + } + + fn selected_plugins<'a>( + &'a self, + context: McpServerContributionContext<'a, Config>, + ) -> ExtensionFuture<'a, Vec>> { + Box::pin(async move { + let Some(thread_store) = context.thread_store() else { + return Vec::new(); + }; + let state = thread_store.get_or_init(PluginsThreadState::default); + if context.auth_changed() { + // Clear the old account before executor loading can yield to other work. state.contributor_state().cloud_generation = None; } let selected_roots = context .ready_selected_capability_roots() .unwrap_or_default(); - let plugin_policies = - configured_plugin_mcp_server_policies(&context.config().config_layer_stack); - let mut contributions = Vec::new(); + // All plugins in this refresh share the same policy, but step-only readers never need it. + let plugin_policies = Arc::new(OnceLock::new()); + let mut plugins = Vec::new(); if let Some(snapshot) = context.executor_capability_discovery() { - for (selection_order, root) in snapshot.roots().iter().enumerate() { + for root in snapshot.roots() { let discovery = match &root.result { Ok(discovery) => discovery.as_ref(), Err(error) => { @@ -146,91 +168,77 @@ impl McpServerContributor for PluginContributor { continue; } }; - let Some(plugin) = - discovery::metadata_from_discovery(&root.selected_root, discovery) + let Some((manifest, plugin_files)) = + discovery::manifest_from_discovery(&root.selected_root, discovery) else { continue; }; - contributions.extend(project_metadata( - context.config(), - plugin_policies.get(&plugin.plugin_id), - selection_order, - &root.selected_root.id, - plugin, - )); + let plugin_policies = Arc::clone(&plugin_policies); + plugins.push(SelectedPlugin { + selected_root_id: root.selected_root.id.clone(), + plugin_id: root.selected_root.id.clone(), + mcp: Box::pin(async move { + let metadata = discovery::metadata_from_discovery( + &root.selected_root, + discovery, + plugin_files, + manifest, + ); + project_metadata( + context.config(), + &root.selected_root.id, + &plugin_policies, + metadata, + ) + }), + }); } } else { - for (selection_order, selected_root) in selected_roots.iter().enumerate() { - let Some(plugin) = self.metadata_for_root(&state, selected_root).await else { + for selected_root in selected_roots { + let Some(metadata) = self.metadata_for_root(&state, selected_root).await else { continue; }; - contributions.extend(project_metadata( - context.config(), - plugin_policies.get(&plugin.plugin_id), - selection_order, - &selected_root.id, - plugin, - )); + let plugin_policies = Arc::clone(&plugin_policies); + plugins.push(SelectedPlugin { + selected_root_id: selected_root.id.clone(), + plugin_id: selected_root.id.clone(), + mcp: Box::pin(async move { + project_metadata( + context.config(), + &selected_root.id, + &plugin_policies, + metadata, + ) + }), + }); } } - - // V1 has no remote plugin identities, so cloud packages are additive for now. - // Clone the current snapshot after executor loading and release the lock before projection. - if cloud_plugins_enabled && let Some(catalog) = state.cloud_catalog() { - for CloudPluginMetadata { - selection_order, - selected_root_id, - metadata, - } in catalog_to_metadata(&catalog) - { - contributions.extend(project_metadata( - context.config(), - /*plugin_policy*/ None, - selection_order, - &selected_root_id, - metadata, - )); - } - } - contributions + plugins }) } } fn project_metadata( config: &Config, - plugin_policy: Option<&HashMap>, - selection_order: usize, - selected_root_id: &str, - plugin: SelectedPluginMetadata, -) -> Vec { + plugin_id: &str, + plugin_policies: &OnceLock>>, + plugin: SelectedPluginContribution, +) -> SelectedPluginContribution { let mut servers = if config.features.enabled(Feature::Plugins) { - plugin.servers.iter().cloned().collect::>() + plugin.servers.into_iter().collect::>() } else { HashMap::new() }; - if let Some(plugin_policy) = plugin_policy { - apply_configured_plugin_mcp_server_policies(plugin_policy, &mut servers); + if !servers.is_empty() { + if let Some(plugin_policy) = plugin_policies + .get_or_init(|| configured_plugin_mcp_server_policies(&config.config_layer_stack)) + .get(plugin_id) + { + apply_configured_plugin_mcp_server_policies(plugin_policy, &mut servers); + } + config.apply_plugin_mcp_server_requirements(plugin_id, &mut servers); } - config.apply_plugin_mcp_server_requirements(&plugin.plugin_id, &mut servers); let mut servers = servers.into_iter().collect::>(); servers.sort_unstable_by(|left, right| left.0.cmp(&right.0)); - let mut contributions = servers - .into_iter() - .map(|(name, config)| McpServerContribution::SelectedPlugin { - name, - plugin_id: plugin.plugin_id.clone(), - plugin_display_name: plugin.plugin_display_name.clone(), - selection_order, - config: Box::new(config), - }) - .collect::>(); - // Keep the package visible even when it contributes only skills. - contributions.push(McpServerContribution::SelectedPluginPackage { - selected_root_id: selected_root_id.to_owned(), - plugin_id: plugin.plugin_id, - plugin_display_name: plugin.plugin_display_name, - connector_ids: plugin.connector_ids, - }); - contributions + SelectedPluginContribution { servers, ..plugin } } diff --git a/codex-rs/ext/mcp/src/plugin/discovery.rs b/codex-rs/ext/mcp/src/plugin/discovery.rs index 583698ff67..d556755e16 100644 --- a/codex-rs/ext/mcp/src/plugin/discovery.rs +++ b/codex-rs/ext/mcp/src/plugin/discovery.rs @@ -1,17 +1,18 @@ use codex_connectors::parse_plugin_app_config; +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::parse_executor_plugin_mcp_config; use codex_plugin::manifest::PluginManifestMcpServers; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; -use super::SelectedPluginMetadata; - -pub(super) fn metadata_from_discovery( +pub(super) fn manifest_from_discovery<'a>( selected_root: &SelectedCapabilityRoot, - discovery: &CapabilityRootDiscovery, -) -> Option { + discovery: &'a CapabilityRootDiscovery, +) -> Option<(UriPluginManifest, &'a DiscoveredPluginFiles)> { for warning in &discovery.warnings { tracing::warn!( selected_root = selected_root.id, @@ -28,12 +29,12 @@ pub(super) fn metadata_from_discovery( return None; } let plugin_files = discovery.plugin.as_ref()?; - let manifest = match parse_plugin_manifest_uri( + match parse_plugin_manifest_uri( &discovery.path, &plugin_files.manifest.path, &plugin_files.manifest.contents, ) { - Ok(manifest) => manifest, + Ok(manifest) => Some((manifest, plugin_files)), Err(error) => { tracing::warn!( selected_root = selected_root.id, @@ -41,9 +42,17 @@ pub(super) fn metadata_from_discovery( %error, "failed to parse exec-server-discovered plugin manifest" ); - return None; + None } - }; + } +} + +pub(super) fn metadata_from_discovery( + selected_root: &SelectedCapabilityRoot, + discovery: &CapabilityRootDiscovery, + plugin_files: &DiscoveredPluginFiles, + manifest: UriPluginManifest, +) -> SelectedPluginContribution { let CapabilityRootLocation::Environment { environment_id, .. } = &selected_root.location; let servers = match manifest.paths.mcp_servers.as_ref() { Some(PluginManifestMcpServers::Object(contents)) => { @@ -117,12 +126,11 @@ pub(super) fn metadata_from_discovery( .map(|declaration| declaration.connector_id.0) .collect(); - Some(SelectedPluginMetadata { - plugin_id: selected_root.id.clone(), + SelectedPluginContribution { plugin_display_name: manifest.display_name().to_string(), servers, connector_ids, - }) + } } fn parse_mcp_servers( diff --git a/codex-rs/ext/mcp/src/plugin_contributor_state.rs b/codex-rs/ext/mcp/src/plugin_contributor_state.rs index ab088f37a2..e86b9cb2c5 100644 --- a/codex-rs/ext/mcp/src/plugin_contributor_state.rs +++ b/codex-rs/ext/mcp/src/plugin_contributor_state.rs @@ -1,6 +1,7 @@ //! Thread-owned cloud catalog and legacy selected-root cache, guarded by a short-lived mutex. use codex_core_plugins::PluginCatalog; +use codex_extension_api::SelectedPluginContribution; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpResourceClient; use codex_mcp::McpResourceClientAuthKey; @@ -109,14 +110,5 @@ impl PluginContributorState { pub(crate) struct CachedSelectedRoot { pub(crate) root: SelectedCapabilityRoot, - pub(crate) metadata: Option, -} - -/// Frozen declarations retain the logical environment ID across executor reconnections. -#[derive(Clone)] -pub(crate) struct SelectedPluginMetadata { - pub(crate) plugin_id: String, - pub(crate) plugin_display_name: String, - pub(crate) servers: Vec<(String, codex_config::McpServerConfig)>, - pub(crate) connector_ids: Vec, + pub(crate) metadata: Option, } diff --git a/codex-rs/ext/mcp/tests/plugin_mcp.rs b/codex-rs/ext/mcp/tests/plugin_mcp.rs index 7b05bea104..9da77e4d76 100644 --- a/codex-rs/ext/mcp/tests/plugin_mcp.rs +++ b/codex-rs/ext/mcp/tests/plugin_mcp.rs @@ -21,8 +21,8 @@ use codex_exec_server::LOCAL_ENVIRONMENT_ID; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistryBuilder; -use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; +use codex_extension_api::SelectedPluginContribution; use codex_features::Feature; use codex_login::CodexAuth; use codex_mcp_extension::PluginProviders; @@ -163,22 +163,14 @@ async fn selected_plugin_package_is_contributed_without_servers_or_connectors() .await?; let contributions = raw_selected_plugin_contributions(&config, plugin_root.path()).await?; - let package = contributions.into_iter().find_map(|contribution| { - let McpServerContribution::SelectedPluginPackage { + let package = contributions + .into_iter() + .next() + .map(|(_, plugin_id, contribution)| PackageSummary { plugin_id, - plugin_display_name, - connector_ids, - .. - } = contribution - else { - return None; - }; - Some(PackageSummary { - plugin_id, - plugin_display_name, - connector_ids, - }) - }); + plugin_display_name: contribution.plugin_display_name, + connector_ids: contribution.connector_ids, + }); assert_eq!( package, @@ -223,8 +215,8 @@ plugins = false assert!( matches!( direct.as_slice(), - [McpServerContribution::SelectedPluginPackage { selected_root_id, .. }] - if selected_root_id == "selected-root" + [(selected_root_id, _, contribution)] + if selected_root_id == "selected-root" && contribution.servers.is_empty() ), "managed Plugins disable should preserve only the direct selected-root identity" ); @@ -237,8 +229,8 @@ plugins = false assert!( matches!( discovered.as_slice(), - [McpServerContribution::SelectedPluginPackage { selected_root_id, .. }] - if selected_root_id == "selected-root" + [(selected_root_id, _, contribution)] + if selected_root_id == "selected-root" && contribution.servers.is_empty() ), "managed Plugins disable should preserve only the discovered selected-root identity" ); @@ -310,10 +302,7 @@ default_tools_approval_mode = "auto" let mut servers = raw_selected_plugin_contributions(&config, plugin_root.path()) .await? .into_iter() - .filter_map(|contribution| match contribution { - McpServerContribution::SelectedPlugin { name, config, .. } => Some((name, config)), - _ => None, - }) + .flat_map(|(_, _, contribution)| contribution.servers) .collect::>(); let server = servers .remove("first") @@ -384,27 +373,18 @@ async fn selected_plugin_contributions( Ok(raw_selected_plugin_contributions(config, plugin_root) .await? .into_iter() - .filter_map(|contribution| match contribution { - McpServerContribution::SelectedPlugin { - name, - plugin_id, - plugin_display_name, - selection_order, - config, - } => Some(ContributionSummary { - name, - plugin_id, - plugin_display_name, - selection_order, - enabled: config.enabled, - }), - McpServerContribution::SelectedPluginPackage { .. } => None, - McpServerContribution::Set { .. } - | McpServerContribution::SetWithProtocolMode { .. } - | McpServerContribution::HostedApps { .. } - | McpServerContribution::Remove { .. } => { - panic!("expected selected plugin contribution") - } + .enumerate() + .flat_map(|(selection_order, (_, plugin_id, contribution))| { + contribution + .servers + .into_iter() + .map(move |(name, config)| ContributionSummary { + name, + plugin_id: plugin_id.clone(), + plugin_display_name: contribution.plugin_display_name.clone(), + selection_order, + enabled: config.enabled, + }) }) .collect()) } @@ -412,7 +392,7 @@ async fn selected_plugin_contributions( async fn raw_selected_plugin_contributions( config: &Config, plugin_root: &std::path::Path, -) -> Result, Box> { +) -> Result, Box> { let mut builder = ExtensionRegistryBuilder::new(); let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); codex_mcp_extension::install_plugins(&mut builder, Arc::clone(&environment_manager)); @@ -439,8 +419,8 @@ async fn raw_selected_plugin_contributions( None }; - let contributions = registry.mcp_server_contributors()[0] - .contribute(McpServerContributionContext::for_step( + let selected = registry.mcp_server_contributors()[0] + .selected_plugins(McpServerContributionContext::for_step( config, &thread_init, &thread_store, @@ -449,6 +429,10 @@ async fn raw_selected_plugin_contributions( executor_capability_discovery.as_ref(), )) .await; + let mut contributions = Vec::new(); + for plugin in selected { + contributions.push((plugin.selected_root_id, plugin.plugin_id, plugin.mcp.await)); + } Ok(contributions) } diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 3415e5b473..ede044539b 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -75,7 +75,7 @@ pub(crate) fn attribute_executor_plugins( if let Some(plugin) = snapshot .plugins .iter() - .find(|plugin| plugin.selected_root_id == skill.authority.id) + .find(|plugin| plugin.selected_root_id.as_ref() == Some(&skill.authority.id)) { skill.plugin_id = Some(plugin.plugin_id.clone()); skill.analytics_scope = Some(SkillScope::User);