[codex] Load plugins and skill roots concurrently

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,8 @@ 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::FutureExt;
+use futures::StreamExt;
 use indexmap::IndexMap;
 use serde::Deserialize;
 use serde_json::Value as JsonValue;
@@ -60,6 +62,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 +152,23 @@ 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)
+            }
+            .boxed()
+        })
+        .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,8 @@ use std::sync::Arc;
 use std::sync::RwLock;
 
 use codex_utils_absolute_path::AbsolutePathBuf;
+use futures::FutureExt;
+use futures::StreamExt;
 
 use crate::SkillLoadOutcome;
 use crate::loader::SkillRoot;
@@ -12,6 +14,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 +31,44 @@ 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);
-        }
+                .boxed()
+            })
+            .buffered(MAX_CONCURRENT_SKILL_ROOT_LOADS)
+            .collect::<Vec<_>>()
+            .await;
 
         merge_skill_root_snapshots(snapshots)
     }