From e8eb60092a95fe87bfa52bc261b2cb85731e427e Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 9 Jul 2026 09:54:24 +0100 Subject: [PATCH] perf(skills): reuse walk inventory for host loading (#31566) ## Why CCA thread startup loads repository skills through the primary remote environment. This still used the older host skill loader, which recursively issued directory and metadata requests over the exec-server connection. On a high-latency connection, one skill scan could turn into hundreds of round trips. Executor-selected skill roots already avoid this pattern: they use one bounded filesystem walk, reuse the returned inventory, and overlap independent reads. ## What changed - Extract the URI-native walk inventory into a shared private discovery module. - Use that discovery path for both environment-owned and host/repository skill roots. - Reuse the walk inventory to avoid per-skill metadata existence checks when the inventory is complete. - Canonicalize host skill identities and parse skill files with bounded concurrency. - Prune hidden host directories during traversal, preserving visible aliases and walk limits. - Cache namespace probes by ancestor and keep all remote namespace reads under the same concurrency bound. - Keep safe metadata probes for incomplete walks, file symlinks, and case aliases. ## Dependency #31570 is now merged. It provides the optional `pruneHiddenDirectories` walk flag used here. That flag defaults to false and is omitted when disabled. No other exec-server API changes are introduced. --- codex-rs/core-skills/src/loader.rs | 421 +++++++----------- codex-rs/core-skills/src/loader/discovery.rs | 207 +++++++++ .../core-skills/src/loader/environment.rs | 161 +------ codex-rs/core-skills/src/loader/namespace.rs | 106 +++-- codex-rs/core-skills/src/loader_tests.rs | 68 ++- .../core-skills/tests/environment_loader.rs | 97 ++++ .../suite/unified_exec_process_events.rs | 10 + 7 files changed, 596 insertions(+), 474 deletions(-) create mode 100644 codex-rs/core-skills/src/loader/discovery.rs diff --git a/codex-rs/core-skills/src/loader.rs b/codex-rs/core-skills/src/loader.rs index e74eb01a0c..aa663e3e60 100644 --- a/codex-rs/core-skills/src/loader.rs +++ b/codex-rs/core-skills/src/loader.rs @@ -1,3 +1,4 @@ +mod discovery; mod environment; mod namespace; @@ -26,15 +27,21 @@ use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::AbsolutePathBufGuard; use codex_utils_path_uri::PathUri; -use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use codex_utils_plugins::PluginSkillRoot; use dirs::home_dir; -use futures::future::join_all; -use namespace::ResolvedSkillNamespace; +use discovery::DirectorySymlinkPolicy; +use discovery::DiscoveredSkill; +use discovery::HiddenDirectoryPolicy; +use discovery::MAX_CONCURRENT_SKILL_LOADS; +use discovery::SkillDiscovery; +use discovery::SkillDiscoveryOptions; +use discovery::SkillMetadataDiscovery; +use discovery::discover_skills; +use futures::FutureExt; +use futures::StreamExt; use namespace::SkillNamespaceResolver; use serde::Deserialize; use std::collections::HashSet; -use std::collections::VecDeque; use std::error::Error; use std::fmt; use std::io; @@ -140,17 +147,10 @@ const MAX_DEPENDENCY_URL_LEN: usize = MAX_DESCRIPTION_LEN; const MAX_SCAN_DEPTH: usize = 6; const MAX_SKILLS_DIRS_PER_ROOT: usize = 2000; -#[derive(Clone, Copy)] -enum SymlinkPolicy { - FollowDirectories, - Ignore, -} - -struct SkillFileDiscovery { - skill_files: Vec, - plugin_roots: HashSet, - namespace_roots: HashSet, - warnings: Vec, +struct ResolvedDiscoveredSkill { + skill: DiscoveredSkill, + path: AbsolutePathBuf, + path_uri: PathUri, } #[derive(Debug)] @@ -194,9 +194,12 @@ pub async fn load_skills_from_roots( plugin_skill_snapshots: Option<&crate::PluginSkillSnapshots>, ) -> SkillLoadOutcome where - I: IntoIterator, + I: IntoIterator + Send, + I::IntoIter: Send, { - crate::root_loader::load_and_merge_skill_roots(roots, plugin_skill_snapshots).await + crate::root_loader::load_and_merge_skill_roots(roots, plugin_skill_snapshots) + .boxed() + .await } #[derive(Clone)] @@ -497,172 +500,6 @@ async fn canonicalize_for_skill_identity( .unwrap_or_else(|_| path.clone()) } -async fn discover_skills_under_root( - fs: &dyn ExecutorFileSystem, - root: &PathUri, - symlink_policy: SymlinkPolicy, -) -> SkillFileDiscovery { - let root = root.clone(); - let mut discovery = SkillFileDiscovery { - skill_files: Vec::new(), - plugin_roots: HashSet::new(), - namespace_roots: HashSet::from([root.clone()]), - warnings: Vec::new(), - }; - match fs.get_metadata(&root, /*sandbox*/ None).await { - Ok(metadata) if metadata.is_directory => {} - Ok(_) => return discovery, - Err(err) if err.kind() == io::ErrorKind::NotFound => return discovery, - Err(err) => { - discovery - .warnings - .push(format!("failed to stat skills root {root}: {err:#}")); - return discovery; - } - } - - fn enqueue_dir( - queue: &mut VecDeque<(PathUri, usize)>, - visited_dirs: &mut HashSet, - truncated_by_dir_limit: &mut bool, - path: PathUri, - depth: usize, - ) { - if depth > MAX_SCAN_DEPTH { - return; - } - if visited_dirs.len() >= MAX_SKILLS_DIRS_PER_ROOT { - *truncated_by_dir_limit = true; - return; - } - if visited_dirs.insert(path.clone()) { - queue.push_back((path, depth)); - } - } - - let follow_symlinks = matches!(symlink_policy, SymlinkPolicy::FollowDirectories); - let mut visited_dirs: HashSet = HashSet::from([root.clone()]); - let mut queue: VecDeque<(PathUri, usize)> = VecDeque::from([(root.clone(), 0)]); - let mut truncated_by_dir_limit = false; - - while let Some((dir, depth)) = queue.pop_front() { - let entries = match fs.read_directory(&dir, /*sandbox*/ None).await { - Ok(entries) => entries, - Err(err) => { - discovery - .warnings - .push(format!("failed to read skills directory {dir}: {err:#}")); - continue; - } - }; - - let paths = entries - .into_iter() - .filter_map(|entry| { - let file_name = entry.file_name; - if DISCOVERABLE_PLUGIN_MANIFEST_PATHS - .iter() - .any(|path| path.split('/').next() == Some(file_name.as_str())) - { - discovery.plugin_roots.insert(dir.clone()); - } - if file_name.starts_with('.') { - return None; - } - match dir.join(&file_name) { - Ok(path) => Some((file_name, path)), - Err(err) => { - discovery.warnings.push(format!( - "failed to resolve skill path {dir}/{file_name}: {err}" - )); - None - } - } - }) - .collect::>(); - let metadata_results = join_all( - paths - .iter() - .map(|(_, path)| fs.get_metadata(path, /*sandbox*/ None)), - ) - .await; - - for ((file_name, path), metadata_result) in paths.into_iter().zip(metadata_results) { - let metadata = match metadata_result { - Ok(metadata) => metadata, - Err(err) => { - discovery - .warnings - .push(format!("failed to stat skill path {path}: {err:#}")); - continue; - } - }; - - if metadata.is_symlink { - if !follow_symlinks { - continue; - } - match fs.read_directory(&path, /*sandbox*/ None).await { - Ok(_) => { - let resolved_dir = canonicalize_uri_for_skill_identity(fs, &path).await; - discovery.namespace_roots.insert(resolved_dir.clone()); - enqueue_dir( - &mut queue, - &mut visited_dirs, - &mut truncated_by_dir_limit, - resolved_dir, - depth + 1, - ); - } - Err(err) - if matches!( - err.kind(), - io::ErrorKind::NotADirectory | io::ErrorKind::NotFound - ) => {} - Err(err) => discovery.warnings.push(format!( - "failed to read skills symlink directory {path}: {err:#}" - )), - } - continue; - } - - if metadata.is_directory { - enqueue_dir( - &mut queue, - &mut visited_dirs, - &mut truncated_by_dir_limit, - path, - depth + 1, - ); - continue; - } - - if metadata.is_file && file_name == SKILLS_FILENAME { - discovery.skill_files.push(path); - } - } - } - - if truncated_by_dir_limit { - tracing::warn!( - "skills scan truncated after {} directories (root: {})", - MAX_SKILLS_DIRS_PER_ROOT, - root - ); - } - discovery -} - -async fn canonicalize_uri_for_skill_identity( - file_system: &dyn ExecutorFileSystem, - path: &PathUri, -) -> PathUri { - file_system - .canonicalize(path, /*sandbox*/ None) - .await - .unwrap_or_else(|_| path.clone()) -} - async fn load_skills_under_root( fs: &dyn ExecutorFileSystem, root: &AbsolutePathBuf, @@ -676,56 +513,113 @@ async fn load_skills_under_root( Some(plugin_root) => Some(canonicalize_for_skill_identity(fs, plugin_root).await), None => None, }; - let symlink_policy = match scope { - SkillScope::User | SkillScope::Repo | SkillScope::Admin => SymlinkPolicy::FollowDirectories, - SkillScope::System => SymlinkPolicy::Ignore, + let directory_symlinks = match scope { + SkillScope::User | SkillScope::Repo | SkillScope::Admin => DirectorySymlinkPolicy::Follow, + SkillScope::System => DirectorySymlinkPolicy::Ignore, }; - let SkillFileDiscovery { - skill_files, + let SkillDiscovery { + skills, plugin_roots, - namespace_roots, + mut namespace_roots, warnings, - } = discover_skills_under_root(fs, &PathUri::from_abs_path(root), symlink_policy).await; + } = discover_skills( + fs, + &PathUri::from_abs_path(root), + // Preserve host discovery behavior: directory aliases are scope-dependent, while hidden + // directories are skipped unless reached through a visible alias. + SkillDiscoveryOptions { + directory_symlinks, + hidden_directories: HiddenDirectoryPolicy::Skip, + }, + ) + .await; for warning in warnings { error!("{warning}"); } let root_uri = PathUri::from_abs_path(root); - let namespace_resolver = match plugin_namespace { - Some(namespace) => SkillNamespaceResolver::with_provided_namespace(namespace), - None => { - SkillNamespaceResolver::discover( - fs, - &root_uri, - &skill_files, - plugin_roots, - namespace_roots, - ) - .await + let resolved_skills = futures::stream::iter(skills) + .map(|skill| async move { + let path_uri = fs + .canonicalize(&skill.path, /*sandbox*/ None) + .await + .unwrap_or_else(|_| skill.path.clone()); + let path = match path_uri.to_abs_path() { + Ok(path) => path, + Err(err) => { + error!("failed to convert discovered skill path {path_uri}: {err}"); + return None; + } + }; + Some(ResolvedDiscoveredSkill { + skill, + path, + path_uri, + }) + }) + .buffered(MAX_CONCURRENT_SKILL_LOADS) + .filter_map(futures::future::ready) + .collect::>() + .await; + namespace_roots.extend(resolved_skills.iter().filter_map(|skill| { + (skill.path_uri != skill.skill.path) + .then(|| skill.path_uri.parent()) + .flatten() + })); + let skill_paths = resolved_skills + .iter() + .map(|skill| skill.path_uri.clone()) + .collect::>(); + let namespace_resolver = async { + match plugin_namespace { + Some(namespace) => SkillNamespaceResolver::with_provided_namespace(namespace), + None => { + SkillNamespaceResolver::discover( + fs, + &root_uri, + &skill_paths, + plugin_roots, + namespace_roots, + ) + .await + } } }; - for path_uri in skill_files { - let path = match path_uri.to_abs_path() { - Ok(path) => path, - Err(err) => { - error!("failed to convert discovered skill path {path_uri}: {err}"); - continue; + let skill_results = futures::stream::iter(resolved_skills) + .map(|skill| { + let plugin_root = plugin_root.as_ref(); + async move { + let result = parse_skill_file( + fs, + &skill.skill, + &skill.path, + &skill.path_uri, + scope, + plugin_id, + plugin_root, + ) + .await + .map_err(|err| err.to_string()); + (skill.path, skill.path_uri, result) } - }; - match parse_skill_file( - fs, - &path, - scope, - plugin_id, - namespace_resolver.for_skill(&root_uri, &path_uri), - plugin_root.as_ref(), - ) - .await - { + }) + .buffered(MAX_CONCURRENT_SKILL_LOADS) + .collect::>() + .boxed(); + let (namespace_resolver, skill_results) = tokio::join!(namespace_resolver, skill_results); + for (path, path_uri, result) in skill_results { + let result = result.and_then(|mut skill| { + skill.name = namespace_resolver + .for_skill(&root_uri, &path_uri) + .qualify(&skill.name); + validate_len(&skill.name, MAX_QUALIFIED_NAME_LEN, "qualified name") + .map_err(|err| err.to_string())?; + Ok(skill) + }); + match result { Ok(skill) => outcome.skills.push(skill), - Err(err) if scope != SkillScope::System => outcome.errors.push(SkillError { - path, - message: err.to_string(), - }), + Err(err) if scope != SkillScope::System => { + outcome.errors.push(SkillError { path, message: err }) + } Err(_) => {} } } @@ -733,41 +627,47 @@ async fn load_skills_under_root( async fn parse_skill_file( fs: &dyn ExecutorFileSystem, + skill: &DiscoveredSkill, path: &AbsolutePathBuf, + path_uri: &PathUri, scope: SkillScope, plugin_id: Option<&str>, - namespace: &ResolvedSkillNamespace, plugin_root: Option<&AbsolutePathBuf>, ) -> Result { - let path_uri = PathUri::from_abs_path(path); - let contents = fs - .read_file_text(&path_uri, /*sandbox*/ None) - .await - .map_err(SkillParseError::Read)?; + let metadata_path = path_uri + .parent() + .and_then(|parent| parent.join(SKILLS_METADATA_DIR).ok()) + .and_then(|directory| directory.join(SKILLS_METADATA_FILENAME).ok()); + let metadata = match &skill.metadata { + SkillMetadataDiscovery::Present(_) => metadata_path.map(SkillMetadataDiscovery::Present), + SkillMetadataDiscovery::Probe(_) => metadata_path.map(SkillMetadataDiscovery::Probe), + SkillMetadataDiscovery::Absent => None, + } + .unwrap_or(SkillMetadataDiscovery::Absent); + let (contents, loaded_metadata) = tokio::join!( + fs.read_file_text(path_uri, /*sandbox*/ None), + load_skill_metadata(fs, path, &metadata, plugin_root), + ); + let contents = contents.map_err(SkillParseError::Read)?; let ParsedSkillFrontmatter { name: base_name, description, short_description, } = parse_skill_frontmatter_metadata_inner(&contents, || default_skill_name(path))?; - let name = namespace.qualify(&base_name); let LoadedSkillMetadata { interface, dependencies, policy, - } = load_skill_metadata(fs, path, plugin_root).await; - - validate_len(&name, MAX_QUALIFIED_NAME_LEN, "qualified name")?; - - let resolved_path = canonicalize_for_skill_identity(fs, path).await; + } = loaded_metadata; Ok(SkillMetadata { - name, + name: base_name, description, short_description, interface, dependencies, policy, - path_to_skills_md: resolved_path, + path_to_skills_md: path.clone(), scope, plugin_id: plugin_id.map(str::to_string), }) @@ -838,41 +738,42 @@ fn default_skill_name(path: &AbsolutePathBuf) -> String { async fn load_skill_metadata( fs: &dyn ExecutorFileSystem, skill_path: &AbsolutePathBuf, + metadata: &SkillMetadataDiscovery, plugin_root: Option<&AbsolutePathBuf>, ) -> LoadedSkillMetadata { // Fail open: optional metadata should not block loading SKILL.md. let Some(skill_dir) = skill_path.parent() else { return LoadedSkillMetadata::default(); }; - let metadata_path = skill_dir - .join(SKILLS_METADATA_DIR) - .join(SKILLS_METADATA_FILENAME); - let metadata_path_uri = PathUri::from_abs_path(&metadata_path); - match fs.get_metadata(&metadata_path_uri, /*sandbox*/ None).await { - Ok(metadata) if metadata.is_file => {} - Ok(_) => return LoadedSkillMetadata::default(), - Err(error) if error.kind() == io::ErrorKind::NotFound => { - return LoadedSkillMetadata::default(); + let metadata_path_uri = match metadata { + SkillMetadataDiscovery::Present(path) => path, + SkillMetadataDiscovery::Absent => return LoadedSkillMetadata::default(), + SkillMetadataDiscovery::Probe(path) => { + match fs.get_metadata(path, /*sandbox*/ None).await { + Ok(metadata) if metadata.is_file => {} + Ok(_) => return LoadedSkillMetadata::default(), + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return LoadedSkillMetadata::default(); + } + Err(error) => { + tracing::warn!( + "ignoring {path}: failed to stat {label}: {error}", + path = path, + label = SKILLS_METADATA_FILENAME + ); + return LoadedSkillMetadata::default(); + } + } + path } - Err(error) => { - tracing::warn!( - "ignoring {path}: failed to stat {label}: {error}", - path = metadata_path.display(), - label = SKILLS_METADATA_FILENAME - ); - return LoadedSkillMetadata::default(); - } - } + }; - let contents = match fs - .read_file_text(&metadata_path_uri, /*sandbox*/ None) - .await - { + let contents = match fs.read_file_text(metadata_path_uri, /*sandbox*/ None).await { Ok(contents) => contents, Err(error) => { tracing::warn!( "ignoring {path}: failed to read {label}: {error}", - path = metadata_path.display(), + path = metadata_path_uri, label = SKILLS_METADATA_FILENAME ); return LoadedSkillMetadata::default(); @@ -886,7 +787,7 @@ async fn load_skill_metadata( Err(error) => { tracing::warn!( "ignoring {path}: invalid {label}: {error}", - path = metadata_path.display(), + path = metadata_path_uri, label = SKILLS_METADATA_FILENAME ); return LoadedSkillMetadata::default(); diff --git a/codex-rs/core-skills/src/loader/discovery.rs b/codex-rs/core-skills/src/loader/discovery.rs new file mode 100644 index 0000000000..547400bc4f --- /dev/null +++ b/codex-rs/core-skills/src/loader/discovery.rs @@ -0,0 +1,207 @@ +use std::collections::HashSet; +use std::io; + +use codex_exec_server::ExecutorFileSystem; +use codex_exec_server::WalkEntryKind; +use codex_exec_server::WalkOptions; +use codex_utils_path_uri::PathUri; +use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; + +use super::MAX_SCAN_DEPTH; +use super::MAX_SKILLS_DIRS_PER_ROOT; +use super::SKILLS_FILENAME; +use super::SKILLS_METADATA_DIR; +use super::SKILLS_METADATA_FILENAME; + +const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; +pub(super) const MAX_CONCURRENT_SKILL_LOADS: usize = 64; + +pub(super) enum DirectorySymlinkPolicy { + Follow, + Ignore, +} + +pub(super) enum HiddenDirectoryPolicy { + Include, + Skip, +} + +pub(super) struct SkillDiscoveryOptions { + pub directory_symlinks: DirectorySymlinkPolicy, + pub hidden_directories: HiddenDirectoryPolicy, +} + +pub(super) struct SkillDiscovery { + pub skills: Vec, + pub plugin_roots: HashSet, + pub namespace_roots: HashSet, + pub warnings: Vec, +} + +pub(super) struct DiscoveredSkill { + pub path: PathUri, + pub metadata: SkillMetadataDiscovery, +} + +pub(super) enum SkillMetadataDiscovery { + Present(PathUri), + Absent, + Probe(PathUri), +} + +pub(super) async fn discover_skills( + file_system: &dyn ExecutorFileSystem, + root: &PathUri, + options: SkillDiscoveryOptions, +) -> SkillDiscovery { + let empty_discovery = || SkillDiscovery { + skills: Vec::new(), + plugin_roots: HashSet::new(), + namespace_roots: HashSet::new(), + warnings: Vec::new(), + }; + let walk = match file_system + .walk( + root, + WalkOptions { + max_depth: MAX_SCAN_DEPTH, + max_directories: MAX_SKILLS_DIRS_PER_ROOT, + max_entries: MAX_SKILLS_ENTRIES_PER_ROOT, + follow_directory_symlinks: matches!( + options.directory_symlinks, + DirectorySymlinkPolicy::Follow + ), + prune_hidden_directories: matches!( + options.hidden_directories, + HiddenDirectoryPolicy::Skip + ), + }, + /*sandbox*/ None, + ) + .await + { + Ok(walk) => walk, + Err(error) if error.kind() == io::ErrorKind::NotFound => return empty_discovery(), + Err(error) => { + let mut discovery = empty_discovery(); + discovery + .warnings + .push(format!("failed to walk skills root {root}: {error:#}")); + return discovery; + } + }; + + let inventory_complete = !walk.truncated && walk.errors.is_empty(); + let mut warnings = walk + .errors + .into_iter() + .map(|error| { + format!( + "failed to scan skill path {}: {}", + error.path, error.message + ) + }) + .collect::>(); + if walk.truncated { + warnings.push(format!( + "skills scan reached its traversal limit (root: {root})" + )); + } + + let skip_hidden = matches!(options.hidden_directories, HiddenDirectoryPolicy::Skip); + let mut skill_files = Vec::new(); + let mut file_paths = HashSet::new(); + let mut metadata_directory_parents = HashSet::new(); + let mut plugin_roots = HashSet::new(); + for entry in walk.entries { + if skip_hidden && has_hidden_ancestor_below_root(&entry.path, root) { + continue; + } + match entry.kind { + WalkEntryKind::Directory => { + if entry + .path + .basename() + .is_some_and(|name| name.eq_ignore_ascii_case(SKILLS_METADATA_DIR)) + && let Some(skill_dir) = entry.path.parent() + { + metadata_directory_parents.insert(skill_dir); + } + if DISCOVERABLE_PLUGIN_MANIFEST_PATHS + .iter() + .any(|path| path.split('/').next() == entry.path.basename().as_deref()) + && let Some(plugin_root) = entry.path.parent() + { + plugin_roots.insert(plugin_root); + } + } + WalkEntryKind::File => { + file_paths.insert(entry.path.clone()); + if entry.path.basename().as_deref() == Some(SKILLS_FILENAME) { + skill_files.push(entry.path); + } + } + } + } + let skills = skill_files + .into_iter() + .map(|path| DiscoveredSkill { + metadata: discover_skill_metadata( + &path, + &file_paths, + &metadata_directory_parents, + inventory_complete, + ), + path, + }) + .collect(); + + SkillDiscovery { + skills, + plugin_roots, + namespace_roots: HashSet::from([root.clone()]), + warnings, + } +} + +fn has_hidden_ancestor_below_root(path: &PathUri, root: &PathUri) -> bool { + let mut ancestor = path.parent(); + while let Some(current) = ancestor { + if ¤t == root { + return false; + } + if current.basename().is_some_and(|name| name.starts_with('.')) { + return true; + } + ancestor = current.parent(); + } + false +} + +fn discover_skill_metadata( + skill_path: &PathUri, + file_paths: &HashSet, + metadata_directory_parents: &HashSet, + inventory_complete: bool, +) -> SkillMetadataDiscovery { + let Some(skill_dir) = skill_path.parent() else { + return SkillMetadataDiscovery::Absent; + }; + let Ok(metadata_dir) = skill_dir.join(SKILLS_METADATA_DIR) else { + return SkillMetadataDiscovery::Absent; + }; + let Ok(metadata_path) = metadata_dir.join(SKILLS_METADATA_FILENAME) else { + return SkillMetadataDiscovery::Absent; + }; + if file_paths.contains(&metadata_path) { + return SkillMetadataDiscovery::Present(metadata_path); + } + + if inventory_complete && !metadata_directory_parents.contains(&skill_dir) { + SkillMetadataDiscovery::Absent + } else { + // A complete walk proves ordinary absence, but keep a filesystem probe for case aliases, + // file symlinks omitted by the walk, and incomplete inventories. + SkillMetadataDiscovery::Probe(metadata_path) + } +} diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index 8d400d7cdc..052f292651 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -1,25 +1,23 @@ -use std::collections::HashSet; use std::io; use codex_exec_server::ExecutorFileSystem; -use codex_exec_server::WalkEntryKind; -use codex_exec_server::WalkOptions; use codex_protocol::protocol::Product; use codex_utils_path_uri::PathUri; -use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use futures::StreamExt; use crate::model::SkillDependencies; use crate::model::SkillPolicy; use super::MAX_QUALIFIED_NAME_LEN; -use super::MAX_SCAN_DEPTH; -use super::MAX_SKILLS_DIRS_PER_ROOT; use super::ParsedSkillFrontmatter; -use super::SKILLS_FILENAME; -use super::SKILLS_METADATA_DIR; -use super::SKILLS_METADATA_FILENAME; use super::SkillMetadataFile; +use super::discovery::DirectorySymlinkPolicy; +use super::discovery::DiscoveredSkill; +use super::discovery::HiddenDirectoryPolicy; +use super::discovery::MAX_CONCURRENT_SKILL_LOADS; +use super::discovery::SkillDiscoveryOptions; +use super::discovery::SkillMetadataDiscovery; +use super::discovery::discover_skills; use super::namespace::SkillNamespaceResolver; use super::parse_skill_frontmatter_metadata_inner; use super::resolve_dependencies; @@ -27,21 +25,6 @@ use super::resolve_policy; use super::sanitize_single_line; use super::validate_len; -const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; -const MAX_CONCURRENT_SKILL_LOADS: usize = 64; - -struct EnvironmentSkillDiscovery { - skills: Vec, - plugin_roots: HashSet, - namespace_roots: HashSet, - warnings: Vec, -} - -struct DiscoveredEnvironmentSkill { - path: PathUri, - metadata: SkillMetadataDiscovery, -} - struct ParsedEnvironmentSkill { path_to_skills_md: PathUri, base_name: String, @@ -51,12 +34,6 @@ struct ParsedEnvironmentSkill { policy: Option, } -enum SkillMetadataDiscovery { - Present(PathUri), - Absent, - Probe(PathUri), -} - /// URI-native metadata for one skill owned by an execution environment. #[derive(Clone, Debug, PartialEq, Eq)] pub struct EnvironmentSkillMetadata { @@ -92,7 +69,7 @@ impl EnvironmentSkillMetadata { impl ParsedEnvironmentSkill { async fn load( file_system: &dyn ExecutorFileSystem, - skill: &DiscoveredEnvironmentSkill, + skill: &DiscoveredSkill, ) -> Result { let (contents, discovered_metadata) = match &skill.metadata { SkillMetadataDiscovery::Present(metadata_path) => { @@ -152,93 +129,17 @@ pub async fn load_environment_skills_from_root( restriction_product: Option, ) -> EnvironmentSkillLoadOutcome { let mut outcome = EnvironmentSkillLoadOutcome::default(); - let discovery = match file_system - .walk( - root, - WalkOptions { - max_depth: MAX_SCAN_DEPTH, - max_directories: MAX_SKILLS_DIRS_PER_ROOT, - max_entries: MAX_SKILLS_ENTRIES_PER_ROOT, - follow_directory_symlinks: true, - prune_hidden_directories: false, - }, - /*sandbox*/ None, - ) - .await - { - Ok(walk) => { - let inventory_complete = !walk.truncated && walk.errors.is_empty(); - let mut warnings = walk - .errors - .into_iter() - .map(|error| { - format!( - "failed to scan skill path {}: {}", - error.path, error.message - ) - }) - .collect::>(); - if walk.truncated { - warnings.push(format!( - "skills scan reached its traversal limit (root: {root})" - )); - } - let mut skill_files = Vec::new(); - let mut file_paths = HashSet::new(); - let mut directory_paths = HashSet::new(); - let mut plugin_roots = HashSet::new(); - for entry in walk.entries { - match entry.kind { - WalkEntryKind::Directory => { - directory_paths.insert(entry.path.clone()); - if DISCOVERABLE_PLUGIN_MANIFEST_PATHS - .iter() - .any(|path| path.split('/').next() == entry.path.basename().as_deref()) - && let Some(plugin_root) = entry.path.parent() - { - plugin_roots.insert(plugin_root); - } - } - WalkEntryKind::File => { - file_paths.insert(entry.path.clone()); - if entry.path.basename().as_deref() == Some(SKILLS_FILENAME) { - skill_files.push(entry.path); - } - } - } - } - let skills = skill_files - .into_iter() - .map(|path| DiscoveredEnvironmentSkill { - metadata: discover_skill_metadata( - &path, - &file_paths, - &directory_paths, - inventory_complete, - ), - path, - }) - .collect(); - EnvironmentSkillDiscovery { - skills, - plugin_roots, - namespace_roots: HashSet::from([root.clone()]), - warnings, - } - } - Err(error) if error.kind() == io::ErrorKind::NotFound => EnvironmentSkillDiscovery { - skills: Vec::new(), - plugin_roots: HashSet::new(), - namespace_roots: HashSet::new(), - warnings: Vec::new(), + let discovery = discover_skills( + file_system, + root, + // Preserve environment discovery behavior by following directory aliases and including + // hidden directories exposed by the executor. + SkillDiscoveryOptions { + directory_symlinks: DirectorySymlinkPolicy::Follow, + hidden_directories: HiddenDirectoryPolicy::Include, }, - Err(error) => EnvironmentSkillDiscovery { - skills: Vec::new(), - plugin_roots: HashSet::new(), - namespace_roots: HashSet::new(), - warnings: vec![format!("failed to walk skills root {root}: {error:#}")], - }, - }; + ) + .await; tracing::Span::current().record("skill_count", discovery.skills.len()); outcome.warnings.extend(discovery.warnings); if discovery.skills.is_empty() { @@ -311,32 +212,6 @@ pub async fn load_environment_skills_from_root( outcome } -fn discover_skill_metadata( - skill_path: &PathUri, - file_paths: &HashSet, - directory_paths: &HashSet, - inventory_complete: bool, -) -> SkillMetadataDiscovery { - let Some(skill_dir) = skill_path.parent() else { - return SkillMetadataDiscovery::Absent; - }; - let Ok(metadata_dir) = skill_dir.join(SKILLS_METADATA_DIR) else { - return SkillMetadataDiscovery::Absent; - }; - let Ok(metadata_path) = metadata_dir.join(SKILLS_METADATA_FILENAME) else { - return SkillMetadataDiscovery::Absent; - }; - if file_paths.contains(&metadata_path) { - SkillMetadataDiscovery::Present(metadata_path) - } else if inventory_complete && !directory_paths.contains(&metadata_dir) { - SkillMetadataDiscovery::Absent - } else { - // The walk can omit entries after an error or traversal limit. It also omits file - // symlinks, so keep the existing probe when the metadata directory itself was observed. - SkillMetadataDiscovery::Probe(metadata_path) - } -} - async fn read_skill_contents( file_system: &dyn ExecutorFileSystem, skill_path: &PathUri, diff --git a/codex-rs/core-skills/src/loader/namespace.rs b/codex-rs/core-skills/src/loader/namespace.rs index 34d8e9b7c4..fa00b1005d 100644 --- a/codex-rs/core-skills/src/loader/namespace.rs +++ b/codex-rs/core-skills/src/loader/namespace.rs @@ -1,10 +1,12 @@ use codex_exec_server::ExecutorFileSystem; use codex_utils_path_uri::PathUri; use codex_utils_plugins::plugin_namespace_for_root_uri; -use codex_utils_plugins::plugin_namespace_for_skill_uri; -use futures::future::join_all; +use futures::StreamExt; +use std::collections::HashMap; use std::collections::HashSet; +use super::discovery::MAX_CONCURRENT_SKILL_LOADS; + /// Resolves the namespace prefix applied to skill names during one skills scan. /// /// A plugin namespace is the plugin name from the nearest valid plugin manifest @@ -55,43 +57,89 @@ impl SkillNamespaceResolver { .filter(|plugin_root| skill_ancestors.contains(plugin_root)) .collect::>(); - // Ordinary descendants fall back to the nearest valid manifest at or above the scan root. - let inherited_namespace = plugin_namespace_for_skill_uri(fs, root) - .await - .map(ResolvedSkillNamespace::Plugin) - .unwrap_or(ResolvedSkillNamespace::Plain); // The scan root is already the fallback above if nothing else matches, exclude from the search. let namespace_roots = namespace_roots .into_iter() .filter(|namespace_root| namespace_root != root) .collect::>(); let namespace_root_set = namespace_roots.iter().cloned().collect::>(); - // Keep independent probes concurrent for remote executor latency. - let namespace_lookups = join_all(namespace_roots.into_iter().map(|namespace_root| async { - let namespace = plugin_namespace_for_skill_uri(fs, &namespace_root) - .await + let plugin_roots = plugin_roots + .into_iter() + .filter(|plugin_root| plugin_root != root && !namespace_root_set.contains(plugin_root)) + .collect::>(); + + let lookup_roots = std::iter::once(root.clone()) + .chain(namespace_roots.iter().cloned()) + .collect::>(); + let mut pending_lookups = lookup_roots + .iter() + .cloned() + .map(|lookup_root| (lookup_root.clone(), lookup_root)) + .collect::>(); + let mut direct_plugin_roots = plugin_roots.iter().cloned().collect::>(); + let mut namespaces_by_root = HashMap::new(); + let mut namespaces_by_lookup_root = HashMap::new(); + while !pending_lookups.is_empty() { + let probe_roots = pending_lookups + .iter() + .map(|(_, ancestor)| ancestor.clone()) + .chain(direct_plugin_roots.drain()) + .filter(|ancestor| !namespaces_by_root.contains_key(ancestor)) + .collect::>(); + namespaces_by_root.extend( + futures::stream::iter(probe_roots) + .map(|manifest_root| async move { + let namespace = plugin_namespace_for_root_uri(fs, &manifest_root).await; + (manifest_root, namespace) + }) + .buffered(MAX_CONCURRENT_SKILL_LOADS) + .collect::>() + .await, + ); + + let mut next_lookups = Vec::new(); + for (lookup_root, ancestor) in pending_lookups { + match namespaces_by_root.get(&ancestor) { + Some(Some(namespace)) => { + namespaces_by_lookup_root.insert(lookup_root, Some(namespace.clone())); + } + Some(None) => match ancestor.parent() { + Some(parent) => next_lookups.push((lookup_root, parent)), + None => { + namespaces_by_lookup_root.insert(lookup_root, None); + } + }, + None => unreachable!("pending namespace ancestor was not probed"), + } + } + pending_lookups = next_lookups; + } + + // Ordinary descendants fall back to the nearest valid manifest at or above the scan root. + let inherited_namespace = namespaces_by_lookup_root + .get(root) + .and_then(Option::as_ref) + .cloned() + .map(ResolvedSkillNamespace::Plugin) + .unwrap_or(ResolvedSkillNamespace::Plain); + let namespace_lookups = namespace_roots.into_iter().map(|namespace_root| { + let namespace = namespaces_by_lookup_root + .get(&namespace_root) + .and_then(Option::as_ref) + .cloned() .map(ResolvedSkillNamespace::Plugin) .unwrap_or(ResolvedSkillNamespace::Plain); (namespace_root, namespace) - })); + }); // Invalid nested manifests are omitted, so the deepest remaining match wins. - let plugin_lookups = join_all( - plugin_roots - .into_iter() - .filter(|plugin_root| { - plugin_root != root && !namespace_root_set.contains(plugin_root) - }) - .map(|plugin_root| async move { - plugin_namespace_for_root_uri(fs, &plugin_root) - .await - .map(|namespace| (plugin_root, ResolvedSkillNamespace::Plugin(namespace))) - }), - ); - let (namespace_lookups, plugin_lookups) = tokio::join!(namespace_lookups, plugin_lookups); - let nested_namespaces = namespace_lookups - .into_iter() - .chain(plugin_lookups.into_iter().flatten()) - .collect(); + let plugin_lookups = plugin_roots.into_iter().filter_map(|plugin_root| { + namespaces_by_root + .get(&plugin_root) + .and_then(Option::as_ref) + .cloned() + .map(|namespace| (plugin_root, ResolvedSkillNamespace::Plugin(namespace))) + }); + let nested_namespaces = namespace_lookups.chain(plugin_lookups).collect(); Self { inherited_namespace, diff --git a/codex-rs/core-skills/src/loader_tests.rs b/codex-rs/core-skills/src/loader_tests.rs index 1bdfe39520..3122dbaf0b 100644 --- a/codex-rs/core-skills/src/loader_tests.rs +++ b/codex-rs/core-skills/src/loader_tests.rs @@ -1021,6 +1021,32 @@ async fn loads_skills_via_symlinked_subdir_for_user_scope() { ); } +// Directory symlinks on Windows can require Developer Mode or administrator privileges. +#[tokio::test] +#[cfg(unix)] +async fn loads_skills_through_visible_alias_to_hidden_directory() { + let root = tempfile::tempdir().expect("tempdir"); + let hidden_root = root.path().join(".hidden"); + let skill_path = write_skill_at(&hidden_root, "search", "search-skill", "search description"); + symlink_dir(&hidden_root, &root.path().join("visible")); + + let outcome = load_user_skills_root(root.path()).await; + + assert!( + outcome.errors.is_empty(), + "unexpected errors: {:?}", + outcome.errors + ); + assert_eq!( + outcome.skills, + vec![expected_user_skill( + &skill_path, + "search-skill", + "search description", + )] + ); +} + #[tokio::test] #[cfg(unix)] async fn ignores_symlinked_skill_file_for_user_scope() { @@ -1485,48 +1511,6 @@ async fn invalid_nested_plugin_manifest_falls_back_to_outer_namespace() { ); } -// Directory symlinks on Windows can require Developer Mode or administrator privileges. -#[cfg(unix)] -#[tokio::test] -async fn namespaces_skills_in_symlinked_plugin_skills_dir() { - // skills/ - // └── linked-plugin -> shared-plugin/skills/ - // shared-plugin/ - // ├── .codex-plugin/plugin.json - // └── skills/search/SKILL.md - let root = tempfile::tempdir().expect("tempdir"); - let shared_plugin_root = tempfile::tempdir().expect("tempdir"); - write_plugin_manifest(shared_plugin_root.path(), r#"{"name":"linked"}"#); - let skill_path = write_skill_at( - &shared_plugin_root.path().join("skills"), - "search", - "search-skill", - "search description", - ); - let skills_root = root.path().join("skills"); - fs::create_dir_all(&skills_root).unwrap(); - symlink_dir( - &shared_plugin_root.path().join("skills"), - &skills_root.join("linked-plugin"), - ); - - let outcome = load_user_skills_root(&skills_root).await; - - assert!( - outcome.errors.is_empty(), - "unexpected errors: {:?}", - outcome.errors - ); - assert_eq!( - outcome.skills, - vec![expected_user_skill( - &skill_path, - "linked:search-skill", - "search description", - )] - ); -} - // Directory symlinks on Windows can require Developer Mode or administrator privileges. #[cfg(unix)] #[tokio::test] diff --git a/codex-rs/core-skills/tests/environment_loader.rs b/codex-rs/core-skills/tests/environment_loader.rs index 74efc6e1a3..502374b1b0 100644 --- a/codex-rs/core-skills/tests/environment_loader.rs +++ b/codex-rs/core-skills/tests/environment_loader.rs @@ -411,3 +411,100 @@ async fn reads_skill_files_while_resolving_plugin_namespaces() { }] ); } + +#[cfg(unix)] +#[tokio::test] +async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() { + use std::os::unix::fs::symlink; + use std::sync::Arc; + + use codex_core_skills::loader::SkillRoot; + use codex_core_skills::loader::load_skills_from_roots; + use codex_protocol::protocol::SkillScope; + use codex_utils_absolute_path::test_support::PathBufExt; + + let root = tempdir().expect("tempdir"); + let shared_plugin_root = tempdir().expect("tempdir"); + let manifest_path = shared_plugin_root.path().join(".codex-plugin/plugin.json"); + fs::create_dir_all(manifest_path.parent().expect("manifest parent")).expect("manifest dir"); + fs::write(&manifest_path, r#"{"name":"linked"}"#).expect("manifest"); + + let skills_root = shared_plugin_root.path().join("skills"); + for name in ["first", "second"] { + let skill_path = skills_root.join(name).join("SKILL.md"); + fs::create_dir_all(skill_path.parent().expect("skill parent")).expect("skill dir"); + fs::write( + &skill_path, + format!("---\nname: {name}\ndescription: {name} skill.\n---\n"), + ) + .expect("skill"); + } + let metadata_path = skills_root.join("first/agents/openai.yaml"); + fs::create_dir_all(metadata_path.parent().expect("metadata parent")).expect("metadata dir"); + fs::write( + &metadata_path, + "policy:\n allow_implicit_invocation: false\n", + ) + .expect("metadata"); + + let host_root = root.path().join("skills"); + fs::create_dir_all(&host_root).expect("host skills dir"); + let linked_root = host_root.join("linked-plugin"); + symlink(&skills_root, &linked_root).expect("skill pack symlink"); + + let recording = Arc::new(RecordingFileSystem::new( + LOCAL_FS.as_ref(), + ManifestMetadataBehavior::Immediate, + )); + let file_system: Arc = recording.clone(); + let future = load_skills_from_roots( + [SkillRoot { + path: host_root.abs(), + scope: SkillScope::User, + file_system, + plugin_id: None, + plugin_namespace: None, + plugin_root: None, + }], + /*plugin_skill_snapshots*/ None, + ); + fn assert_send(_: &T) {} + assert_send(&future); + let outcome = future.await; + + assert_eq!(outcome.errors, Vec::new()); + assert_eq!( + outcome + .skills + .iter() + .map(|skill| skill.name.as_str()) + .collect::>(), + vec!["linked:first", "linked:second"] + ); + + let calls = recording.calls(); + assert_eq!(calls.walks, 1); + let linked_root = PathUri::from_host_native_path(linked_root).unwrap(); + assert!( + calls + .read_files + .iter() + .all(|path| !path.starts_with(&linked_root)) + ); + assert!( + calls + .metadata_files + .iter() + .all(|path| path.basename().as_deref() != Some("openai.yaml")) + ); + let manifest_uri = + PathUri::from_host_native_path(dunce::canonicalize(manifest_path).unwrap()).unwrap(); + assert_eq!( + calls + .metadata_files + .iter() + .filter(|path| **path == manifest_uri) + .count(), + 1 + ); +} diff --git a/codex-rs/core/tests/suite/unified_exec_process_events.rs b/codex-rs/core/tests/suite/unified_exec_process_events.rs index 8ebd332ea2..7fade19bce 100644 --- a/codex-rs/core/tests/suite/unified_exec_process_events.rs +++ b/codex-rs/core/tests/suite/unified_exec_process_events.rs @@ -137,6 +137,16 @@ async fn serve_exec_with_pushed_events( ) .await; } + Some("fs/walk") => { + send_exec_server_json( + &mut websocket, + json!({ + "id": request["id"], + "result": { "entries": [], "errors": [], "truncated": false } + }), + ) + .await; + } method => panic!("unexpected exec-server request before process/start: {method:?}"), } };