Load plugin skills with bounded concurrency

codex-rs/Cargo.lock

@@ -2766,6 +2766,7 @@ dependencies = [
  "codex-utils-plugins",
  "dirs",
  "flate2",
+ "futures",
  "indexmap 2.14.0",
  "libc",
  "pretty_assertions",
@@ -2808,6 +2809,7 @@ dependencies = [
  "codex-utils-plugins",
  "dirs",
  "dunce",
+ "futures",
  "pretty_assertions",
  "serde",
  "serde_json",

codex-rs/core-plugins/Cargo.toml

@@ -34,6 +34,7 @@ codex-utils-plugins = { workspace = true }
 chrono = { workspace = true }
 dirs = { workspace = true }
 flate2 = { workspace = true }
+futures = { workspace = true }
 indexmap = { workspace = true, features = ["serde"] }
 reqwest = { workspace = true }
 semver = { workspace = true }

codex-rs/core-plugins/src/loader.rs

@@ -41,6 +41,7 @@ use codex_protocol::protocol::Product;
 use codex_protocol::protocol::SkillScope;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_plugins::find_plugin_manifest_path;
+use futures::StreamExt;
 use indexmap::IndexMap;
 use serde::Deserialize;
 use serde_json::Value as JsonValue;
@@ -60,6 +61,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)]
@@ -149,10 +151,22 @@ async fn load_plugins_from_layer_stack_with_scope(
     let mut configured_plugins: Vec<_> = configured_plugins.into_iter().collect();
     configured_plugins.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));
 
-    let mut plugins = Vec::with_capacity(configured_plugins.len());
+    let loaded_plugins = futures::stream::iter(configured_plugins)
+        .map(|(configured_name, plugin)| {
+            let scope = &scope;
+            async move {
+                let loaded_plugin =
+                    load_plugin(configured_name.clone(), &plugin, store, scope).await;
+                (configured_name, loaded_plugin)
+            }
+        })
+        .buffered(MAX_CONCURRENT_PLUGIN_LOADS)
+        .collect::<Vec<_>>()
+        .await;
+
+    let mut plugins = Vec::with_capacity(loaded_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;
+    for (configured_name, loaded_plugin) in loaded_plugins {
         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

@@ -192,6 +192,19 @@ enabled = true
             .collect::<Vec<_>>()
     };
     assert_eq!(validation_state(&hooks_only), validation_state(&full));
+    assert_eq!(
+        full.iter()
+            .map(|plugin| plugin.config_name.as_str())
+            .collect::<Vec<_>>(),
+        vec![
+            "disabled@test",
+            "invalid",
+            "malformed@test",
+            "missing@test",
+            "valid@test",
+            "warning@test",
+        ]
+    );
 
     let full_valid = full
         .iter()

codex-rs/core-skills/Cargo.toml

@@ -31,6 +31,7 @@ codex-utils-path-uri = { workspace = true }
 codex-utils-plugins = { workspace = true }
 dirs = { workspace = true }
 dunce = { workspace = true }
+futures = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }

codex-rs/core-skills/src/plugin_skill_root_cache.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::sync::RwLock;
 
 use codex_utils_absolute_path::AbsolutePathBuf;
+use futures::StreamExt;
 
 use crate::SkillLoadOutcome;
 use crate::loader::SkillRoot;
@@ -12,6 +13,7 @@ use crate::loader::load_skill_root;
 use crate::model::SkillFileSystemsByPath;
 
 const MAX_CACHED_PLUGIN_SKILL_ROOTS: usize = 256;
+const MAX_CONCURRENT_SKILL_ROOT_LOADS: usize = 8;
 
 /// Shares parsed plugin skill-root snapshots between plugin and skill loading.
 ///
@@ -28,36 +30,40 @@ impl PluginSkillRootCache {
     where
         I: IntoIterator<Item = SkillRoot>,
     {
-        let mut snapshots = Vec::new();
-        for root in roots {
-            // Plugin skill roots always use local filesystem and User scope, so the absolute skill
-            // root path is sufficient to share their snapshot between plugin and skill loading.
-            let cache_key = root.plugin_root.as_ref().map(|_| root.path.clone());
-            let cached_snapshot = cache_key
-                .as_ref()
-                .and_then(|root| match self.snapshots.read() {
-                    Ok(cache) => cache.get(root).cloned(),
-                    Err(err) => err.into_inner().get(root).cloned(),
-                });
-            let snapshot = match cached_snapshot {
-                Some(snapshot) => snapshot,
-                None => {
-                    let snapshot = load_skill_root(root).await;
-                    if let Some(root) = cache_key {
-                        let mut cache = self
-                            .snapshots
-                            .write()
-                            .unwrap_or_else(std::sync::PoisonError::into_inner);
-                        if cache.len() < MAX_CACHED_PLUGIN_SKILL_ROOTS || cache.contains_key(&root)
-                        {
-                            cache.insert(root, snapshot.clone());
+        let snapshots = futures::stream::iter(roots)
+            .map(|root| async move {
+                // Plugin skill roots always use local filesystem and User scope, so the absolute
+                // skill root path is sufficient to share their snapshot between plugin and skill
+                // loading.
+                let cache_key = root.plugin_root.as_ref().map(|_| root.path.clone());
+                let cached_snapshot = cache_key
+                    .as_ref()
+                    .and_then(|root| match self.snapshots.read() {
+                        Ok(cache) => cache.get(root).cloned(),
+                        Err(err) => err.into_inner().get(root).cloned(),
+                    });
+                match cached_snapshot {
+                    Some(snapshot) => snapshot,
+                    None => {
+                        let snapshot = load_skill_root(root).await;
+                        if let Some(root) = cache_key {
+                            let mut cache = self
+                                .snapshots
+                                .write()
+                                .unwrap_or_else(std::sync::PoisonError::into_inner);
+                            if cache.len() < MAX_CACHED_PLUGIN_SKILL_ROOTS
+                                || cache.contains_key(&root)
+                            {
+                                cache.insert(root, snapshot.clone());
+                            }
                         }
+                        snapshot
                     }
-                    snapshot
                 }
-            };
-            snapshots.push(snapshot);
-        }
+            })
+            .buffered(MAX_CONCURRENT_SKILL_ROOT_LOADS)
+            .collect::<Vec<_>>()
+            .await;
 
         merge_skill_root_snapshots(snapshots)
     }