Load up to eight plugins concurrently (#57819)

## What changed

Overlap independent plugin filesystem and skill loads with a concurrency limit of eight. Consume results in configured-name order to preserve duplicate MCP server precedence and startup hook warning order.

## Testing

Add a regression test with ten plugins that verifies skill loads overlap within the concurrency limit, results match sequential loading, and MCP precedence and hook warnings are preserved.

GitOrigin-RevId: 6ce8a05954fad157d8559abe9fd2dc684692cf3e

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::<String, String>::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())

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<dyn SkillRootLoader<PluginSkillRoot>>,
+    active: AtomicUsize,
+    max_active: AtomicUsize,
+}
+
+impl SkillRootLoader<PluginSkillRoot> for OverlapTrackingSkillRootLoader {
+    fn load_roots(
+        &self,
+        request: SkillRootLoadRequest<PluginSkillRoot>,
+    ) -> 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");