diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 844a84ef739b..582c39b46de0 100644 --- a/codex-rs/Cargo.lock +++ b/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", diff --git a/codex-rs/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index b90192bfe960..bd823998e535 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/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 } diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index 4cc27583f591..65c75ff76b93 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/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::>() + .await; + + let mut plugins = Vec::with_capacity(loaded_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; + 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()) diff --git a/codex-rs/core-plugins/src/loader_tests.rs b/codex-rs/core-plugins/src/loader_tests.rs index 5c0a26115afd..e32d8d5700c5 100644 --- a/codex-rs/core-plugins/src/loader_tests.rs +++ b/codex-rs/core-plugins/src/loader_tests.rs @@ -192,6 +192,19 @@ enabled = true .collect::>() }; assert_eq!(validation_state(&hooks_only), validation_state(&full)); + assert_eq!( + full.iter() + .map(|plugin| plugin.config_name.as_str()) + .collect::>(), + vec![ + "disabled@test", + "invalid", + "malformed@test", + "missing@test", + "valid@test", + "warning@test", + ] + ); let full_valid = full .iter() diff --git a/codex-rs/core-skills/Cargo.toml b/codex-rs/core-skills/Cargo.toml index 7748540ac18a..3ac82be9a8f8 100644 --- a/codex-rs/core-skills/Cargo.toml +++ b/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 } diff --git a/codex-rs/core-skills/src/plugin_skill_root_cache.rs b/codex-rs/core-skills/src/plugin_skill_root_cache.rs index 9cd58caa80bc..4008b30829a3 100644 --- a/codex-rs/core-skills/src/plugin_skill_root_cache.rs +++ b/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, { - 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::>() + .await; merge_skill_root_snapshots(snapshots) }