From 2e4f55608b4ad26d9c48ea45a6fcd20bfd5e9fe8 Mon Sep 17 00:00:00 2001 From: "Adam Perry @ OpenAI" Date: Wed, 15 Jul 2026 15:51:59 +0000 Subject: [PATCH] Scan skill roots concurrently (#33369) ## What changed - Scan up to eight skill roots concurrently instead of loading them one at a time. - Share scan capacity across host-skill and plugin-skill loads to bound aggregate filesystem work. - Merge completed scans in input order so root precedence and error ordering remain deterministic. - Skip canonicalization and parsing work for empty roots. ## Testing - Add coverage for out-of-order scan completion and shared-capacity blocking. GitOrigin-RevId: 33a47aaa02aee7fe31a050bb04d51c1e2043d6ae --- codex-rs/core-plugins/src/loader.rs | 11 +- codex-rs/core-plugins/src/loader_tests.rs | 2 + codex-rs/core-plugins/src/manager.rs | 13 +- codex-rs/core-plugins/src/manager_tests.rs | 2 + .../core-plugins/src/tool_suggest_metadata.rs | 11 +- codex-rs/core-skills/src/loader.rs | 11 +- codex-rs/core-skills/src/loader_tests.rs | 166 +++++++++++++++++- codex-rs/core-skills/src/root_loader.rs | 104 ++++++----- codex-rs/core-skills/src/service.rs | 12 +- .../core-skills/tests/environment_loader.rs | 2 + .../tests/executor_file_system_authority.rs | 2 + 11 files changed, 288 insertions(+), 48 deletions(-) diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index 2b5c4782e8..160b710e9b 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/codex-rs/core-plugins/src/loader.rs @@ -53,6 +53,7 @@ use std::path::Path; use std::process::Command; use std::sync::Arc; use tempfile::TempDir; +use tokio::sync::Semaphore; use tracing::instrument; use tracing::warn; @@ -75,6 +76,7 @@ enum PluginLoadScope<'a> { restriction_product: Option, skill_config_rules: &'a SkillConfigRules, plugin_skill_snapshots: Option<&'a PluginSkillSnapshots>, + root_scan_slots: Arc, }, HooksOnly, } @@ -118,6 +120,7 @@ pub(crate) async fn load_plugins_from_layer_stack( plugin_skill_snapshots: Option<&PluginSkillSnapshots>, restriction_product: Option, remote_global_catalog_active: bool, + root_scan_slots: Arc, ) -> Vec> { let skill_config_rules = skill_config_rules_from_stack(config_layer_stack); load_plugins_from_layer_stack_with_scope( @@ -129,6 +132,7 @@ pub(crate) async fn load_plugins_from_layer_stack( restriction_product, skill_config_rules: &skill_config_rules, plugin_skill_snapshots, + root_scan_slots, }, ) .await @@ -809,6 +813,7 @@ async fn load_plugin( restriction_product, skill_config_rules, plugin_skill_snapshots, + root_scan_slots, } => { loaded_plugin.manifest_name = Some(manifest.display_name().to_string()); loaded_plugin.manifest_description = manifest.description.clone(); @@ -820,6 +825,7 @@ async fn load_plugin( *restriction_product, skill_config_rules, *plugin_skill_snapshots, + Arc::clone(root_scan_slots), ) .await; let has_enabled_skills = resolved_skills.has_enabled_skills(); @@ -917,6 +923,7 @@ pub async fn load_plugin_skills( restriction_product: Option, skill_config_rules: &SkillConfigRules, plugin_skill_snapshots: Option<&PluginSkillSnapshots>, + root_scan_slots: Arc, ) -> ResolvedPluginSkills { load_plugin_skill_inventory( plugin_root, @@ -924,6 +931,7 @@ pub async fn load_plugin_skills( manifest, restriction_product, plugin_skill_snapshots, + root_scan_slots, ) .await .resolve(skill_config_rules) @@ -935,6 +943,7 @@ pub(crate) async fn load_plugin_skill_inventory( manifest: &PluginManifest, restriction_product: Option, plugin_skill_snapshots: Option<&PluginSkillSnapshots>, + root_scan_slots: Arc, ) -> PluginSkillInventory { let roots = plugin_skill_roots(plugin_root, &manifest.paths) .into_iter() @@ -947,7 +956,7 @@ pub(crate) async fn load_plugin_skill_inventory( plugin_root: Some(plugin_root.clone()), }) .collect::>(); - let outcome = load_skills_from_roots(roots, plugin_skill_snapshots).await; + let outcome = load_skills_from_roots(roots, plugin_skill_snapshots, root_scan_slots).await; let had_errors = !outcome.errors.is_empty(); let skills = outcome .skills diff --git a/codex-rs/core-plugins/src/loader_tests.rs b/codex-rs/core-plugins/src/loader_tests.rs index 8e5a76b4bf..f0835ff614 100644 --- a/codex-rs/core-plugins/src/loader_tests.rs +++ b/codex-rs/core-plugins/src/loader_tests.rs @@ -5,6 +5,7 @@ use codex_config::ConfigLayerEntry; use codex_config::ConfigLayerSource; use codex_config::ConfigRequirements; use codex_config::ConfigRequirementsToml; +use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS; use codex_plugin::PluginId; use pretty_assertions::assert_eq; use tempfile::TempDir; @@ -163,6 +164,7 @@ enabled = true /*plugin_skill_snapshots*/ None, Some(Product::Codex), /*remote_global_catalog_active*/ false, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; let hooks_only = load_plugins_from_layer_stack_with_scope( diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index b58886f97c..1cea26990a 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -70,6 +70,7 @@ use codex_core_skills::PluginSkillSnapshots; use codex_core_skills::SkillMetadata; use codex_core_skills::config_rules::SkillConfigRules; use codex_core_skills::config_rules::skill_config_rules_from_stack; +use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS; use codex_hooks::plugin_hook_declarations; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -376,6 +377,7 @@ pub struct PluginsManager { // Keep the cache auth-independent so auth changes only need to resolve capabilities again. loaded_plugins_cache: RwLock, loaded_plugins_load_semaphore: Semaphore, + skill_root_scan_slots: Arc, tool_suggest_metadata_cache: ToolSuggestMetadataCache, remote_installed_plugins_cache: RwLock>>, remote_installed_plugins_cache_refresh_state: RwLock, @@ -452,6 +454,7 @@ impl PluginsManager { non_curated_cache_refresh_state: RwLock::new(NonCuratedCacheRefreshState::default()), loaded_plugins_cache: RwLock::new(LoadedPluginsCache::default()), loaded_plugins_load_semaphore: Semaphore::new(/*permits*/ 1), + skill_root_scan_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), tool_suggest_metadata_cache: ToolSuggestMetadataCache::new(), remote_installed_plugins_cache: RwLock::new(None), remote_installed_plugins_cache_refresh_state: RwLock::new( @@ -585,6 +588,7 @@ impl PluginsManager { Some(&plugin_skill_snapshots), self.restriction_product, remote_global_catalog_active, + Arc::clone(&self.skill_root_scan_slots), ) .await; log_plugin_load_errors(&plugins); @@ -670,6 +674,7 @@ impl PluginsManager { /*plugin_skill_snapshots*/ None, self.restriction_product, self.remote_global_catalog_active(config), + Arc::clone(&self.skill_root_scan_slots), ) .await; self.resolve_loaded_plugins_for_auth(plugins) @@ -1708,7 +1713,12 @@ impl PluginsManager { ) -> Result { let fragment = self .tool_suggest_metadata_cache - .metadata_for_plugin(marketplace_name, plugin, self.restriction_product) + .metadata_for_plugin( + marketplace_name, + plugin, + self.restriction_product, + Arc::clone(&self.skill_root_scan_slots), + ) .await?; Ok(fragment.project(skill_config_rules, self.auth_mode())) } @@ -1890,6 +1900,7 @@ impl PluginsManager { &config.config_layer_stack, ), /*plugin_skill_snapshots*/ None, + Arc::clone(&self.skill_root_scan_slots), ) .await; let plugin_data_root = self.store.plugin_data_root(&plugin_id); diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index 5a3d6dfd5f..0819f0a6fb 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -1928,6 +1928,7 @@ async fn load_plugin_skills_dedupes_overlapping_manifest_roots() { /*restriction_product*/ None, &SkillConfigRules::default(), /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -5822,6 +5823,7 @@ async fn load_plugins_ignores_project_config_files() { /*plugin_skill_snapshots*/ None, Some(Product::Codex), /*remote_global_catalog_active*/ false, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; diff --git a/codex-rs/core-plugins/src/tool_suggest_metadata.rs b/codex-rs/core-plugins/src/tool_suggest_metadata.rs index 30065e6d94..d535f18a2f 100644 --- a/codex-rs/core-plugins/src/tool_suggest_metadata.rs +++ b/codex-rs/core-plugins/src/tool_suggest_metadata.rs @@ -118,6 +118,7 @@ impl ToolSuggestMetadataCache { marketplace_name: &str, plugin: &ConfiguredMarketplacePlugin, restriction_product: Option, + root_scan_slots: Arc, ) -> Result, MarketplaceError> { let artifact = PluginArtifactIdentity { plugin_id: plugin.id.clone(), @@ -138,7 +139,13 @@ impl ToolSuggestMetadataCache { } let generation = self.generation(); - let entry = load_plugin_metadata(marketplace_name, plugin, restriction_product).await; + let entry = load_plugin_metadata( + marketplace_name, + plugin, + restriction_product, + Arc::clone(&root_scan_slots), + ) + .await; if self.cache_entry_if_current(generation, artifact.clone(), entry.clone()) { return entry.map_err(MarketplaceError::InvalidPlugin); } @@ -186,6 +193,7 @@ async fn load_plugin_metadata( marketplace_name: &str, plugin: &ConfiguredMarketplacePlugin, restriction_product: Option, + root_scan_slots: Arc, ) -> ToolSuggestMetadataEntry { let plugin_id = PluginId::new(plugin.name.clone(), marketplace_name.to_string()).map_err( |err| match err { @@ -216,6 +224,7 @@ async fn load_plugin_metadata( &manifest, restriction_product, /*plugin_skill_snapshots*/ None, + root_scan_slots, ) .await; let mut mcp_server_names = diff --git a/codex-rs/core-skills/src/loader.rs b/codex-rs/core-skills/src/loader.rs index 9f496d60d7..d03e4fc12d 100644 --- a/codex-rs/core-skills/src/loader.rs +++ b/codex-rs/core-skills/src/loader.rs @@ -49,9 +49,13 @@ use std::path::Component; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use tokio::sync::Semaphore; use toml::Value as TomlValue; use tracing::error; +// TODO(anp): Tune this eight-scan limit after revisiting byte-based backpressure. +pub const MAX_CONCURRENT_ROOT_SCANS: usize = 8; + #[derive(Debug, Deserialize)] struct SkillFrontmatter { #[serde(default)] @@ -195,12 +199,13 @@ pub struct SkillRoot { pub async fn load_skills_from_roots( roots: I, plugin_skill_snapshots: Option<&crate::PluginSkillSnapshots>, + root_scan_slots: Arc, ) -> SkillLoadOutcome where I: IntoIterator + Send, I::IntoIter: Send, { - crate::root_loader::load_and_merge_skill_roots(roots, plugin_skill_snapshots) + crate::root_loader::load_and_merge_skill_roots(roots, plugin_skill_snapshots, &root_scan_slots) .boxed() .await } @@ -558,6 +563,10 @@ async fn load_skills_under_root( for warning in warnings { error!("{warning}"); } + // With no skills, there is nothing to canonicalize, parse, or namespace-qualify. + if skills.is_empty() { + return; + } let root_uri = PathUri::from_abs_path(root); let resolved_skills = futures::stream::iter(skills) .map(|skill| async move { diff --git a/codex-rs/core-skills/src/loader_tests.rs b/codex-rs/core-skills/src/loader_tests.rs index 0a4559d3ee..71985ceb0d 100644 --- a/codex-rs/core-skills/src/loader_tests.rs +++ b/codex-rs/core-skills/src/loader_tests.rs @@ -29,6 +29,8 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use tempfile::TempDir; use tokio::sync::Notify; use tokio::sync::Semaphore; @@ -44,6 +46,10 @@ struct TestConfig { struct BlockingRepoSkillRootFileSystem { inner: Arc, metadata_calls: Arc, + blocked_walk_root: Option, + blocked_walk_gate: Semaphore, + walks_started: AtomicUsize, + walk_started: Notify, } struct BlockingMetadataCalls { @@ -149,7 +155,19 @@ impl ExecutorFileSystem for BlockingRepoSkillRootFileSystem { options: WalkOptions, sandbox: Option<&'a FileSystemSandboxContext>, ) -> ExecutorFileSystemFuture<'a, WalkOutcome> { - self.inner.walk(path, options, sandbox) + self.walks_started.fetch_add(/*val*/ 1, Ordering::AcqRel); + self.walk_started.notify_waiters(); + if self.blocked_walk_root.as_ref() != Some(path) { + return self.inner.walk(path, options, sandbox); + } + Box::pin(async move { + self.blocked_walk_gate + .acquire() + .await + .expect("blocked walk gate should remain open") + .forget(); + self.inner.walk(path, options, sandbox).await + }) } fn remove<'a>( @@ -276,6 +294,7 @@ async fn load_skills_for_test(config: &TestConfig) -> SkillLoadOutcome { ) .await, /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await } @@ -463,7 +482,12 @@ async fn loads_skills_from_home_agents_dir_for_user_scope() -> anyhow::Result<() Some(&home_folder_abs), ) .await; - let outcome = load_skills_from_roots(roots, /*plugin_skill_snapshots*/ None).await; + let outcome = load_skills_from_roots( + roots, + /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), + ) + .await; assert!( outcome.errors.is_empty(), "unexpected errors: {:?}", @@ -552,6 +576,7 @@ async fn load_user_skills_root(root: &Path) -> SkillLoadOutcome { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await } @@ -1038,6 +1063,7 @@ interface: plugin_root: Some(plugin_root_abs.clone()), }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1099,6 +1125,7 @@ interface: plugin_root: Some(plugin_root.abs()), }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1275,6 +1302,7 @@ async fn loads_skills_via_symlinked_subdir_for_admin_scope() { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1361,6 +1389,7 @@ async fn system_scope_ignores_symlinked_subdir() { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; assert!( @@ -1399,6 +1428,7 @@ async fn respects_max_scan_depth_for_user_scope() { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1510,6 +1540,7 @@ async fn namespaces_plugin_skills_using_provided_namespace() { plugin_root: Some(plugin_root.abs()), }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1757,6 +1788,7 @@ async fn plugin_skill_name_length_limit_allows_max_qualified_name() { plugin_root: Some(plugin_root.abs()), }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -1806,6 +1838,7 @@ async fn plugin_skill_name_length_limit_rejects_overlong_qualified_name() { plugin_root: Some(plugin_root.abs()), }], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; @@ -2234,6 +2267,10 @@ async fn repo_skill_root_search_limits_concurrent_probes_and_preserves_order() { let fs: Arc = Arc::new(BlockingRepoSkillRootFileSystem { inner: Arc::clone(&LOCAL_FS), metadata_calls: Arc::clone(&metadata_calls), + blocked_walk_root: None, + blocked_walk_gate: Semaphore::new(/*permits*/ 0), + walks_started: AtomicUsize::new(/*v*/ 0), + walk_started: Notify::new(), }); let assertions = async { @@ -2303,6 +2340,130 @@ async fn repo_skill_root_search_limits_concurrent_probes_and_preserves_order() { ); } +#[tokio::test] +async fn merges_root_results_in_input_order_when_scans_finish_out_of_order() { + const ROOT_COUNT: usize = MAX_CONCURRENT_ROOT_SCANS + 1; + + let temp = tempfile::tempdir().expect("tempdir"); + let roots = (0..ROOT_COUNT) + .map(|index| temp.path().join(format!("root-{index}"))) + .collect::>(); + for root in &roots { + fs::create_dir_all(root).expect("create root"); + } + let first_skill = roots[0].join("broken/SKILL.md"); + let second_skill = roots[1].join("broken/SKILL.md"); + for (path, contents) in [ + (&first_skill, "missing frontmatter"), + (&second_skill, "also missing frontmatter"), + ] { + fs::create_dir_all(path.parent().expect("skill parent")).expect("create skill directory"); + fs::write(path, contents).expect("write skill"); + } + + let blocked_walk_root = PathUri::from_abs_path(&roots[0].abs()); + let file_system = Arc::new(BlockingRepoSkillRootFileSystem { + inner: Arc::clone(&LOCAL_FS), + metadata_calls: Arc::new(BlockingMetadataCalls::default()), + blocked_walk_root: Some(blocked_walk_root), + blocked_walk_gate: Semaphore::new(/*permits*/ 0), + walks_started: AtomicUsize::new(/*v*/ 0), + walk_started: Notify::new(), + }); + let root_file_system: Arc = file_system.clone(); + let skill_roots = roots + .iter() + .enumerate() + .map(|(index, root)| SkillRoot { + path: root.abs(), + scope: if index == 0 { + SkillScope::Repo + } else { + SkillScope::User + }, + file_system: Arc::clone(&root_file_system), + plugin_id: None, + plugin_namespace: Some("test".to_string()), + plugin_root: None, + }) + .collect::>(); + let root_scan_slots = Semaphore::new(MAX_CONCURRENT_ROOT_SCANS); + let load = tokio::spawn(async move { + crate::root_loader::load_and_merge_skill_roots( + skill_roots, + /*plugin_skill_snapshots*/ None, + &root_scan_slots, + ) + .await + }); + + tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), async { + loop { + let started = file_system.walk_started.notified(); + if file_system.walks_started.load(Ordering::Acquire) == ROOT_COUNT { + break; + } + started.await; + } + }) + .await + .expect("all skill-root walks should start despite the blocked first root"); + file_system.blocked_walk_gate.add_permits(/*n*/ 1); + let outcome = load.await.expect("skill-root load should finish"); + + assert_eq!(outcome.skills, Vec::new()); + assert_eq!( + outcome.errors, + vec![ + SkillError { + path: canonicalize_path(first_skill) + .expect("canonical first skill") + .abs(), + message: "missing YAML frontmatter delimited by ---".to_string(), + }, + SkillError { + path: canonicalize_path(second_skill) + .expect("canonical second skill") + .abs(), + message: "missing YAML frontmatter delimited by ---".to_string(), + }, + ] + ); +} + +#[tokio::test] +async fn skill_root_scans_wait_for_shared_capacity() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("root"); + fs::create_dir_all(&root).expect("create root"); + let root_scan_slots = Semaphore::new(MAX_CONCURRENT_ROOT_SCANS); + let held_slots = root_scan_slots + .try_acquire_many( + u32::try_from(MAX_CONCURRENT_ROOT_SCANS).expect("root scan limit should fit in u32"), + ) + .expect("root scan slots should be available"); + let load = crate::root_loader::load_and_merge_skill_roots( + [SkillRoot { + path: root.abs(), + scope: SkillScope::Repo, + file_system: Arc::clone(&LOCAL_FS), + plugin_id: None, + plugin_namespace: Some("test".to_string()), + plugin_root: None, + }], + /*plugin_skill_snapshots*/ None, + &root_scan_slots, + ); + tokio::pin!(load); + + assert!(futures::poll!(load.as_mut()).is_pending()); + drop(held_slots); + let outcome = load.await; + + assert_eq!(outcome.skills, Vec::new()); + assert_eq!(outcome.errors, Vec::new()); +} + #[tokio::test] async fn loads_skills_from_codex_dir_when_not_git_repo() { let codex_home = tempfile::tempdir().expect("tempdir"); @@ -2368,6 +2529,7 @@ async fn deduplicates_by_path_preferring_first_root() { }, ], /*plugin_skill_snapshots*/ None, + Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; diff --git a/codex-rs/core-skills/src/root_loader.rs b/codex-rs/core-skills/src/root_loader.rs index a31c8a2753..f15887d513 100644 --- a/codex-rs/core-skills/src/root_loader.rs +++ b/codex-rs/core-skills/src/root_loader.rs @@ -5,8 +5,11 @@ use std::sync::Arc; use std::sync::Mutex; use codex_utils_plugins::PluginSkillRoot; +use futures::StreamExt; +use tokio::sync::Semaphore; use crate::SkillLoadOutcome; +use crate::loader::MAX_CONCURRENT_ROOT_SCANS; use crate::loader::SkillRoot; use crate::loader::SkillRootSnapshot; use crate::loader::load_skill_root; @@ -40,52 +43,71 @@ impl fmt::Debug for PluginSkillSnapshots { pub(crate) async fn load_and_merge_skill_roots( roots: I, plugin_skill_snapshots: Option<&PluginSkillSnapshots>, + root_scan_slots: &Semaphore, ) -> SkillLoadOutcome where I: IntoIterator, { - let mut root_snapshots = Vec::new(); - for root in roots { - let cache_key = match ( - root.plugin_id.clone(), - root.plugin_namespace.clone(), - root.plugin_root.clone(), - ) { - (Some(plugin_id), Some(plugin_namespace), Some(plugin_root)) => Some(PluginSkillRoot { - path: root.path.clone(), - plugin_id, - plugin_namespace, - plugin_root, - }), - _ => None, - }; - let cached_snapshot = cache_key.as_ref().and_then(|cache_key| { - let plugin_skill_snapshots = plugin_skill_snapshots?; - plugin_skill_snapshots - .snapshots_by_root - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(cache_key) - .cloned() - }); - let snapshot = match cached_snapshot { - Some(snapshot) => snapshot, - None => { - let snapshot = load_skill_root(root).await; - if let Some(plugin_skill_snapshots) = plugin_skill_snapshots - && let Some(cache_key) = cache_key - { - plugin_skill_snapshots - .snapshots_by_root - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert(cache_key, snapshot.clone()); + let mut indexed_root_snapshots = futures::stream::iter(roots.into_iter().enumerate()) + .map(|(root_index, root)| async move { + // Bound root scans across all concurrent loads sharing this pool. + let _root_scan_slot = root_scan_slots + .acquire() + .await + .unwrap_or_else(|_| unreachable!()); + let cache_key = match ( + root.plugin_id.clone(), + root.plugin_namespace.clone(), + root.plugin_root.clone(), + ) { + (Some(plugin_id), Some(plugin_namespace), Some(plugin_root)) => { + Some(PluginSkillRoot { + path: root.path.clone(), + plugin_id, + plugin_namespace, + plugin_root, + }) } - snapshot - } - }; - root_snapshots.push(snapshot); - } + _ => None, + }; + let cached_snapshot = cache_key.as_ref().and_then(|cache_key| { + let plugin_skill_snapshots = plugin_skill_snapshots?; + plugin_skill_snapshots + .snapshots_by_root + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(cache_key) + .cloned() + }); + + let snapshot = match cached_snapshot { + Some(snapshot) => snapshot, + None => { + let snapshot = load_skill_root(root).await; + if let Some(plugin_skill_snapshots) = plugin_skill_snapshots + && let Some(cache_key) = cache_key + { + plugin_skill_snapshots + .snapshots_by_root + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(cache_key, snapshot.clone()); + } + snapshot + } + }; + (root_index, snapshot) + }) + // Keep each load's scan queue bounded while avoiding head-of-line blocking. + .buffer_unordered(MAX_CONCURRENT_ROOT_SCANS) + .collect::>() + .await; + // Keep every scan slot productive, then restore root precedence for deterministic merging. + indexed_root_snapshots.sort_unstable_by_key(|(root_index, _)| *root_index); + let root_snapshots = indexed_root_snapshots + .into_iter() + .map(|(_, snapshot)| snapshot) + .collect(); merge_skill_root_snapshots(root_snapshots) } diff --git a/codex-rs/core-skills/src/service.rs b/codex-rs/core-skills/src/service.rs index b4e0d7396c..60c8c87485 100644 --- a/codex-rs/core-skills/src/service.rs +++ b/codex-rs/core-skills/src/service.rs @@ -9,6 +9,7 @@ use codex_protocol::protocol::Product; use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_plugins::PluginSkillRoot; +use tokio::sync::Semaphore; use tracing::info; use tracing::instrument; use tracing::warn; @@ -20,6 +21,7 @@ use crate::build_implicit_skill_path_indexes; use crate::config_rules::SkillConfigRules; use crate::config_rules::resolve_disabled_skill_paths; use crate::config_rules::skill_config_rules_from_stack; +use crate::loader::MAX_CONCURRENT_ROOT_SCANS; use crate::loader::SkillRoot; use crate::loader::load_skills_from_roots; use crate::loader::skill_roots; @@ -71,6 +73,8 @@ pub struct SkillsService { extra_roots: RwLock>, cache_by_cwd: RwLock>, cache_by_config: RwLock>, + // Shared across cwds so root scheduling cannot multiply per-root I/O fanout. + root_scan_slots: Arc, } impl SkillsService { @@ -89,6 +93,7 @@ impl SkillsService { extra_roots: RwLock::new(Vec::new()), cache_by_cwd: RwLock::new(HashMap::new()), cache_by_config: RwLock::new(HashMap::new()), + root_scan_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), }; if !bundled_skills_enabled { // The loader caches bundled skills under `skills/.system`. Clearing that directory is @@ -213,7 +218,12 @@ impl SkillsService { roots: Vec, skill_config_rules: &SkillConfigRules, ) -> SkillLoadOutcome { - let outcome = load_skills_from_roots(roots, input.plugin_skill_snapshots.as_ref()).await; + let outcome = load_skills_from_roots( + roots, + input.plugin_skill_snapshots.as_ref(), + Arc::clone(&self.root_scan_slots), + ) + .await; let outcome = crate::filter_skill_load_outcome_for_product(outcome, self.restriction_product); let disabled_paths = resolve_disabled_skill_paths(&outcome.skills, skill_config_rules); diff --git a/codex-rs/core-skills/tests/environment_loader.rs b/codex-rs/core-skills/tests/environment_loader.rs index a2c9903aea..aef4fb0af0 100644 --- a/codex-rs/core-skills/tests/environment_loader.rs +++ b/codex-rs/core-skills/tests/environment_loader.rs @@ -420,6 +420,7 @@ async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() { use codex_core_skills::SkillMetadata; use codex_core_skills::SkillPolicy; + use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS; use codex_core_skills::loader::SkillRoot; use codex_core_skills::loader::load_skills_from_roots; use codex_protocol::protocol::SkillScope; @@ -469,6 +470,7 @@ async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ); fn assert_send(_: &T) {} assert_send(&future); diff --git a/codex-rs/ext/skills/tests/executor_file_system_authority.rs b/codex-rs/ext/skills/tests/executor_file_system_authority.rs index 06bf062eac..83445c745c 100644 --- a/codex-rs/ext/skills/tests/executor_file_system_authority.rs +++ b/codex-rs/ext/skills/tests/executor_file_system_authority.rs @@ -4,6 +4,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use codex_core_skills::HostSkillsSnapshot; +use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS; use codex_core_skills::loader::SkillRoot; use codex_core_skills::loader::load_skills_from_roots; use codex_exec_server::CopyOptions; @@ -211,6 +212,7 @@ async fn skill_loading_and_reads_use_the_supplied_executor_file_system() { plugin_root: None, }], /*plugin_skill_snapshots*/ None, + Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), ) .await; assert_eq!(outcome.errors, Vec::new());