openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Separate selected plugin identities from MCP contributions (#47939)

Browse Source 
## Why

Plugin identity and skill ownership need to be available independently of MCP data, including for plugins without servers or connectors. Hosted plugins have no executor root and should not own executor skills.

## What changed

- Add `McpServerContributor::selected_plugins` to declare executor plugin identities with deferred MCP data. Let the host assign server attribution and precedence across contributors.
- Represent hosted connectors with `HostedPluginConnectors` and make `SelectedPluginIdentity::selected_root_id` optional instead of assigning synthetic roots to hosted plugins.
- Resolve executor contributions before reading hosted connectors, and clear stale hosted state on account changes before executor loading can wait.
- Preserve connector declarations from disabled plugins so Apps can hide those connectors.

## Testing

Add coverage for account changes while executor loading is pending. Update tests to verify hosted identities have no executor root and disabled plugins hide their connectors even when plugin IDs differ from root IDs.

GitOrigin-RevId: e7f511264c5874c1b56d5f3e4a46820724046db0
This commit is contained in:
sayan-oai
2026-09-24 21:03:50 +00:00
committed by copyberry
parent fbd0c0d96c
commit e8098eb406
14 changed files with 389 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/mcp.rs
+69 -42
View File
@@ -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();
codex-rs/core/src/session/active_plugin_inventory_tests.rs
+3 -3
View File
@@ -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::<Vec<_>>();
@@ -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::<Vec<_>>();
codex-rs/core/src/session/mod.rs
+1 -1
View File
@@ -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(
codex-rs/core/src/thread_manager_tests.rs
+36 -20
View File
@@ -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<codex_extension_api::McpServerContribution>>
{
Box::pin(async { Vec::new() })
}
fn selected_plugins<'a>(
&'a self,
context: codex_extension_api::McpServerContributionContext<'a, Config>,
) -> codex_extension_api::ExtensionFuture<'a, Vec<codex_extension_api::SelectedPlugin<'a>>>
{
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())
codex-rs/ext/extension-api/src/contributors.rs
+17 -3
View File
@@ -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<Box<dyn Future<Output = T> + 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<C: Sync>: Send + Sync {
/// Stable identity used for registration provenance and conflict diagnostics.
fn id(&self) -> &'static str;
@@ -85,6 +88,17 @@ pub trait McpServerContributor<C: Sync>: Send + Sync {
&'a self,
context: McpServerContributionContext<'a, C>,
) -> ExtensionFuture<'a, Vec<McpServerContribution>>;
/// 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<SelectedPlugin<'a>>> {
Box::pin(async { Vec::new() })
}
}
/// Extension contribution that adds prompt fragments during prompt assembly.
codex-rs/ext/extension-api/src/contributors/mcp.rs
+24 -13
View File
@@ -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<SelectedPluginIdentity>,
@@ -135,11 +136,29 @@ pub struct SelectedPluginSnapshot {
pub disabled_plugin_roots: Vec<String>,
}
/// 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<String>,
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<String>,
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<crate::McpProtocolMode>,
},
/// 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<McpServerConfig>,
},
/// 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<String>,
codex-rs/ext/extension-api/src/lib.rs
+2
View File
@@ -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;
codex-rs/ext/mcp/src/cloud_plugin.rs
+17 -29
View File
@@ -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<CloudPluginMetadata> {
pub(crate) fn hosted_plugin_connectors(catalog: &PluginCatalog) -> Vec<McpServerContribution> {
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)
codex-rs/ext/mcp/src/cloud_plugin_tests.rs
+79 -7
View File
@@ -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<Result<PluginCatalog, String>>,
calls: AtomicUsize,
@@ -187,8 +252,10 @@ struct CloudProvider {
resume_discovery: Notify,
}
type ObservedPackages = Vec<(String, Option<String>)>;
#[derive(Default)]
struct StepPackages(Mutex<Option<Vec<String>>>);
struct StepPackages(Mutex<Option<ObservedPackages>>);
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!(
codex-rs/ext/mcp/src/plugin.rs
+86 -78
View File
@@ -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<SelectedPluginMetadata> {
) -> Option<SelectedPluginContribution> {
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<Config> 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<SelectedPlugin<'a>>> {
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<Config> 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<String, PluginMcpServerConfig>>,
selection_order: usize,
selected_root_id: &str,
plugin: SelectedPluginMetadata,
) -> Vec<McpServerContribution> {
plugin_id: &str,
plugin_policies: &OnceLock<HashMap<String, HashMap<String, PluginMcpServerConfig>>>,
plugin: SelectedPluginContribution,
) -> SelectedPluginContribution {
let mut servers = if config.features.enabled(Feature::Plugins) {
plugin.servers.iter().cloned().collect::<HashMap<_, _>>()
plugin.servers.into_iter().collect::<HashMap<_, _>>()
} 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::<Vec<_>>();
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::<Vec<_>>();
// 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 }
}
codex-rs/ext/mcp/src/plugin/discovery.rs
+20 -12
View File
@@ -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<SelectedPluginMetadata> {
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(
codex-rs/ext/mcp/src/plugin_contributor_state.rs
+2 -10
View File
@@ -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<SelectedPluginMetadata>,
}
/// 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<String>,
pub(crate) metadata: Option<SelectedPluginContribution>,
}
codex-rs/ext/mcp/tests/plugin_mcp.rs
+32 -48
View File
@@ -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::<HashMap<_, _>>();
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<Vec<McpServerContribution>, Box<dyn std::error::Error>> {
) -> Result<Vec<(String, String, SelectedPluginContribution)>, Box<dyn std::error::Error>> {
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)
}
codex-rs/ext/skills/src/provider/executor.rs
+1 -1
View File
@@ -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);