From c6942e78c5ee57abec21686e9d165cd3a9dd8499 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 08:46:52 +0000 Subject: [PATCH] perf(skills): prioritize agents skill roots (cherry picked from commit 1225c729e5092ca26c8182a072afed5f7120bba1) --- codex-rs/core-skills/src/loader.rs | 8 ++ codex-rs/core-skills/src/root_loader.rs | 22 +++- codex-rs/core-skills/src/root_loader_tests.rs | 103 +++++++++++++++++- 3 files changed, 126 insertions(+), 7 deletions(-) diff --git a/codex-rs/core-skills/src/loader.rs b/codex-rs/core-skills/src/loader.rs index ed2b657cab..60f2634024 100644 --- a/codex-rs/core-skills/src/loader.rs +++ b/codex-rs/core-skills/src/loader.rs @@ -189,6 +189,14 @@ pub struct SkillRoot { pub plugin_root: Option, } +impl SkillRoot { + pub(crate) fn is_agents_skills_root(&self) -> bool { + self.path.file_name() == Some(std::ffi::OsStr::new(SKILLS_DIR_NAME)) + && self.path.parent().is_some_and(|parent| { + parent.file_name() == Some(std::ffi::OsStr::new(AGENTS_DIR_NAME)) + }) + } +} pub async fn load_skills_from_roots( roots: I, plugin_skill_snapshots: Option<&crate::PluginSkillSnapshots>, diff --git a/codex-rs/core-skills/src/root_loader.rs b/codex-rs/core-skills/src/root_loader.rs index df3c3980dd..8bb1214ee9 100644 --- a/codex-rs/core-skills/src/root_loader.rs +++ b/codex-rs/core-skills/src/root_loader.rs @@ -15,11 +15,18 @@ use crate::model::SkillFileSystemsByPath; const MAX_CONCURRENT_ROOT_SCANS: usize = 4; +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +enum RootScanPriority { + AgentsSkills, + Default, +} + struct PendingRootLoad { root: SkillRoot, root_index: usize, duplicate_root_indices: Vec, cache_key: Option, + scheduling_priority: RootScanPriority, } struct CompletedRootLoad { @@ -95,16 +102,26 @@ where { pending_load_by_cache_key.insert(cache_key.clone(), pending_loads.len()); } + let scheduling_priority = if root.is_agents_skills_root() { + RootScanPriority::AgentsSkills + } else { + RootScanPriority::Default + }; pending_loads.push(PendingRootLoad { root, root_index, duplicate_root_indices: Vec::new(), cache_key, + scheduling_priority, }); } - // Unordered buffering lets fast host and cached roots free slots while an earlier executor - // root is still pending. The indexed snapshots below restore the original merge order. + // Group duplicate cache keys before prioritization so the original first occurrence remains + // the scan/cache owner. Sorting by the original index within each priority retains FIFO order. + pending_loads.sort_by_key(|pending| (pending.scheduling_priority, pending.root_index)); + // Unordered buffering starts known `.agents/skills` roots first and lets fast host and cached + // roots free slots while an earlier executor root is still pending. The indexed snapshots + // below restore the original merge order. let completed_loads = futures::stream::iter(pending_loads) .map(|pending| async move { let PendingRootLoad { @@ -112,6 +129,7 @@ where root_index, duplicate_root_indices, cache_key, + scheduling_priority: _, } = pending; let cached_snapshot = cache_key.as_ref().and_then(|cache_key| { let plugin_skill_snapshots = plugin_skill_snapshots?; diff --git a/codex-rs/core-skills/src/root_loader_tests.rs b/codex-rs/core-skills/src/root_loader_tests.rs index fef8dfd2de..bad9ebaf45 100644 --- a/codex-rs/core-skills/src/root_loader_tests.rs +++ b/codex-rs/core-skills/src/root_loader_tests.rs @@ -2,6 +2,7 @@ use std::fs; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use std::sync::Mutex; use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -37,6 +38,7 @@ const TEST_WAIT: Duration = Duration::from_secs(/*secs*/ 5); struct ControlledFileSystem { inner: Arc, walk_gate: Option>, + walk_paths: Mutex>, walks_started: AtomicUsize, walk_started: Notify, blocked_read_root: Option, @@ -51,6 +53,7 @@ impl ControlledFileSystem { Self { inner, walk_gate: None, + walk_paths: Mutex::new(Vec::new()), walks_started: AtomicUsize::new(/*v*/ 0), walk_started: Notify::new(), blocked_read_root: None, @@ -84,6 +87,13 @@ impl ControlledFileSystem { self.walks_started.load(Ordering::Acquire) } + fn walk_paths(&self) -> Vec { + self.walk_paths + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + async fn wait_for_walks(&self, expected: usize) { tokio::time::timeout(TEST_WAIT, async { loop { @@ -203,6 +213,10 @@ impl ExecutorFileSystem for ControlledFileSystem { options: WalkOptions, sandbox: Option<&'a FileSystemSandboxContext>, ) -> ExecutorFileSystemFuture<'a, WalkOutcome> { + self.walk_paths + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(path.clone()); self.walks_started.fetch_add(/*val*/ 1, Ordering::AcqRel); self.walk_started.notify_waiters(); Box::pin(async move { @@ -287,7 +301,13 @@ async fn loads_roots_unordered_but_merges_errors_in_input_order() { let temp = tempfile::tempdir().expect("tempdir"); let roots = (0..ROOT_COUNT) - .map(|index| temp.path().join(format!("root-{index}"))) + .map(|index| { + if index == 1 { + temp.path().join("root-1").join(".agents").join("skills") + } else { + temp.path().join(format!("root-{index}")) + } + }) .collect::>(); for root in &roots { fs::create_dir_all(root).expect("create root"); @@ -347,6 +367,57 @@ async fn loads_roots_unordered_but_merges_errors_in_input_order() { ); } +#[tokio::test] +async fn starts_agents_skill_roots_before_other_pending_scans() { + let temp = tempfile::tempdir().expect("tempdir"); + let default_roots = (0..MAX_CONCURRENT_ROOT_SCANS) + .map(|index| temp.path().join(format!("default-{index}"))) + .collect::>(); + let agents_roots = (0..MAX_CONCURRENT_ROOT_SCANS) + .map(|index| { + temp.path() + .join(format!("repo-{index}")) + .join(".agents") + .join("skills") + }) + .collect::>(); + for root in default_roots.iter().chain(&agents_roots) { + fs::create_dir_all(root).expect("create root"); + } + + let walk_gate = Arc::new(Semaphore::new(/*permits*/ 0)); + let file_system = Arc::new( + ControlledFileSystem::new(Arc::clone(&LOCAL_FS)).with_walk_gate(Arc::clone(&walk_gate)), + ); + let root_file_system: Arc = file_system.clone(); + let skill_roots = default_roots + .iter() + .chain(&agents_roots) + .map(|root| plain_root(root, SkillScope::Repo, Arc::clone(&root_file_system))) + .collect::>(); + + let load = tokio::spawn(async move { + load_and_merge_skill_roots(skill_roots, /*plugin_skill_snapshots*/ None).await + }); + file_system.wait_for_walks(MAX_CONCURRENT_ROOT_SCANS).await; + assert_eq!( + file_system.walk_paths(), + agents_roots + .iter() + .map(|root| PathUri::from_host_native_path(root).expect("agents root")) + .collect::>() + ); + + walk_gate.add_permits(MAX_CONCURRENT_ROOT_SCANS); + file_system + .wait_for_walks(MAX_CONCURRENT_ROOT_SCANS * 2) + .await; + walk_gate.add_permits(MAX_CONCURRENT_ROOT_SCANS); + let outcome = load.await.expect("skill-root load should finish"); + assert_eq!(outcome.skills, Vec::new()); + assert_eq!(outcome.errors, Vec::new()); +} + #[tokio::test] async fn limits_concurrent_root_scans() { const ROOT_COUNT: usize = MAX_CONCURRENT_ROOT_SCANS + 1; @@ -388,19 +459,26 @@ async fn limits_concurrent_root_scans() { #[tokio::test] async fn duplicate_plugin_roots_share_work_only_when_snapshots_are_available() { let temp = tempfile::tempdir().expect("tempdir"); - let skills_root = temp.path().join("skills"); + let skills_root = temp.path().join("plugin").join(".agents").join("skills"); write_skill( &skills_root, "demo", "---\nname: demo\ndescription: demo skill\n---\n", ); let broken_skill = write_skill(&skills_root, "broken", "missing frontmatter"); + let default_root = temp.path().join("default-skills"); + let default_broken_skill = write_skill(&default_root, "broken", "default missing frontmatter"); let cached_file_system = Arc::new(ControlledFileSystem::new(Arc::clone(&LOCAL_FS))); let cached_root_file_system: Arc = cached_file_system.clone(); let snapshots = PluginSkillSnapshots::for_plugin_load(); let cached_outcome = load_and_merge_skill_roots( [ + plain_root( + &default_root, + SkillScope::User, + Arc::clone(&cached_root_file_system), + ), plugin_root( &skills_root, temp.path(), @@ -415,12 +493,17 @@ async fn duplicate_plugin_roots_share_work_only_when_snapshots_are_available() { Some(&snapshots), ) .await; - assert_eq!(cached_file_system.walks_started(), 1); + assert_eq!(cached_file_system.walks_started(), 2); let uncached_file_system = Arc::new(ControlledFileSystem::new(Arc::clone(&LOCAL_FS))); let uncached_root_file_system: Arc = uncached_file_system.clone(); let uncached_outcome = load_and_merge_skill_roots( [ + plain_root( + &default_root, + SkillScope::User, + Arc::clone(&uncached_root_file_system), + ), plugin_root( &skills_root, temp.path(), @@ -435,7 +518,7 @@ async fn duplicate_plugin_roots_share_work_only_when_snapshots_are_available() { /*plugin_skill_snapshots*/ None, ) .await; - assert_eq!(uncached_file_system.walks_started(), 2); + assert_eq!(uncached_file_system.walks_started(), 3); assert_eq!(cached_outcome.skills, uncached_outcome.skills); assert_eq!(cached_outcome.errors, uncached_outcome.errors); @@ -446,8 +529,18 @@ async fn duplicate_plugin_roots_share_work_only_when_snapshots_are_available() { .abs(), message: "missing YAML frontmatter delimited by ---".to_string(), }; + let expected_default_error = SkillError { + path: dunce::canonicalize(default_broken_skill) + .expect("canonical default broken skill") + .abs(), + message: "missing YAML frontmatter delimited by ---".to_string(), + }; assert_eq!( cached_outcome.errors, - vec![expected_error.clone(), expected_error] + vec![ + expected_default_error, + expected_error.clone(), + expected_error, + ] ); }