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/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<DiscoveredSkill>,
+    pub plugin_roots: HashSet<PathUri>,
+    pub namespace_roots: HashSet<PathUri>,
+    pub warnings: Vec<String>,
+}
+
+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::<Vec<_>>();
+    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 &current == 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<PathUri>,
+    metadata_directory_parents: &HashSet<PathUri>,
+    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)
+    }
+}

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<DiscoveredEnvironmentSkill>,
-    plugin_roots: HashSet<PathUri>,
-    namespace_roots: HashSet<PathUri>,
-    warnings: Vec<String>,
-}
-
-struct DiscoveredEnvironmentSkill {
-    path: PathUri,
-    metadata: SkillMetadataDiscovery,
-}
-
 struct ParsedEnvironmentSkill {
     path_to_skills_md: PathUri,
     base_name: String,
@@ -51,12 +34,6 @@ struct ParsedEnvironmentSkill {
     policy: Option<SkillPolicy>,
 }
 
-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<Self, String> {
         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<Product>,
 ) -> 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::<Vec<_>>();
-            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<PathUri>,
-    directory_paths: &HashSet<PathUri>,
-    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,

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::<HashSet<_>>();
 
-        // 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::<Vec<_>>();
         let namespace_root_set = namespace_roots.iter().cloned().collect::<HashSet<_>>();
-        // 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::<Vec<_>>();
+
+        let lookup_roots = std::iter::once(root.clone())
+            .chain(namespace_roots.iter().cloned())
+            .collect::<Vec<_>>();
+        let mut pending_lookups = lookup_roots
+            .iter()
+            .cloned()
+            .map(|lookup_root| (lookup_root.clone(), lookup_root))
+            .collect::<Vec<_>>();
+        let mut direct_plugin_roots = plugin_roots.iter().cloned().collect::<HashSet<_>>();
+        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::<HashSet<_>>();
+            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::<HashMap<_, _>>()
+                    .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,