diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index 83e691a99f..851ff4ecd1 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/codex-rs/core-plugins/src/loader.rs @@ -56,6 +56,8 @@ use codex_utils_plugins::PluginSkillRoot; use codex_utils_plugins::SkillDiscoveryMode; use codex_utils_plugins::find_plugin_manifest_path; use codex_utils_plugins::migrated_command_skills_root; +use futures::StreamExt; +use futures::stream; use serde::Deserialize; use serde_json::Value as JsonValue; use std::collections::HashMap; @@ -75,6 +77,7 @@ const DEFAULT_MCP_CONFIG_FILE: &str = ".mcp.json"; const DEFAULT_APP_CONFIG_FILE: &str = ".app.json"; const CONFIG_TOML_FILE: &str = "config.toml"; const CURATED_PLUGIN_CACHE_VERSION_SHA_PREFIX_LEN: usize = 8; +const MAX_CONCURRENT_PLUGIN_LOADS: usize = 8; /// Hook declarations and warnings resolved without loading other plugin capabilities. #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -191,8 +194,16 @@ async fn load_plugins_from_layer_stack_with_scope( let mut plugins = Vec::with_capacity(configured_plugins.len()); let mut seen_mcp_server_names = HashMap::::new(); - for (configured_name, plugin) in configured_plugins { - let loaded_plugin = load_plugin(configured_name.clone(), &plugin, store, &scope).await; + let scope = &scope; + // Overlap independent filesystem/skill loads, but consume them in configured-name order: + // that order determines duplicate MCP precedence and startup hook warning order. + let mut pending_plugins = stream::iter(configured_plugins) + .map(|(configured_name, plugin)| async move { + let loaded_plugin = load_plugin(configured_name.clone(), &plugin, store, scope).await; + (configured_name, loaded_plugin) + }) + .buffered(MAX_CONCURRENT_PLUGIN_LOADS); + while let Some((configured_name, loaded_plugin)) = pending_plugins.next().await { for name in loaded_plugin.mcp_servers.keys() { if let Some(previous_plugin) = seen_mcp_server_names.insert(name.clone(), configured_name.clone()) diff --git a/codex-rs/core-plugins/src/loader_tests.rs b/codex-rs/core-plugins/src/loader_tests.rs index cd96e72408..238fa6ff42 100644 --- a/codex-rs/core-plugins/src/loader_tests.rs +++ b/codex-rs/core-plugins/src/loader_tests.rs @@ -9,10 +9,38 @@ use codex_config::ConfigRequirements; use codex_config::ConfigRequirementsToml; use codex_config::config_toml::ConfigToml; use codex_plugin::PluginId; +use codex_skills::LoadedSkills; +use codex_skills::SkillLoadFuture; use codex_utils_plugins::AGENT_PLUGIN_SCHEMA_URI; use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use tempfile::TempDir; +// Yield at the skill loader so the buffered stream can poll its other plugin loads. +struct OverlapTrackingSkillRootLoader { + delegate: Arc>, + active: AtomicUsize, + max_active: AtomicUsize, +} + +impl SkillRootLoader for OverlapTrackingSkillRootLoader { + fn load_roots( + &self, + request: SkillRootLoadRequest, + ) -> SkillLoadFuture<'_, LoadedSkills> { + Box::pin(async move { + let active = self.active.fetch_add(1, Ordering::Relaxed) + 1; + self.max_active.fetch_max(active, Ordering::Relaxed); + tokio::task::yield_now().await; + let result = self.delegate.load_roots(request).await; + self.active.fetch_sub(1, Ordering::Relaxed); + result + }) + } +} + fn user_config_path(temp_dir: &TempDir, file_name: &str) -> AbsolutePathBuf { AbsolutePathBuf::from_absolute_path(temp_dir.path().join(file_name)) .expect("test user config path should be absolute") @@ -28,6 +56,103 @@ fn user_layer(path: AbsolutePathBuf, config: &str) -> ConfigLayerEntry { ) } +#[tokio::test] +async fn concurrent_plugin_loads_preserve_mcp_precedence_and_hook_warnings() { + let temp_dir = TempDir::new().expect("tempdir"); + let mut config = String::new(); + // Exercise more than one buffer of actual file reads, with conflicting MCP definitions. + let names: Vec<_> = (0..MAX_CONCURRENT_PLUGIN_LOADS + 2) + .map(|index| format!("plugin-{index:02}")) + .collect(); + // Declare plugin-00 last. Its name sorts first, so its shared MCP server should win. + for name in names.iter().rev() { + let root = temp_dir + .path() + .join(format!("plugins/cache/test/{name}/local")); + write_file( + &root.join(".codex-plugin/plugin.json"), + &format!(r#"{{"name":"{name}"}}"#), + ); + write_file( + &root.join(".mcp.json"), + &format!(r#"{{"mcpServers":{{"shared":{{"command":"{name}"}}}}}}"#), + ); + write_file( + &root.join("skills/example/SKILL.md"), + &format!("---\nname: {name}\ndescription: Example skill\n---\nInstructions\n"), + ); + write_file(&root.join("hooks/hooks.json"), "not valid json"); + config.push_str(&format!("[plugins.\"{name}@test\"]\nenabled = true\n")); + } + let stack = ConfigLayerStack::new( + vec![user_layer( + user_config_path(&temp_dir, "config.toml"), + &config, + )], + ConfigRequirements::default(), + ConfigRequirementsToml::default(), + ) + .expect("valid config layer stack"); + let store = PluginStore::new(temp_dir.path().to_path_buf()); + let config: ConfigToml = stack.effective_config().try_into().expect("plugin config"); + let skill_config_rules = skill_config_rules_from_stack(&stack); + let remote_plugin_id_resolver = RemotePluginIdResolver::default(); + let skill_root_loader = test_skill_root_loader(); + let scope = PluginLoadScope::AllCapabilities { + restriction_product: Some(Product::Codex), + skill_config_rules: &skill_config_rules, + plugin_skill_snapshots: None, + remote_plugin_id_resolver: &remote_plugin_id_resolver, + skill_root_loader: skill_root_loader.as_ref(), + }; + let mut expected = Vec::new(); + for name in &names { + let configured_name = format!("{name}@test"); + expected.push( + load_plugin( + configured_name.clone(), + &config.plugins.plugins[&configured_name], + &store, + &scope, + ) + .await, + ); + } + let overlap_loader = OverlapTrackingSkillRootLoader { + delegate: Arc::clone(&skill_root_loader), + active: AtomicUsize::new(0), + max_active: AtomicUsize::new(0), + }; + let loaded = load_plugins_from_layer_stack( + &stack, + &config.plugins, + RemoteInstalledPluginsSnapshot::default(), + &store, + /*plugin_skill_snapshots*/ None, + Some(Product::Codex), + /*remote_global_catalog_active*/ false, + &overlap_loader, + ) + .await; + assert_eq!(loaded, expected); + assert!( + (2..=MAX_CONCURRENT_PLUGIN_LOADS) + .contains(&overlap_loader.max_active.load(Ordering::Relaxed)), + "plugin skill loads should overlap without exceeding the buffer", + ); + assert!(loaded.iter().all(|plugin| plugin.has_enabled_skills)); + assert!( + loaded + .iter() + .all(|plugin| plugin.hook_load_warnings.len() == 1) + ); + let expected_server = expected[0].mcp_servers["shared"].clone(); + assert_eq!( + crate::PluginLoadOutcome::from_plugins(loaded).effective_mcp_servers(), + HashMap::from([("shared".to_string(), expected_server)]) + ); +} + #[tokio::test] async fn agent_plugin_overlay_apps_are_not_runtime_active() { let temp_dir = TempDir::new().expect("tempdir");