Move direct executor skill discovery into the skills extension (#36880)

## What changed

- Add extension-owned discovery and namespace resolution for skills loaded directly through an `ExecutorFileSystem`.
- Preserve hidden and symlinked skills, nested plugin namespaces, optional `agents/openai.yaml` metadata, product restrictions, and deterministic ordering.
- Reuse the filesystem walk inventory and bound concurrent skill, metadata, and manifest reads.
- Route direct executor catalog loading through the new extension loader.

## Testing

- Cover namespace lookup, metadata probing, walk reuse, concurrent reads, and parity with the existing environment loader.

GitOrigin-RevId: 4e0b821eb84d03f0dc1c2dee7b2b9a072ee3fd44

codex-rs/core-skills/tests/environment_loader.rs

@@ -1,12 +1,10 @@
+#![cfg(unix)]
+
 use std::fs;
 use std::sync::Mutex;
-use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
-use std::time::Duration;
 
-use codex_core_skills::loader::EnvironmentSkillMetadata;
-use codex_core_skills::loader::load_environment_skills_from_root;
 use codex_exec_server::CopyOptions;
 use codex_exec_server::CreateDirectoryOptions;
 use codex_exec_server::ExecutorFileSystem;
@@ -22,22 +20,12 @@ use codex_exec_server::WalkOutcome;
 use codex_utils_path_uri::PathUri;
 use pretty_assertions::assert_eq;
 use tempfile::tempdir;
-use tokio::sync::Notify;
-
-#[derive(Clone, Copy)]
-enum ManifestMetadataBehavior {
-    Immediate,
-    WaitForSkillRead,
-}
 
 struct RecordingFileSystem<'a> {
     inner: &'a dyn ExecutorFileSystem,
     read_files: Mutex<Vec<PathUri>>,
     metadata_files: Mutex<Vec<PathUri>>,
     walks: AtomicUsize,
-    manifest_metadata_behavior: ManifestMetadataBehavior,
-    skill_read_started: AtomicBool,
-    skill_read_started_notify: Notify,
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -48,18 +36,12 @@ struct FileSystemCalls {
 }
 
 impl<'a> RecordingFileSystem<'a> {
-    fn new(
-        inner: &'a dyn ExecutorFileSystem,
-        manifest_metadata_behavior: ManifestMetadataBehavior,
-    ) -> Self {
+    fn new(inner: &'a dyn ExecutorFileSystem) -> Self {
         Self {
             inner,
             read_files: Mutex::new(Vec::new()),
             metadata_files: Mutex::new(Vec::new()),
             walks: AtomicUsize::new(0),
-            manifest_metadata_behavior,
-            skill_read_started: AtomicBool::new(false),
-            skill_read_started_notify: Notify::new(),
         }
     }
 
@@ -102,10 +84,6 @@ impl ExecutorFileSystem for RecordingFileSystem<'_> {
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner)
             .push(path.clone());
-        if path.basename().as_deref() == Some("SKILL.md") {
-            self.skill_read_started.store(true, Ordering::Release);
-            self.skill_read_started_notify.notify_waiters();
-        }
         self.inner.read_file(path, sandbox)
     }
 
@@ -144,22 +122,6 @@ impl ExecutorFileSystem for RecordingFileSystem<'_> {
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner)
             .push(path.clone());
-        if matches!(
-            self.manifest_metadata_behavior,
-            ManifestMetadataBehavior::WaitForSkillRead
-        ) && path.basename().as_deref() == Some("plugin.json")
-        {
-            return Box::pin(async move {
-                loop {
-                    let notified = self.skill_read_started_notify.notified();
-                    if self.skill_read_started.load(Ordering::Acquire) {
-                        break;
-                    }
-                    notified.await;
-                }
-                self.inner.get_metadata(path, sandbox).await
-            });
-        }
         self.inner.get_metadata(path, sandbox)
     }
 
@@ -202,216 +164,6 @@ impl ExecutorFileSystem for RecordingFileSystem<'_> {
     }
 }
 
-#[tokio::test]
-async fn loads_nearest_plugin_namespaces_without_reading_unused_sibling_manifests() {
-    let root = tempdir().expect("tempdir");
-    let standalone_skill = root.path().join("standalone/SKILL.md");
-    let outer_root = root.path().join("plugins/outer");
-    let outer_skill = outer_root.join("skills/deploy/SKILL.md");
-    let inner_root = outer_root.join("nested/inner");
-    let inner_skill = inner_root.join("skills/audit/SKILL.md");
-    let unused_root = root.path().join("plugins/unused");
-
-    for path in [&standalone_skill, &outer_skill, &inner_skill] {
-        fs::create_dir_all(path.parent().expect("skill parent")).expect("skill dir");
-    }
-    for (plugin_root, name) in [
-        (&outer_root, "outer"),
-        (&inner_root, "inner"),
-        (&unused_root, "unused"),
-    ] {
-        fs::create_dir_all(plugin_root.join(".codex-plugin")).expect("manifest dir");
-        fs::write(
-            plugin_root.join(".codex-plugin/plugin.json"),
-            format!(r#"{{"name":"{name}"}}"#),
-        )
-        .expect("manifest");
-    }
-    for (path, name) in [
-        (&standalone_skill, "standalone"),
-        (&outer_skill, "deploy"),
-        (&inner_skill, "audit"),
-    ] {
-        fs::write(
-            path,
-            format!("---\nname: {name}\ndescription: {name} skill.\n---\n"),
-        )
-        .expect("skill");
-    }
-
-    let file_system =
-        RecordingFileSystem::new(LOCAL_FS.as_ref(), ManifestMetadataBehavior::Immediate);
-    let root_uri = PathUri::from_host_native_path(root.path()).expect("root URI");
-    let outcome = load_environment_skills_from_root(
-        &file_system,
-        &root_uri,
-        /*restriction_product*/ None,
-    )
-    .await;
-
-    assert_eq!(outcome.warnings, Vec::<String>::new());
-    assert_eq!(
-        outcome.skills,
-        vec![
-            EnvironmentSkillMetadata {
-                path_to_skills_md: PathUri::from_host_native_path(&inner_skill).unwrap(),
-                name: "inner:audit".to_string(),
-                description: "audit skill.".to_string(),
-                short_description: None,
-                dependencies: None,
-                policy: None,
-            },
-            EnvironmentSkillMetadata {
-                path_to_skills_md: PathUri::from_host_native_path(&outer_skill).unwrap(),
-                name: "outer:deploy".to_string(),
-                description: "deploy skill.".to_string(),
-                short_description: None,
-                dependencies: None,
-                policy: None,
-            },
-            EnvironmentSkillMetadata {
-                path_to_skills_md: PathUri::from_host_native_path(&standalone_skill).unwrap(),
-                name: "standalone".to_string(),
-                description: "standalone skill.".to_string(),
-                short_description: None,
-                dependencies: None,
-                policy: None,
-            },
-        ]
-    );
-
-    let mut manifest_reads = file_system
-        .calls()
-        .read_files
-        .into_iter()
-        .filter(|path| path.basename().as_deref() == Some("plugin.json"))
-        .collect::<Vec<_>>();
-    manifest_reads.sort_by_key(ToString::to_string);
-    let mut expected_manifest_reads = [&outer_root, &inner_root]
-        .into_iter()
-        .map(|plugin_root| {
-            PathUri::from_host_native_path(plugin_root.join(".codex-plugin/plugin.json")).unwrap()
-        })
-        .collect::<Vec<_>>();
-    expected_manifest_reads.sort_by_key(ToString::to_string);
-    assert_eq!(manifest_reads, expected_manifest_reads);
-}
-
-#[tokio::test]
-async fn reuses_walk_inventory_for_missing_skill_metadata() {
-    const SKILL_COUNT: usize = 66;
-
-    let root = tempdir().expect("tempdir");
-    let manifest_path = 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":"inventory"}"#).expect("manifest");
-
-    let mut skill_paths = Vec::new();
-    for index in 0..SKILL_COUNT {
-        let name = format!("skill-{index}");
-        let skill_path = root.path().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");
-        skill_paths.push(skill_path);
-    }
-
-    let file_system =
-        RecordingFileSystem::new(LOCAL_FS.as_ref(), ManifestMetadataBehavior::Immediate);
-    let root_uri = PathUri::from_host_native_path(root.path()).expect("root URI");
-    let outcome = load_environment_skills_from_root(
-        &file_system,
-        &root_uri,
-        /*restriction_product*/ None,
-    )
-    .await;
-
-    let mut expected_skills = skill_paths
-        .iter()
-        .enumerate()
-        .map(|(index, skill_path)| EnvironmentSkillMetadata {
-            path_to_skills_md: PathUri::from_host_native_path(skill_path).unwrap(),
-            name: format!("inventory:skill-{index}"),
-            description: format!("skill-{index} skill."),
-            short_description: None,
-            dependencies: None,
-            policy: None,
-        })
-        .collect::<Vec<_>>();
-    expected_skills.sort_by(|left, right| {
-        left.name.cmp(&right.name).then_with(|| {
-            left.path_to_skills_md
-                .to_string()
-                .cmp(&right.path_to_skills_md.to_string())
-        })
-    });
-    assert_eq!(outcome.skills, expected_skills);
-    assert_eq!(outcome.warnings, Vec::<String>::new());
-
-    let mut expected_read_files = skill_paths
-        .iter()
-        .map(|path| PathUri::from_host_native_path(path).unwrap())
-        .collect::<Vec<_>>();
-    let manifest_uri = PathUri::from_host_native_path(manifest_path).unwrap();
-    expected_read_files.push(manifest_uri.clone());
-    expected_read_files.sort_by_key(ToString::to_string);
-    assert_eq!(
-        file_system.calls(),
-        FileSystemCalls {
-            walks: 1,
-            read_files: expected_read_files,
-            metadata_files: vec![manifest_uri],
-        }
-    );
-}
-
-#[tokio::test]
-async fn reads_skill_files_while_resolving_plugin_namespaces() {
-    let root = tempdir().expect("tempdir");
-    let manifest_path = 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":"parallel"}"#).expect("manifest");
-    let skill_path = root.path().join("demo/SKILL.md");
-    fs::create_dir_all(skill_path.parent().expect("skill parent")).expect("skill dir");
-    fs::write(
-        &skill_path,
-        "---\nname: demo\ndescription: demo skill.\n---\n",
-    )
-    .expect("skill");
-
-    let file_system = RecordingFileSystem::new(
-        LOCAL_FS.as_ref(),
-        ManifestMetadataBehavior::WaitForSkillRead,
-    );
-    let root_uri = PathUri::from_host_native_path(root.path()).expect("root URI");
-    let outcome = tokio::time::timeout(
-        Duration::from_secs(5),
-        load_environment_skills_from_root(
-            &file_system,
-            &root_uri,
-            /*restriction_product*/ None,
-        ),
-    )
-    .await
-    .expect("skill reads should start before namespace resolution finishes");
-
-    assert_eq!(outcome.warnings, Vec::<String>::new());
-    assert_eq!(
-        outcome.skills,
-        vec![EnvironmentSkillMetadata {
-            path_to_skills_md: PathUri::from_host_native_path(skill_path).unwrap(),
-            name: "parallel:demo".to_string(),
-            description: "demo skill.".to_string(),
-            short_description: None,
-            dependencies: None,
-            policy: None,
-        }]
-    );
-}
-
 #[cfg(unix)]
 #[tokio::test]
 async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() {
@@ -455,10 +207,7 @@ async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() {
     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 recording = Arc::new(RecordingFileSystem::new(LOCAL_FS.as_ref()));
     let file_system: Arc<dyn ExecutorFileSystem> = recording.clone();
     let future = load_skills_from_roots(
         [SkillRoot {