Merge a783d0fa6b into sapling-pr-archive-bolinfest

codex-rs/Cargo.lock

@@ -2015,6 +2015,7 @@ dependencies = [
  "codex-analytics",
  "codex-app-server-protocol",
  "codex-config",
+ "codex-exec-server",
  "codex-instructions",
  "codex-login",
  "codex-otel",
@@ -3075,10 +3076,13 @@ dependencies = [
 name = "codex-utils-plugins"
 version = "0.0.0"
 dependencies = [
+ "codex-exec-server",
  "codex-login",
+ "codex-utils-absolute-path",
  "serde",
  "serde_json",
  "tempfile",
+ "tokio",
 ]
 
 [[package]]

codex-rs/app-server/src/codex_message_processor.rs

@@ -6292,6 +6292,23 @@ impl CodexMessageProcessor {
         };
         let skills_manager = self.thread_manager.skills_manager();
         let plugins_manager = self.thread_manager.plugins_manager();
+        let fs = match self.thread_manager.environment_manager().current().await {
+            Ok(Some(environment)) => Some(environment.get_filesystem()),
+            Ok(None) => None,
+            Err(err) => {
+                self.outgoing
+                    .send_error(
+                        request_id,
+                        JSONRPCErrorError {
+                            code: INTERNAL_ERROR_CODE,
+                            message: format!("failed to create environment: {err}"),
+                            data: None,
+                        },
+                    )
+                    .await;
+                return;
+            }
+        };
         let cli_overrides = self.current_cli_overrides();
         let mut data = Vec::new();
         for cwd in cwds {
@@ -6349,7 +6366,12 @@ impl CodexMessageProcessor {
                 config.bundled_skills_enabled(),
             );
             let outcome = skills_manager
-                .skills_for_cwd_with_extra_user_roots(&skills_input, force_reload, extra_roots)
+                .skills_for_cwd_with_extra_user_roots(
+                    &skills_input,
+                    force_reload,
+                    extra_roots,
+                    fs.clone(),
+                )
                 .await;
             let errors = errors_to_info(&outcome.errors);
             let skills = skills_to_info(&outcome.skills, &outcome.disabled_paths);

codex-rs/app-server/tests/suite/v2/marketplace_add.rs

@@ -1,7 +1,11 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
+use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::MarketplaceAddParams;
+use codex_app_server_protocol::MarketplaceAddResponse;
 use codex_app_server_protocol::RequestId;
+use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use tokio::time::Duration;
 use tokio::time::timeout;
@@ -9,8 +13,20 @@ use tokio::time::timeout;
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 
 #[tokio::test]
-async fn marketplace_add_rejects_local_directory_source() -> Result<()> {
+async fn marketplace_add_local_directory_source() -> Result<()> {
     let codex_home = TempDir::new()?;
+    let source = codex_home.path().join("marketplace");
+    std::fs::create_dir_all(source.join(".agents/plugins"))?;
+    std::fs::create_dir_all(source.join("plugins/sample/.codex-plugin"))?;
+    std::fs::write(
+        source.join(".agents/plugins/marketplace.json"),
+        r#"{"name":"debug","plugins":[]}"#,
+    )?;
+    std::fs::write(
+        source.join("plugins/sample/.codex-plugin/plugin.json"),
+        r#"{"name":"sample"}"#,
+    )?;
+    std::fs::write(source.join("plugins/sample/marker.txt"), "local ref")?;
     let mut mcp = McpProcess::new(codex_home.path()).await?;
     timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
 
@@ -22,19 +38,24 @@ async fn marketplace_add_rejects_local_directory_source() -> Result<()> {
         })
         .await?;
 
-    let err = timeout(
+    let response: JSONRPCResponse = timeout(
         DEFAULT_TIMEOUT,
-        mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
     )
     .await??;
+    let MarketplaceAddResponse {
+        marketplace_name,
+        installed_root,
+        already_added,
+    } = to_response(response)?;
+    let expected_root = source.canonicalize()?;
 
-    assert_eq!(err.error.code, -32600);
-    assert!(
-        err.error.message.contains(
-            "local marketplace sources are not supported yet; use an HTTP(S) Git URL, SSH Git URL, or GitHub owner/repo"
-        ),
-        "unexpected error: {}",
-        err.error.message
+    assert_eq!(marketplace_name, "debug");
+    assert_eq!(installed_root.as_path(), expected_root.as_path());
+    assert!(!already_added);
+    assert_eq!(
+        std::fs::read_to_string(installed_root.as_path().join("plugins/sample/marker.txt"))?,
+        "local ref"
     );
     Ok(())
 }

codex-rs/app-server/tests/suite/v2/skills_list.rs

@@ -11,6 +11,7 @@ use codex_app_server_protocol::SkillsListExtraRootsForCwd;
 use codex_app_server_protocol::SkillsListParams;
 use codex_app_server_protocol::SkillsListResponse;
 use codex_app_server_protocol::ThreadStartParams;
+use codex_exec_server::CODEX_EXEC_SERVER_URL_ENV_VAR;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use tokio::time::timeout;
@@ -64,6 +65,56 @@ async fn skills_list_includes_skills_from_per_cwd_extra_user_roots() -> Result<(
     Ok(())
 }
 
+#[tokio::test]
+async fn skills_list_skips_cwd_roots_when_environment_disabled() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let cwd = TempDir::new()?;
+    let extra_root = TempDir::new()?;
+    write_skill(&codex_home, "home-skill")?;
+    write_skill(&extra_root, "extra-skill")?;
+
+    let mut mcp = McpProcess::new_with_env(
+        codex_home.path(),
+        &[(CODEX_EXEC_SERVER_URL_ENV_VAR, Some("none"))],
+    )
+    .await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let request_id = mcp
+        .send_skills_list_request(SkillsListParams {
+            cwds: vec![cwd.path().to_path_buf()],
+            force_reload: true,
+            per_cwd_extra_user_roots: Some(vec![SkillsListExtraRootsForCwd {
+                cwd: cwd.path().to_path_buf(),
+                extra_user_roots: vec![extra_root.path().to_path_buf()],
+            }]),
+        })
+        .await?;
+
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let SkillsListResponse { data } = to_response(response)?;
+    assert_eq!(data.len(), 1);
+    assert_eq!(data[0].cwd, cwd.path().to_path_buf());
+    assert_eq!(data[0].errors, Vec::new());
+    assert!(
+        data[0]
+            .skills
+            .iter()
+            .any(|skill| skill.name == "home-skill")
+    );
+    assert!(
+        data[0]
+            .skills
+            .iter()
+            .all(|skill| skill.name != "extra-skill")
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn skills_list_rejects_relative_extra_user_roots() -> Result<()> {
     let codex_home = TempDir::new()?;

codex-rs/cli/src/marketplace_cmd.rs

@@ -23,7 +23,8 @@ enum MarketplaceSubcommand {
 
 #[derive(Debug, Parser)]
 struct AddMarketplaceArgs {
-    /// Marketplace source. Supports owner/repo[@ref], HTTP(S) Git URLs, or SSH URLs.
+    /// Marketplace source. Supports owner/repo[@ref], HTTP(S) Git URLs, SSH URLs,
+    /// or local marketplace root directories.
     source: String,
 
     /// Git ref to check out. Overrides any @ref or #ref suffix in SOURCE.

codex-rs/cli/tests/marketplace_add.rs

@@ -1,6 +1,8 @@
 use anyhow::Result;
+use codex_config::CONFIG_TOML_FILE;
 use codex_core::plugins::marketplace_install_root;
 use predicates::str::contains;
+use pretty_assertions::assert_eq;
 use std::path::Path;
 use tempfile::TempDir;
 
@@ -37,7 +39,7 @@ fn write_marketplace_source(source: &Path, marker: &str) -> Result<()> {
 }
 
 #[tokio::test]
-async fn marketplace_add_rejects_local_directory_source() -> Result<()> {
+async fn marketplace_add_local_directory_source() -> Result<()> {
     let codex_home = TempDir::new()?;
     let source = TempDir::new()?;
     write_marketplace_source(source.path(), "local ref")?;
@@ -48,16 +50,40 @@ async fn marketplace_add_rejects_local_directory_source() -> Result<()> {
         .current_dir(source_parent)
         .args(["marketplace", "add", source_arg.as_str()])
         .assert()
-        .failure()
-        .stderr(contains(
-            "local marketplace sources are not supported yet; use an HTTP(S) Git URL, SSH Git URL, or GitHub owner/repo",
-        ));
+        .success();
 
-    assert!(
-        !marketplace_install_root(codex_home.path())
-            .join("debug")
-            .exists()
+    let installed_root = marketplace_install_root(codex_home.path()).join("debug");
+    assert!(!installed_root.exists());
+
+    let config = std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE))?;
+    let config: toml::Value = toml::from_str(&config)?;
+    let expected_source = source.path().canonicalize()?.display().to_string();
+    assert_eq!(
+        config["marketplaces"]["debug"]["source_type"].as_str(),
+        Some("local")
+    );
+    assert_eq!(
+        config["marketplaces"]["debug"]["source"].as_str(),
+        Some(expected_source.as_str())
     );
 
     Ok(())
 }
+
+#[tokio::test]
+async fn marketplace_add_rejects_local_manifest_file_source() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let source = TempDir::new()?;
+    write_marketplace_source(source.path(), "local ref")?;
+    let manifest_path = source.path().join(".agents/plugins/marketplace.json");
+
+    codex_command(codex_home.path())?
+        .args(["marketplace", "add", manifest_path.to_str().unwrap()])
+        .assert()
+        .failure()
+        .stderr(contains(
+            "local marketplace source must be a directory, not a file",
+        ));
+
+    Ok(())
+}

codex-rs/config/src/types.rs

@@ -632,6 +632,7 @@ pub struct MarketplaceConfig {
 #[serde(rename_all = "snake_case")]
 pub enum MarketplaceSourceType {
     Git,
+    Local,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema)]

codex-rs/core-skills/Cargo.toml

@@ -17,6 +17,7 @@ anyhow = { workspace = true }
 codex-analytics = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-config = { workspace = true }
+codex-exec-server = { workspace = true }
 codex-instructions = { workspace = true }
 codex-login = { workspace = true }
 codex-otel = { workspace = true }

codex-rs/core-skills/src/injection.rs

@@ -1,19 +1,21 @@
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::sync::Arc;
 
+use crate::SkillLoadOutcome;
 use crate::SkillMetadata;
 use crate::build_skill_name_counts;
 use codex_analytics::AnalyticsEventsClient;
 use codex_analytics::InvocationType;
 use codex_analytics::SkillInvocation;
 use codex_analytics::TrackEventsContext;
+use codex_exec_server::LOCAL_FS;
 use codex_instructions::SkillInstructions;
 use codex_otel::SessionTelemetry;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::user_input::UserInput;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_plugins::mention_syntax::TOOL_MENTION_SIGIL;
-use tokio::fs;
 
 #[derive(Debug, Default)]
 pub struct SkillInjections {
@@ -23,6 +25,7 @@ pub struct SkillInjections {
 
 pub async fn build_skill_injections(
     mentioned_skills: &[SkillMetadata],
+    loaded_skills: Option<&SkillLoadOutcome>,
     otel: Option<&SessionTelemetry>,
     analytics_client: &AnalyticsEventsClient,
     tracking: TrackEventsContext,
@@ -38,7 +41,13 @@ pub async fn build_skill_injections(
     let mut invocations = Vec::new();
 
     for skill in mentioned_skills {
-        match fs::read_to_string(&skill.path_to_skills_md).await {
+        let fs = loaded_skills
+            .and_then(|outcome| outcome.file_system_for_skill(skill))
+            .unwrap_or_else(|| Arc::clone(&LOCAL_FS));
+        match fs
+            .read_file_text(&skill.path_to_skills_md, /*sandbox*/ None)
+            .await
+        {
             Ok(contents) => {
                 emit_skill_injected_metric(otel, skill, "ok");
                 invocations.push(SkillInvocation {

codex-rs/core-skills/src/loader.rs

@@ -1,5 +1,6 @@
 use crate::model::SkillDependencies;
 use crate::model::SkillError;
+use crate::model::SkillFileSystemsByPath;
 use crate::model::SkillInterface;
 use crate::model::SkillLoadOutcome;
 use crate::model::SkillMetadata;
@@ -12,22 +13,24 @@ use codex_config::ConfigLayerStackOrdering;
 use codex_config::default_project_root_markers;
 use codex_config::merge_toml_values;
 use codex_config::project_root_markers_from_config;
+use codex_exec_server::ExecutorFileSystem;
+use codex_exec_server::LOCAL_FS;
 use codex_protocol::protocol::Product;
 use codex_protocol::protocol::SkillScope;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_absolute_path::AbsolutePathBufGuard;
 use codex_utils_plugins::plugin_namespace_for_skill_path;
 use dirs::home_dir;
-use dunce::canonicalize as canonicalize_path;
 use serde::Deserialize;
+use std::collections::HashMap;
 use std::collections::HashSet;
 use std::collections::VecDeque;
 use std::error::Error;
 use std::fmt;
-use std::fs;
+use std::io;
 use std::path::Component;
-use std::path::Path;
 use std::path::PathBuf;
+use std::sync::Arc;
 use toml::Value as TomlValue;
 use tracing::error;
 
@@ -148,21 +151,38 @@ impl Error for SkillParseError {}
 pub struct SkillRoot {
     pub path: AbsolutePathBuf,
     pub scope: SkillScope,
+    pub file_system: Arc<dyn ExecutorFileSystem>,
 }
 
-pub fn load_skills_from_roots<I>(roots: I) -> SkillLoadOutcome
+pub async fn load_skills_from_roots<I>(roots: I) -> SkillLoadOutcome
 where
     I: IntoIterator<Item = SkillRoot>,
 {
     let mut outcome = SkillLoadOutcome::default();
+    let mut file_systems_by_skill_path: HashMap<AbsolutePathBuf, Arc<dyn ExecutorFileSystem>> =
+        HashMap::new();
     for root in roots {
-        discover_skills_under_root(&root.path, root.scope, &mut outcome);
+        let fs = root.file_system;
+        let skills_before_root = outcome.skills.len();
+        discover_skills_under_root(fs.as_ref(), &root.path, root.scope, &mut outcome).await;
+        for skill in &outcome.skills[skills_before_root..] {
+            file_systems_by_skill_path
+                .entry(skill.path_to_skills_md.clone())
+                .or_insert_with(|| Arc::clone(&fs));
+        }
     }
 
     let mut seen: HashSet<AbsolutePathBuf> = HashSet::new();
     outcome
         .skills
         .retain(|skill| seen.insert(skill.path_to_skills_md.clone()));
+    let retained_skill_paths: HashSet<AbsolutePathBuf> = outcome
+        .skills
+        .iter()
+        .map(|skill| skill.path_to_skills_md.clone())
+        .collect();
+    file_systems_by_skill_path.retain(|path, _| retained_skill_paths.contains(path));
+    outcome.file_systems_by_skill_path = SkillFileSystemsByPath::new(file_systems_by_skill_path);
 
     fn scope_rank(scope: SkillScope) -> u8 {
         // Higher-priority scopes first (matches root scan order for dedupe).
@@ -184,7 +204,8 @@ where
     outcome
 }
 
-pub(crate) fn skill_roots(
+pub(crate) async fn skill_roots(
+    fs: Option<Arc<dyn ExecutorFileSystem>>,
     config_layer_stack: &ConfigLayerStack,
     cwd: &AbsolutePathBuf,
     plugin_skill_roots: Vec<AbsolutePathBuf>,
@@ -192,25 +213,29 @@ pub(crate) fn skill_roots(
     let home_dir =
         home_dir().and_then(|path| AbsolutePathBuf::from_absolute_path_checked(path).ok());
     skill_roots_with_home_dir(
+        fs,
         config_layer_stack,
         cwd,
         home_dir.as_ref(),
         plugin_skill_roots,
     )
+    .await
 }
 
-fn skill_roots_with_home_dir(
+async fn skill_roots_with_home_dir(
+    fs: Option<Arc<dyn ExecutorFileSystem>>,
     config_layer_stack: &ConfigLayerStack,
     cwd: &AbsolutePathBuf,
     home_dir: Option<&AbsolutePathBuf>,
     plugin_skill_roots: Vec<AbsolutePathBuf>,
 ) -> Vec<SkillRoot> {
-    let mut roots = skill_roots_from_layer_stack_inner(config_layer_stack, home_dir);
+    let mut roots = skill_roots_from_layer_stack_inner(config_layer_stack, home_dir, fs.clone());
     roots.extend(plugin_skill_roots.into_iter().map(|path| SkillRoot {
         path,
         scope: SkillScope::User,
+        file_system: Arc::clone(&LOCAL_FS),
     }));
-    roots.extend(repo_agents_skill_roots(config_layer_stack, cwd));
+    roots.extend(repo_agents_skill_roots(fs, config_layer_stack, cwd).await);
     dedupe_skill_roots_by_path(&mut roots);
     roots
 }
@@ -218,6 +243,7 @@ fn skill_roots_with_home_dir(
 fn skill_roots_from_layer_stack_inner(
     config_layer_stack: &ConfigLayerStack,
     home_dir: Option<&AbsolutePathBuf>,
+    repo_fs: Option<Arc<dyn ExecutorFileSystem>>,
 ) -> Vec<SkillRoot> {
     let mut roots = Vec::new();
 
@@ -231,10 +257,13 @@ fn skill_roots_from_layer_stack_inner(
 
         match &layer.name {
             ConfigLayerSource::Project { .. } => {
-                roots.push(SkillRoot {
-                    path: config_folder.join(SKILLS_DIR_NAME),
-                    scope: SkillScope::Repo,
-                });
+                if let Some(repo_fs) = &repo_fs {
+                    roots.push(SkillRoot {
+                        path: config_folder.join(SKILLS_DIR_NAME),
+                        scope: SkillScope::Repo,
+                        file_system: Arc::clone(repo_fs),
+                    });
+                }
             }
             ConfigLayerSource::User { .. } => {
                 // Deprecated user skills location (`$CODEX_HOME/skills`), kept for backward
@@ -242,6 +271,7 @@ fn skill_roots_from_layer_stack_inner(
                 roots.push(SkillRoot {
                     path: config_folder.join(SKILLS_DIR_NAME),
                     scope: SkillScope::User,
+                    file_system: Arc::clone(&LOCAL_FS),
                 });
 
                 // `$HOME/.agents/skills` (user-installed skills).
@@ -249,6 +279,7 @@ fn skill_roots_from_layer_stack_inner(
                     roots.push(SkillRoot {
                         path: home_dir.join(AGENTS_DIR_NAME).join(SKILLS_DIR_NAME),
                         scope: SkillScope::User,
+                        file_system: Arc::clone(&LOCAL_FS),
                     });
                 }
 
@@ -257,6 +288,7 @@ fn skill_roots_from_layer_stack_inner(
                 roots.push(SkillRoot {
                     path: system_cache_root_dir(&config_folder),
                     scope: SkillScope::System,
+                    file_system: Arc::clone(&LOCAL_FS),
                 });
             }
             ConfigLayerSource::System { .. } => {
@@ -265,6 +297,7 @@ fn skill_roots_from_layer_stack_inner(
                 roots.push(SkillRoot {
                     path: config_folder.join(SKILLS_DIR_NAME),
                     scope: SkillScope::Admin,
+                    file_system: Arc::clone(&LOCAL_FS),
                 });
             }
             ConfigLayerSource::Mdm { .. }
@@ -277,21 +310,34 @@ fn skill_roots_from_layer_stack_inner(
     roots
 }
 
-fn repo_agents_skill_roots(
+async fn repo_agents_skill_roots(
+    fs: Option<Arc<dyn ExecutorFileSystem>>,
     config_layer_stack: &ConfigLayerStack,
     cwd: &AbsolutePathBuf,
 ) -> Vec<SkillRoot> {
+    let Some(fs) = fs else {
+        return Vec::new();
+    };
     let project_root_markers = project_root_markers_from_stack(config_layer_stack);
-    let project_root = find_project_root(cwd, &project_root_markers);
+    let project_root = find_project_root(fs.as_ref(), cwd, &project_root_markers).await;
     let dirs = dirs_between_project_root_and_cwd(cwd, &project_root);
     let mut roots = Vec::new();
     for dir in dirs {
         let agents_skills = dir.join(AGENTS_DIR_NAME).join(SKILLS_DIR_NAME);
-        if agents_skills.is_dir() {
-            roots.push(SkillRoot {
+        match fs.get_metadata(&agents_skills, /*sandbox*/ None).await {
+            Ok(metadata) if metadata.is_directory => roots.push(SkillRoot {
                 path: agents_skills,
                 scope: SkillScope::Repo,
-            });
+                file_system: Arc::clone(&fs),
+            }),
+            Ok(_) => {}
+            Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+            Err(err) => {
+                tracing::warn!(
+                    "failed to stat repo skills root {}: {err:#}",
+                    agents_skills.display()
+                );
+            }
         }
     }
     roots
@@ -319,16 +365,27 @@ fn project_root_markers_from_stack(config_layer_stack: &ConfigLayerStack) -> Vec
     }
 }
 
-fn find_project_root(cwd: &AbsolutePathBuf, project_root_markers: &[String]) -> AbsolutePathBuf {
+async fn find_project_root(
+    fs: &dyn ExecutorFileSystem,
+    cwd: &AbsolutePathBuf,
+    project_root_markers: &[String],
+) -> AbsolutePathBuf {
     if project_root_markers.is_empty() {
         return cwd.clone();
     }
 
-    for path in cwd.ancestors() {
+    for ancestor in cwd.ancestors() {
         for marker in project_root_markers {
-            let marker_path = path.join(marker);
-            if marker_path.exists() {
-                return path;
+            let marker_path = ancestor.join(marker);
+            match fs.get_metadata(&marker_path, /*sandbox*/ None).await {
+                Ok(_) => return ancestor,
+                Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+                Err(err) => {
+                    tracing::warn!(
+                        "failed to stat project root marker {}: {err:#}",
+                        marker_path.display()
+                    );
+                }
             }
         }
     }
@@ -362,24 +419,33 @@ fn dedupe_skill_roots_by_path(roots: &mut Vec<SkillRoot>) {
     roots.retain(|root| seen.insert(root.path.clone()));
 }
 
-fn discover_skills_under_root(
+fn canonicalize_for_skill_identity(path: &AbsolutePathBuf) -> AbsolutePathBuf {
+    path.canonicalize().unwrap_or_else(|_| path.clone())
+}
+
+async fn discover_skills_under_root(
+    fs: &dyn ExecutorFileSystem,
     root: &AbsolutePathBuf,
     scope: SkillScope,
     outcome: &mut SkillLoadOutcome,
 ) {
-    let Ok(root) = canonicalize_path(root.as_path()) else {
-        return;
-    };
+    let root = canonicalize_for_skill_identity(root);
 
-    if !root.is_dir() {
-        return;
+    match fs.get_metadata(&root, /*sandbox*/ None).await {
+        Ok(metadata) if metadata.is_directory => {}
+        Ok(_) => return,
+        Err(err) if err.kind() == io::ErrorKind::NotFound => return,
+        Err(err) => {
+            error!("failed to stat skills root {}: {err:#}", root.display());
+            return;
+        }
     }
 
     fn enqueue_dir(
-        queue: &mut VecDeque<(PathBuf, usize)>,
-        visited_dirs: &mut HashSet<PathBuf>,
+        queue: &mut VecDeque<(AbsolutePathBuf, usize)>,
+        visited_dirs: &mut HashSet<AbsolutePathBuf>,
         truncated_by_dir_limit: &mut bool,
-        path: PathBuf,
+        path: AbsolutePathBuf,
         depth: usize,
     ) {
         if depth > MAX_SCAN_DEPTH {
@@ -400,14 +466,14 @@ fn discover_skills_under_root(
         SkillScope::Repo | SkillScope::User | SkillScope::Admin
     );
 
-    let mut visited_dirs: HashSet<PathBuf> = HashSet::new();
+    let mut visited_dirs: HashSet<AbsolutePathBuf> = HashSet::new();
     visited_dirs.insert(root.clone());
 
-    let mut queue: VecDeque<(PathBuf, usize)> = VecDeque::from([(root.clone(), 0)]);
+    let mut queue: VecDeque<(AbsolutePathBuf, 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_dir(&dir) {
+        let entries = match fs.read_directory(&dir, /*sandbox*/ None).await {
             Ok(entries) => entries,
             Err(e) => {
                 error!("failed to read skills dir {}: {e:#}", dir.display());
@@ -415,65 +481,53 @@ fn discover_skills_under_root(
             }
         };
 
-        for entry in entries.flatten() {
-            let path = match AbsolutePathBuf::from_absolute_path_checked(entry.path()) {
-                Ok(path) => path,
-                Err(err) => {
-                    error!("failed to normalize skills entry path: {err:#}");
-                    continue;
-                }
-            };
-            let file_name = match path.file_name().and_then(|f| f.to_str()) {
-                Some(name) => name,
-                None => continue,
-            };
-
+        for entry in entries {
+            let file_name = entry.file_name;
             if file_name.starts_with('.') {
                 continue;
             }
 
-            let Ok(file_type) = entry.file_type() else {
-                continue;
+            let path = dir.join(&file_name);
+            let metadata = match fs.get_metadata(&path, /*sandbox*/ None).await {
+                Ok(metadata) => metadata,
+                Err(e) => {
+                    error!("failed to stat skills path {}: {e:#}", path.display());
+                    continue;
+                }
             };
 
-            if file_type.is_symlink() {
+            if metadata.is_symlink {
                 if !follow_symlinks {
                     continue;
                 }
-
-                // Follow the symlink to determine what it points to.
-                let metadata = match fs::metadata(&path) {
-                    Ok(metadata) => metadata,
-                    Err(e) => {
+                match fs.read_directory(&path, /*sandbox*/ None).await {
+                    Ok(_) => {
+                        let resolved_dir = canonicalize_for_skill_identity(&path);
+                        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) => {
                         error!(
-                            "failed to stat skills entry {} (symlink): {e:#}",
+                            "failed to read skills symlink dir {}: {err:#}",
                             path.display()
                         );
-                        continue;
                     }
-                };
-
-                if metadata.is_dir() {
-                    let Ok(resolved_dir) = canonicalize_path(&path) else {
-                        continue;
-                    };
-                    enqueue_dir(
-                        &mut queue,
-                        &mut visited_dirs,
-                        &mut truncated_by_dir_limit,
-                        resolved_dir,
-                        depth + 1,
-                    );
-                    continue;
                 }
-
                 continue;
             }
 
-            if file_type.is_dir() {
-                let Ok(resolved_dir) = canonicalize_path(&path) else {
-                    continue;
-                };
+            if metadata.is_directory {
+                let resolved_dir = canonicalize_for_skill_identity(&path);
                 enqueue_dir(
                     &mut queue,
                     &mut visited_dirs,
@@ -484,15 +538,15 @@ fn discover_skills_under_root(
                 continue;
             }
 
-            if file_type.is_file() && file_name == SKILLS_FILENAME {
-                match parse_skill_file(&path, scope) {
+            if metadata.is_file && file_name == SKILLS_FILENAME {
+                match parse_skill_file(fs, &path, scope).await {
                     Ok(skill) => {
                         outcome.skills.push(skill);
                     }
                     Err(err) => {
                         if scope != SkillScope::System {
                             outcome.errors.push(SkillError {
-                                path,
+                                path: path.clone(),
                                 message: err.to_string(),
                             });
                         }
@@ -511,11 +565,15 @@ fn discover_skills_under_root(
     }
 }
 
-fn parse_skill_file(
+async fn parse_skill_file(
+    fs: &dyn ExecutorFileSystem,
     path: &AbsolutePathBuf,
     scope: SkillScope,
 ) -> Result<SkillMetadata, SkillParseError> {
-    let contents = fs::read_to_string(path.as_path()).map_err(SkillParseError::Read)?;
+    let contents = fs
+        .read_file_text(path, /*sandbox*/ None)
+        .await
+        .map_err(SkillParseError::Read)?;
 
     let frontmatter = extract_frontmatter(&contents).ok_or(SkillParseError::MissingFrontmatter)?;
 
@@ -527,8 +585,8 @@ fn parse_skill_file(
         .as_deref()
         .map(sanitize_single_line)
         .filter(|value| !value.is_empty())
-        .unwrap_or_else(|| default_skill_name(path.as_path()));
-    let name = namespaced_skill_name(path.as_path(), &base_name);
+        .unwrap_or_else(|| default_skill_name(path));
+    let name = namespaced_skill_name(fs, path, &base_name).await;
     let description = parsed
         .description
         .as_deref()
@@ -544,7 +602,7 @@ fn parse_skill_file(
         interface,
         dependencies,
         policy,
-    } = load_skill_metadata(path);
+    } = load_skill_metadata(fs, path).await;
 
     validate_len(&name, MAX_NAME_LEN, "name")?;
     validate_len(&description, MAX_DESCRIPTION_LEN, "description")?;
@@ -556,7 +614,7 @@ fn parse_skill_file(
         )?;
     }
 
-    let resolved_path = path.canonicalize().map_err(SkillParseError::Read)?;
+    let resolved_path = canonicalize_for_skill_identity(path);
 
     Ok(SkillMetadata {
         name,
@@ -570,22 +628,33 @@ fn parse_skill_file(
     })
 }
 
-fn default_skill_name(path: &Path) -> String {
+fn default_skill_name(path: &AbsolutePathBuf) -> String {
     path.parent()
-        .and_then(Path::file_name)
-        .and_then(|name| name.to_str())
-        .map(sanitize_single_line)
+        .and_then(|parent| {
+            parent
+                .file_name()
+                .and_then(|name| name.to_str())
+                .map(sanitize_single_line)
+        })
         .filter(|value| !value.is_empty())
         .unwrap_or_else(|| "skill".to_string())
 }
 
-fn namespaced_skill_name(path: &Path, base_name: &str) -> String {
-    plugin_namespace_for_skill_path(path)
+async fn namespaced_skill_name(
+    fs: &dyn ExecutorFileSystem,
+    path: &AbsolutePathBuf,
+    base_name: &str,
+) -> String {
+    plugin_namespace_for_skill_path(fs, path)
+        .await
         .map(|namespace| format!("{namespace}:{base_name}"))
         .unwrap_or_else(|| base_name.to_string())
 }
 
-fn load_skill_metadata(skill_path: &AbsolutePathBuf) -> LoadedSkillMetadata {
+async fn load_skill_metadata(
+    fs: &dyn ExecutorFileSystem,
+    skill_path: &AbsolutePathBuf,
+) -> LoadedSkillMetadata {
     // Fail open: optional metadata should not block loading SKILL.md.
     let Some(skill_dir) = skill_path.parent() else {
         return LoadedSkillMetadata::default();
@@ -593,11 +662,23 @@ fn load_skill_metadata(skill_path: &AbsolutePathBuf) -> LoadedSkillMetadata {
     let metadata_path = skill_dir
         .join(SKILLS_METADATA_DIR)
         .join(SKILLS_METADATA_FILENAME);
-    if !metadata_path.as_path().exists() {
-        return LoadedSkillMetadata::default();
+    match fs.get_metadata(&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 = metadata_path.display(),
+                label = SKILLS_METADATA_FILENAME
+            );
+            return LoadedSkillMetadata::default();
+        }
     }
 
-    let contents = match fs::read_to_string(metadata_path.as_path()) {
+    let contents = match fs.read_file_text(&metadata_path, /*sandbox*/ None).await {
         Ok(contents) => contents,
         Err(error) => {
             tracing::warn!(
@@ -863,12 +944,13 @@ fn extract_frontmatter(contents: &str) -> Option<String> {
     Some(frontmatter_lines.join("\n"))
 }
 #[cfg(test)]
-pub(crate) fn skill_roots_from_layer_stack(
+pub(crate) async fn skill_roots_from_layer_stack(
+    fs: Arc<dyn ExecutorFileSystem>,
     config_layer_stack: &ConfigLayerStack,
     cwd: &AbsolutePathBuf,
     home_dir: Option<&AbsolutePathBuf>,
 ) -> Vec<SkillRoot> {
-    skill_roots_with_home_dir(config_layer_stack, cwd, home_dir, Vec::new())
+    skill_roots_with_home_dir(Some(fs), config_layer_stack, cwd, home_dir, Vec::new()).await
 }
 
 #[cfg(test)]

codex-rs/core-skills/src/loader_tests.rs

@@ -4,14 +4,18 @@ use codex_config::ConfigLayerEntry;
 use codex_config::ConfigLayerStack;
 use codex_config::ConfigRequirements;
 use codex_config::ConfigRequirementsToml;
+use codex_exec_server::LOCAL_FS;
 use codex_protocol::protocol::Product;
 use codex_protocol::protocol::SkillScope;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_absolute_path::test_support::PathBufExt;
 use codex_utils_absolute_path::test_support::PathExt;
+use dunce::canonicalize as canonicalize_path;
 use pretty_assertions::assert_eq;
+use std::fs;
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::Arc;
 use tempfile::TempDir;
 use toml::Value as TomlValue;
 
@@ -113,14 +117,18 @@ async fn make_config_for_cwd(codex_home: &TempDir, cwd: PathBuf) -> TestConfig {
     }
 }
 
-fn load_skills_for_test(config: &TestConfig) -> SkillLoadOutcome {
+async fn load_skills_for_test(config: &TestConfig) -> SkillLoadOutcome {
     // Keep unit tests hermetic by never scanning the real `$HOME/.agents/skills`.
-    super::load_skills_from_roots(super::skill_roots_with_home_dir(
-        &config.config_layer_stack,
-        &config.cwd,
-        /*home_dir*/ None,
-        Vec::new(),
-    ))
+    super::load_skills_from_roots(
+        super::skill_roots_from_layer_stack(
+            Arc::clone(&LOCAL_FS),
+            &config.config_layer_stack,
+            &config.cwd,
+            /*home_dir*/ None,
+        )
+        .await,
+    )
+    .await
 }
 
 fn mark_as_git_repo(dir: &Path) {
@@ -135,8 +143,8 @@ fn normalized(path: &Path) -> AbsolutePathBuf {
         .abs()
 }
 
-#[test]
-fn skill_roots_from_layer_stack_maps_user_to_user_and_system_cache_and_system_to_admin()
+#[tokio::test]
+async fn skill_roots_from_layer_stack_maps_user_to_user_and_system_cache_and_system_to_admin()
 -> anyhow::Result<()> {
     let tmp = tempfile::tempdir()?;
 
@@ -167,10 +175,16 @@ fn skill_roots_from_layer_stack_maps_user_to_user_and_system_cache_and_system_to
     )?;
 
     let home_folder_abs = home_folder.abs();
-    let got = skill_roots_from_layer_stack(&stack, &home_folder_abs, Some(&home_folder_abs))
-        .into_iter()
-        .map(|root| (root.scope, root.path.to_path_buf()))
-        .collect::<Vec<_>>();
+    let got = skill_roots_from_layer_stack(
+        Arc::clone(&LOCAL_FS),
+        &stack,
+        &home_folder_abs,
+        Some(&home_folder_abs),
+    )
+    .await
+    .into_iter()
+    .map(|root| (root.scope, root.path.to_path_buf()))
+    .collect::<Vec<_>>();
 
     assert_eq!(
         got,
@@ -191,8 +205,8 @@ fn skill_roots_from_layer_stack_maps_user_to_user_and_system_cache_and_system_to
     Ok(())
 }
 
-#[test]
-fn skill_roots_from_layer_stack_includes_disabled_project_layers() -> anyhow::Result<()> {
+#[tokio::test]
+async fn skill_roots_from_layer_stack_includes_disabled_project_layers() -> anyhow::Result<()> {
     let tmp = tempfile::tempdir()?;
 
     let home_folder = tmp.path().join("home");
@@ -227,10 +241,16 @@ fn skill_roots_from_layer_stack_includes_disabled_project_layers() -> anyhow::Re
 
     let home_folder_abs = home_folder.abs();
     let project_root_abs = project_root.abs();
-    let got = skill_roots_from_layer_stack(&stack, &project_root_abs, Some(&home_folder_abs))
-        .into_iter()
-        .map(|root| (root.scope, root.path.to_path_buf()))
-        .collect::<Vec<_>>();
+    let got = skill_roots_from_layer_stack(
+        Arc::clone(&LOCAL_FS),
+        &stack,
+        &project_root_abs,
+        Some(&home_folder_abs),
+    )
+    .await
+    .into_iter()
+    .map(|root| (root.scope, root.path.to_path_buf()))
+    .collect::<Vec<_>>();
 
     assert_eq!(
         got,
@@ -251,8 +271,8 @@ fn skill_roots_from_layer_stack_includes_disabled_project_layers() -> anyhow::Re
     Ok(())
 }
 
-#[test]
-fn loads_skills_from_home_agents_dir_for_user_scope() -> anyhow::Result<()> {
+#[tokio::test]
+async fn loads_skills_from_home_agents_dir_for_user_scope() -> anyhow::Result<()> {
     let tmp = tempfile::tempdir()?;
 
     let home_folder = tmp.path().join("home");
@@ -278,11 +298,14 @@ fn loads_skills_from_home_agents_dir_for_user_scope() -> anyhow::Result<()> {
     );
 
     let home_folder_abs = home_folder.abs();
-    let outcome = load_skills_from_roots(skill_roots_from_layer_stack(
+    let roots = skill_roots_from_layer_stack(
+        Arc::clone(&LOCAL_FS),
         &stack,
         &home_folder_abs,
         Some(&home_folder_abs),
-    ));
+    )
+    .await;
+    let outcome = load_skills_from_roots(roots).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -396,7 +419,7 @@ async fn loads_skill_dependencies_metadata_from_yaml() {
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -474,7 +497,7 @@ interface:
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -523,7 +546,7 @@ policy:
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -555,7 +578,7 @@ policy: {}
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -594,7 +617,7 @@ policy:
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -632,7 +655,7 @@ async fn accepts_icon_paths_under_assets_dir() {
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -679,7 +702,7 @@ async fn ignores_invalid_brand_color() {
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -725,7 +748,7 @@ async fn ignores_default_prompt_over_max_length() {
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -773,7 +796,7 @@ async fn drops_interface_when_icons_are_invalid() {
     );
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -817,7 +840,7 @@ async fn loads_skills_via_symlinked_subdir_for_user_scope() {
     symlink_dir(shared.path(), &codex_home.path().join("skills/shared"));
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -852,7 +875,7 @@ async fn ignores_symlinked_skill_file_for_user_scope() {
     symlink_file(&shared_skill_path, &skill_dir.join(SKILLS_FILENAME));
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -876,7 +899,7 @@ async fn does_not_loop_on_symlink_cycle_for_user_scope() {
     let skill_path = write_skill_at(&cycle_dir, "demo", "cycle-skill", "still loads");
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -898,9 +921,9 @@ async fn does_not_loop_on_symlink_cycle_for_user_scope() {
     );
 }
 
-#[test]
+#[tokio::test]
 #[cfg(unix)]
-fn loads_skills_via_symlinked_subdir_for_admin_scope() {
+async fn loads_skills_via_symlinked_subdir_for_admin_scope() {
     let admin_root = tempfile::tempdir().expect("tempdir");
     let shared = tempfile::tempdir().expect("tempdir");
 
@@ -912,7 +935,9 @@ fn loads_skills_via_symlinked_subdir_for_admin_scope() {
     let outcome = load_skills_from_roots([SkillRoot {
         path: admin_root.path().abs(),
         scope: SkillScope::Admin,
-    }]);
+        file_system: Arc::clone(&LOCAL_FS),
+    }])
+    .await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -951,7 +976,7 @@ async fn loads_skills_via_symlinked_subdir_for_repo_scope() {
     symlink_dir(shared.path(), &repo_skills_root.join("shared"));
 
     let cfg = make_config_for_cwd(&codex_home, repo_dir.path().to_path_buf()).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -988,7 +1013,9 @@ async fn system_scope_ignores_symlinked_subdir() {
     let outcome = load_skills_from_roots([SkillRoot {
         path: system_root.abs(),
         scope: SkillScope::System,
-    }]);
+        file_system: Arc::clone(&LOCAL_FS),
+    }])
+    .await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1018,7 +1045,9 @@ async fn respects_max_scan_depth_for_user_scope() {
     let outcome = load_skills_from_roots([SkillRoot {
         path: skills_root.abs(),
         scope: SkillScope::User,
-    }]);
+        file_system: Arc::clone(&LOCAL_FS),
+    }])
+    .await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -1046,7 +1075,7 @@ async fn loads_valid_skill() {
     let skill_path = write_skill(&codex_home, "demo", "demo-skill", "does things\ncarefully");
     let cfg = make_config(&codex_home).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1077,7 +1106,7 @@ async fn falls_back_to_directory_name_when_skill_name_is_missing() {
     );
     let cfg = make_config(&codex_home).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -1118,7 +1147,9 @@ async fn namespaces_plugin_skills_using_plugin_name() {
     let outcome = load_skills_from_roots([SkillRoot {
         path: plugin_root.join("skills").abs(),
         scope: SkillScope::User,
-    }]);
+        file_system: Arc::clone(&LOCAL_FS),
+    }])
+    .await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -1150,7 +1181,7 @@ async fn loads_short_description_from_metadata() {
     fs::write(&skill_path, contents).unwrap();
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1183,7 +1214,7 @@ async fn enforces_short_description_length_limits() {
     fs::write(skill_dir.join(SKILLS_FILENAME), contents).unwrap();
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert_eq!(outcome.skills.len(), 0);
     assert_eq!(outcome.errors.len(), 1);
     assert!(
@@ -1212,7 +1243,7 @@ async fn skips_hidden_and_invalid() {
     fs::write(invalid_dir.join(SKILLS_FILENAME), "---\nname: bad").unwrap();
 
     let cfg = make_config(&codex_home).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert_eq!(outcome.skills.len(), 0);
     assert_eq!(outcome.errors.len(), 1);
     assert!(
@@ -1230,7 +1261,7 @@ async fn enforces_length_limits() {
     write_skill(&codex_home, "max-len", "max-len", &max_desc);
     let cfg = make_config(&codex_home).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1240,7 +1271,7 @@ async fn enforces_length_limits() {
 
     let too_long_desc = "\u{1F4A1}".repeat(MAX_DESCRIPTION_LEN + 1);
     write_skill(&codex_home, "too-long", "too-long", &too_long_desc);
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert_eq!(outcome.skills.len(), 1);
     assert_eq!(outcome.errors.len(), 1);
     assert!(
@@ -1262,7 +1293,7 @@ async fn loads_skills_from_repo_root() {
     let skill_path = write_skill_at(&skills_root, "repo", "repo-skill", "from repo");
     let cfg = make_config_for_cwd(&codex_home, repo_dir.path().to_path_buf()).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1297,7 +1328,7 @@ async fn loads_skills_from_agents_dir_without_codex_dir() {
     );
     let cfg = make_config_for_cwd(&codex_home, repo_dir.path().to_path_buf()).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1349,7 +1380,7 @@ async fn loads_skills_from_all_codex_dirs_under_project_root() {
 
     let cfg = make_config_for_cwd(&codex_home, nested_dir).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1399,7 +1430,7 @@ async fn loads_skills_from_codex_dir_when_not_git_repo() {
 
     let cfg = make_config_for_cwd(&codex_home, work_dir.path().to_path_buf()).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1430,12 +1461,15 @@ async fn deduplicates_by_path_preferring_first_root() {
         SkillRoot {
             path: root.path().abs(),
             scope: SkillScope::Repo,
+            file_system: Arc::clone(&LOCAL_FS),
         },
         SkillRoot {
             path: root.path().abs(),
             scope: SkillScope::User,
+            file_system: Arc::clone(&LOCAL_FS),
         },
-    ]);
+    ])
+    .await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -1476,7 +1510,7 @@ async fn keeps_duplicate_names_from_repo_and_user() {
 
     let cfg = make_config_for_cwd(&codex_home, repo_dir.path().to_path_buf()).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1539,7 +1573,7 @@ async fn keeps_duplicate_names_from_nested_codex_dirs() {
     );
 
     let cfg = make_config_for_cwd(&codex_home, nested_dir).await;
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
 
     assert!(
         outcome.errors.is_empty(),
@@ -1601,7 +1635,7 @@ async fn repo_skills_search_does_not_escape_repo_root() {
 
     let cfg = make_config_for_cwd(&codex_home, repo_dir).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1630,7 +1664,7 @@ async fn loads_skills_when_cwd_is_file_in_repo() {
 
     let cfg = make_config_for_cwd(&codex_home, file_path).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1670,7 +1704,7 @@ async fn non_git_repo_skills_search_does_not_walk_parents() {
 
     let cfg = make_config_for_cwd(&codex_home, nested_dir).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1688,7 +1722,7 @@ async fn loads_skills_from_system_cache_when_present() {
 
     let cfg = make_config_for_cwd(&codex_home, work_dir.path().to_path_buf()).await;
 
-    let outcome = load_skills_for_test(&cfg);
+    let outcome = load_skills_for_test(&cfg).await;
     assert!(
         outcome.errors.is_empty(),
         "unexpected errors: {:?}",
@@ -1714,10 +1748,16 @@ async fn skill_roots_include_admin_with_lowest_priority() {
     let codex_home = tempfile::tempdir().expect("tempdir");
     let cfg = make_config(&codex_home).await;
 
-    let scopes: Vec<SkillScope> = super::skill_roots(&cfg.config_layer_stack, &cfg.cwd, Vec::new())
-        .into_iter()
-        .map(|root| root.scope)
-        .collect();
+    let scopes: Vec<SkillScope> = super::skill_roots(
+        Some(Arc::clone(&LOCAL_FS)),
+        &cfg.config_layer_stack,
+        &cfg.cwd,
+        Vec::new(),
+    )
+    .await
+    .into_iter()
+    .map(|root| root.scope)
+    .collect();
     let mut expected = vec![SkillScope::User, SkillScope::System];
     if home_dir().is_some() {
         expected.insert(1, SkillScope::User);

codex-rs/core-skills/src/manager.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::sync::RwLock;
 
 use codex_config::ConfigLayerStack;
+use codex_exec_server::ExecutorFileSystem;
 use codex_protocol::protocol::Product;
 use codex_protocol::protocol::SkillScope;
 use codex_utils_absolute_path::AbsolutePathBuf;
@@ -85,15 +86,19 @@ impl SkillsManager {
     /// This path uses a cache keyed by the effective skill-relevant config state rather than just
     /// cwd so role-local and session-local skill overrides cannot bleed across sessions that happen
     /// to share a directory.
-    pub fn skills_for_config(&self, input: &SkillsLoadInput) -> SkillLoadOutcome {
-        let roots = self.skill_roots_for_config(input);
+    pub async fn skills_for_config(
+        &self,
+        input: &SkillsLoadInput,
+        fs: Option<Arc<dyn ExecutorFileSystem>>,
+    ) -> SkillLoadOutcome {
+        let roots = self.skill_roots_for_config(input, fs).await;
         let skill_config_rules = skill_config_rules_from_stack(&input.config_layer_stack);
         let cache_key = config_skills_cache_key(&roots, &skill_config_rules);
         if let Some(outcome) = self.cached_outcome_for_config(&cache_key) {
             return outcome;
         }
 
-        let outcome = self.build_skill_outcome(roots, &skill_config_rules);
+        let outcome = self.build_skill_outcome(roots, &skill_config_rules).await;
         let mut cache = self
             .cache_by_config
             .write()
@@ -102,12 +107,18 @@ impl SkillsManager {
         outcome
     }
 
-    pub fn skill_roots_for_config(&self, input: &SkillsLoadInput) -> Vec<SkillRoot> {
+    pub async fn skill_roots_for_config(
+        &self,
+        input: &SkillsLoadInput,
+        fs: Option<Arc<dyn ExecutorFileSystem>>,
+    ) -> Vec<SkillRoot> {
         let mut roots = skill_roots(
+            fs,
             &input.config_layer_stack,
             &input.cwd,
             input.effective_skill_roots.clone(),
-        );
+        )
+        .await;
         if !input.bundled_skills_enabled {
             roots.retain(|root| root.scope != SkillScope::System);
         }
@@ -118,12 +129,9 @@ impl SkillsManager {
         &self,
         input: &SkillsLoadInput,
         force_reload: bool,
+        fs: Option<Arc<dyn ExecutorFileSystem>>,
     ) -> SkillLoadOutcome {
-        if !force_reload && let Some(outcome) = self.cached_outcome_for_cwd(&input.cwd) {
-            return outcome;
-        }
-
-        self.skills_for_cwd_with_extra_user_roots(input, force_reload, &[])
+        self.skills_for_cwd_with_extra_user_roots(input, force_reload, &[], fs)
             .await
     }
 
@@ -132,46 +140,56 @@ impl SkillsManager {
         input: &SkillsLoadInput,
         force_reload: bool,
         extra_user_roots: &[AbsolutePathBuf],
+        fs: Option<Arc<dyn ExecutorFileSystem>>,
     ) -> SkillLoadOutcome {
-        if !force_reload && let Some(outcome) = self.cached_outcome_for_cwd(&input.cwd) {
+        let use_cwd_cache = fs.is_some();
+        if use_cwd_cache
+            && !force_reload
+            && let Some(outcome) = self.cached_outcome_for_cwd(&input.cwd)
+        {
             return outcome;
         }
-        let normalized_extra_user_roots = normalize_extra_user_roots(extra_user_roots);
 
         let mut roots = skill_roots(
+            fs.clone(),
             &input.config_layer_stack,
             &input.cwd,
             input.effective_skill_roots.clone(),
-        );
+        )
+        .await;
         if !bundled_skills_enabled_from_stack(&input.config_layer_stack) {
             roots.retain(|root| root.scope != SkillScope::System);
         }
-        roots.extend(
-            normalized_extra_user_roots
-                .iter()
-                .cloned()
-                .map(|path| SkillRoot {
-                    path,
-                    scope: SkillScope::User,
-                }),
-        );
+        if let Some(fs) = fs {
+            roots.extend(
+                normalize_extra_user_roots(extra_user_roots)
+                    .into_iter()
+                    .map(|path| SkillRoot {
+                        path,
+                        scope: SkillScope::User,
+                        file_system: Arc::clone(&fs),
+                    }),
+            );
+        }
         let skill_config_rules = skill_config_rules_from_stack(&input.config_layer_stack);
-        let outcome = self.build_skill_outcome(roots, &skill_config_rules);
-        let mut cache = self
-            .cache_by_cwd
-            .write()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        cache.insert(input.cwd.clone(), outcome.clone());
+        let outcome = self.build_skill_outcome(roots, &skill_config_rules).await;
+        if use_cwd_cache {
+            let mut cache = self
+                .cache_by_cwd
+                .write()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            cache.insert(input.cwd.clone(), outcome.clone());
+        }
         outcome
     }
 
-    fn build_skill_outcome(
+    async fn build_skill_outcome(
         &self,
         roots: Vec<SkillRoot>,
         skill_config_rules: &SkillConfigRules,
     ) -> SkillLoadOutcome {
         let outcome = crate::filter_skill_load_outcome_for_product(
-            load_skills_from_roots(roots),
+            load_skills_from_roots(roots).await,
             self.restriction_product,
         );
         let disabled_paths = resolve_disabled_skill_paths(&outcome.skills, skill_config_rules);

codex-rs/core-skills/src/manager_tests.rs

@@ -7,6 +7,7 @@ use codex_config::CONFIG_TOML_FILE;
 use codex_config::ConfigLayerEntry;
 use codex_config::ConfigLayerStack;
 use codex_config::ConfigRequirementsToml;
+use codex_exec_server::LOCAL_FS;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_absolute_path::test_support::PathBufExt;
 use codex_utils_absolute_path::test_support::PathExt;
@@ -15,6 +16,7 @@ use pretty_assertions::assert_eq;
 use std::collections::HashSet;
 use std::fs;
 use std::path::PathBuf;
+use std::sync::Arc;
 use tempfile::TempDir;
 
 fn write_user_skill(codex_home: &TempDir, dir: &str, name: &str, description: &str) {
@@ -136,22 +138,21 @@ enabled = {enabled}
     )
 }
 
-fn skills_for_config_with_stack(
+async fn skills_for_config_with_stack(
     skills_manager: &SkillsManager,
     cwd: &TempDir,
     config_layer_stack: &ConfigLayerStack,
-    effective_skill_roots: &[PathBuf],
+    effective_skill_roots: &[AbsolutePathBuf],
 ) -> SkillLoadOutcome {
     let skills_input = SkillsLoadInput::new(
         cwd.path().abs(),
-        effective_skill_roots
-            .iter()
-            .map(codex_utils_absolute_path::test_support::PathBufExt::abs)
-            .collect(),
+        effective_skill_roots.to_vec(),
         config_layer_stack.clone(),
         bundled_skills_enabled_from_stack(config_layer_stack),
     );
-    skills_manager.skills_for_config(&skills_input)
+    skills_manager
+        .skills_for_config(&skills_input, Some(Arc::clone(&LOCAL_FS)))
+        .await
 }
 
 #[test]
@@ -184,7 +185,8 @@ async fn skills_for_config_reuses_cache_for_same_effective_config() {
     );
 
     write_user_skill(&codex_home, "a", "skill-a", "from a");
-    let outcome1 = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]);
+    let outcome1 =
+        skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]).await;
     assert!(
         outcome1.skills.iter().any(|s| s.name == "skill-a"),
         "expected skill-a to be discovered"
@@ -193,7 +195,8 @@ async fn skills_for_config_reuses_cache_for_same_effective_config() {
     // Write a new skill after the first call; the second call should reuse the config-aware cache
     // entry because the effective skill config is unchanged.
     write_user_skill(&codex_home, "b", "skill-b", "from b");
-    let outcome2 = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]);
+    let outcome2 =
+        skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]).await;
     assert_eq!(outcome2.errors, outcome1.errors);
     assert_eq!(outcome2.skills, outcome1.skills);
 }
@@ -218,7 +221,7 @@ async fn skills_for_config_disables_plugin_skills_by_name() {
         .parent()
         .and_then(std::path::Path::parent)
         .expect("plugin skill should live under a skills root")
-        .to_path_buf();
+        .abs();
     let skills_manager = SkillsManager::new(
         codex_home.path().abs(),
         /*bundled_skills_enabled*/ true,
@@ -229,7 +232,8 @@ async fn skills_for_config_disables_plugin_skills_by_name() {
         &cwd,
         &config_layer_stack,
         &[plugin_skill_root],
-    );
+    )
+    .await;
     let skill = outcome
         .skills
         .iter()
@@ -259,7 +263,7 @@ async fn skills_for_cwd_reuses_cached_entry_even_when_entry_has_extra_roots() {
         codex_home.path().abs(),
         /*bundled_skills_enabled*/ true,
     );
-    let _ = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]);
+    let _ = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]).await;
 
     write_user_skill(&extra_root, "x", "extra-skill", "from extra root");
     let extra_root_path = extra_root.path().abs();
@@ -274,6 +278,7 @@ async fn skills_for_cwd_reuses_cached_entry_even_when_entry_has_extra_roots() {
             &base_input,
             /*force_reload*/ true,
             std::slice::from_ref(&extra_root_path),
+            Some(Arc::clone(&LOCAL_FS)),
         )
         .await;
     assert!(
@@ -298,12 +303,150 @@ async fn skills_for_cwd_reuses_cached_entry_even_when_entry_has_extra_roots() {
         bundled_skills_enabled_from_stack(&config_layer_stack),
     );
     let outcome_without_extra = skills_manager
-        .skills_for_cwd(&base_input, /*force_reload*/ false)
+        .skills_for_cwd(
+            &base_input,
+            /*force_reload*/ false,
+            Some(Arc::clone(&LOCAL_FS)),
+        )
         .await;
     assert_eq!(outcome_without_extra.skills, outcome_with_extra.skills);
     assert_eq!(outcome_without_extra.errors, outcome_with_extra.errors);
 }
 
+#[tokio::test]
+async fn skills_for_cwd_loads_repo_user_and_extra_roots_with_local_fs() {
+    let codex_home = tempfile::tempdir().expect("tempdir");
+    let cwd = tempfile::tempdir().expect("tempdir");
+    let extra_root = tempfile::tempdir().expect("tempdir");
+    let repo_dot_codex = cwd.path().join(".codex");
+    fs::create_dir_all(&repo_dot_codex).expect("create repo config dir");
+
+    write_user_skill(&codex_home, "user", "user-skill", "from local user root");
+    write_user_skill(&extra_root, "extra", "extra-skill", "from extra root");
+    let repo_skill_dir = repo_dot_codex.join("skills/repo");
+    fs::create_dir_all(&repo_skill_dir).expect("create repo skill dir");
+    fs::write(
+        repo_skill_dir.join("SKILL.md"),
+        "---\nname: repo-skill\ndescription: from repo root\n---\n\n# Body\n",
+    )
+    .expect("write repo skill");
+
+    let config_layer_stack = ConfigLayerStack::new(
+        vec![
+            user_config_layer(&codex_home, ""),
+            ConfigLayerEntry::new(
+                ConfigLayerSource::Project {
+                    dot_codex_folder: repo_dot_codex.abs(),
+                },
+                toml::Value::Table(toml::map::Map::new()),
+            ),
+        ],
+        Default::default(),
+        ConfigRequirementsToml::default(),
+    )
+    .expect("valid config layer stack");
+    let skills_input = SkillsLoadInput::new(
+        cwd.path().abs(),
+        Vec::new(),
+        config_layer_stack.clone(),
+        bundled_skills_enabled_from_stack(&config_layer_stack),
+    );
+    let skills_manager = SkillsManager::new(
+        codex_home.path().abs(),
+        /*bundled_skills_enabled*/ true,
+    );
+
+    let outcome = skills_manager
+        .skills_for_cwd_with_extra_user_roots(
+            &skills_input,
+            /*force_reload*/ true,
+            &[extra_root.path().abs()],
+            Some(Arc::clone(&LOCAL_FS)),
+        )
+        .await;
+
+    assert!(
+        outcome.errors.is_empty(),
+        "unexpected errors: {:?}",
+        outcome.errors
+    );
+    let loaded_names = outcome
+        .skills
+        .iter()
+        .map(|skill| skill.name.as_str())
+        .collect::<HashSet<_>>();
+    assert!(loaded_names.contains("user-skill"));
+    assert!(loaded_names.contains("repo-skill"));
+    assert!(loaded_names.contains("extra-skill"));
+}
+
+#[tokio::test]
+async fn skills_for_cwd_without_fs_skips_repo_and_extra_roots() {
+    let codex_home = tempfile::tempdir().expect("tempdir");
+    let cwd = tempfile::tempdir().expect("tempdir");
+    let extra_root = tempfile::tempdir().expect("tempdir");
+    let repo_dot_codex = cwd.path().join(".codex");
+    fs::create_dir_all(&repo_dot_codex).expect("create repo config dir");
+
+    write_user_skill(&codex_home, "user", "user-skill", "from local user root");
+    write_user_skill(&extra_root, "extra", "extra-skill", "from extra root");
+    let repo_skill_dir = repo_dot_codex.join("skills/repo");
+    fs::create_dir_all(&repo_skill_dir).expect("create repo skill dir");
+    fs::write(
+        repo_skill_dir.join("SKILL.md"),
+        "---\nname: repo-skill\ndescription: from repo root\n---\n\n# Body\n",
+    )
+    .expect("write repo skill");
+
+    let config_layer_stack = ConfigLayerStack::new(
+        vec![
+            user_config_layer(&codex_home, ""),
+            ConfigLayerEntry::new(
+                ConfigLayerSource::Project {
+                    dot_codex_folder: repo_dot_codex.abs(),
+                },
+                toml::Value::Table(toml::map::Map::new()),
+            ),
+        ],
+        Default::default(),
+        ConfigRequirementsToml::default(),
+    )
+    .expect("valid config layer stack");
+    let skills_input = SkillsLoadInput::new(
+        cwd.path().abs(),
+        Vec::new(),
+        config_layer_stack.clone(),
+        bundled_skills_enabled_from_stack(&config_layer_stack),
+    );
+    let skills_manager = SkillsManager::new(
+        codex_home.path().abs(),
+        /*bundled_skills_enabled*/ true,
+    );
+
+    let outcome = skills_manager
+        .skills_for_cwd_with_extra_user_roots(
+            &skills_input,
+            /*force_reload*/ true,
+            &[extra_root.path().abs()],
+            /*fs*/ None,
+        )
+        .await;
+
+    assert!(
+        outcome.errors.is_empty(),
+        "unexpected errors: {:?}",
+        outcome.errors
+    );
+    let loaded_names = outcome
+        .skills
+        .iter()
+        .map(|skill| skill.name.as_str())
+        .collect::<HashSet<_>>();
+    assert!(loaded_names.contains("user-skill"));
+    assert!(!loaded_names.contains("repo-skill"));
+    assert!(!loaded_names.contains("extra-skill"));
+}
+
 #[tokio::test]
 async fn skills_for_config_excludes_bundled_skills_when_disabled_in_config() {
     let codex_home = tempfile::tempdir().expect("tempdir");
@@ -330,7 +473,8 @@ async fn skills_for_config_excludes_bundled_skills_when_disabled_in_config() {
     )
     .expect("rewrite bundled skill");
 
-    let outcome = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]);
+    let outcome =
+        skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]).await;
     assert!(
         outcome
             .skills
@@ -356,7 +500,7 @@ async fn skills_for_cwd_with_extra_roots_only_refreshes_on_force_reload() {
         codex_home.path().abs(),
         /*bundled_skills_enabled*/ true,
     );
-    let _ = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]);
+    let _ = skills_for_config_with_stack(&skills_manager, &cwd, &config_layer_stack, &[]).await;
 
     write_user_skill(&extra_root_a, "x", "extra-skill-a", "from extra root a");
     write_user_skill(&extra_root_b, "x", "extra-skill-b", "from extra root b");
@@ -373,6 +517,7 @@ async fn skills_for_cwd_with_extra_roots_only_refreshes_on_force_reload() {
             &base_input,
             /*force_reload*/ true,
             std::slice::from_ref(&extra_root_a_path),
+            Some(Arc::clone(&LOCAL_FS)),
         )
         .await;
     assert!(
@@ -394,6 +539,7 @@ async fn skills_for_cwd_with_extra_roots_only_refreshes_on_force_reload() {
             &base_input,
             /*force_reload*/ false,
             std::slice::from_ref(&extra_root_b_path),
+            Some(Arc::clone(&LOCAL_FS)),
         )
         .await;
     assert!(
@@ -414,6 +560,7 @@ async fn skills_for_cwd_with_extra_roots_only_refreshes_on_force_reload() {
             &base_input,
             /*force_reload*/ true,
             std::slice::from_ref(&extra_root_b_path),
+            Some(Arc::clone(&LOCAL_FS)),
         )
         .await;
     assert!(
@@ -600,7 +747,11 @@ async fn skills_for_config_ignores_cwd_cache_when_session_flags_reenable_skill()
     );
 
     let parent_outcome = skills_manager
-        .skills_for_cwd(&parent_input, /*force_reload*/ true)
+        .skills_for_cwd(
+            &parent_input,
+            /*force_reload*/ true,
+            Some(Arc::clone(&LOCAL_FS)),
+        )
         .await;
     let parent_skill = parent_outcome
         .skills
@@ -609,7 +760,8 @@ async fn skills_for_config_ignores_cwd_cache_when_session_flags_reenable_skill()
         .expect("demo skill should be discovered");
     assert_eq!(parent_outcome.is_skill_enabled(parent_skill), false);
 
-    let child_outcome = skills_for_config_with_stack(&skills_manager, &cwd, &child_stack, &[]);
+    let child_outcome =
+        skills_for_config_with_stack(&skills_manager, &cwd, &child_stack, &[]).await;
     let child_skill = child_outcome
         .skills
         .iter()

codex-rs/core-skills/src/model.rs

@@ -1,7 +1,9 @@
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::fmt;
 use std::sync::Arc;
 
+use codex_exec_server::ExecutorFileSystem;
 use codex_protocol::protocol::Product;
 use codex_protocol::protocol::SkillScope;
 use codex_utils_absolute_path::AbsolutePathBuf;
@@ -87,6 +89,7 @@ pub struct SkillLoadOutcome {
     pub skills: Vec<SkillMetadata>,
     pub errors: Vec<SkillError>,
     pub disabled_paths: HashSet<AbsolutePathBuf>,
+    pub(crate) file_systems_by_skill_path: SkillFileSystemsByPath,
     pub(crate) implicit_skills_by_scripts_dir: Arc<HashMap<AbsolutePathBuf, SkillMetadata>>,
     pub(crate) implicit_skills_by_doc_path: Arc<HashMap<AbsolutePathBuf, SkillMetadata>>,
 }
@@ -113,6 +116,49 @@ impl SkillLoadOutcome {
             .iter()
             .map(|skill| (skill, self.is_skill_enabled(skill)))
     }
+
+    pub(crate) fn file_system_for_skill(
+        &self,
+        skill: &SkillMetadata,
+    ) -> Option<Arc<dyn ExecutorFileSystem>> {
+        self.file_systems_by_skill_path
+            .get(&skill.path_to_skills_md)
+    }
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct SkillFileSystemsByPath {
+    values: Arc<HashMap<AbsolutePathBuf, Arc<dyn ExecutorFileSystem>>>,
+}
+
+impl SkillFileSystemsByPath {
+    pub(crate) fn new(values: HashMap<AbsolutePathBuf, Arc<dyn ExecutorFileSystem>>) -> Self {
+        Self {
+            values: Arc::new(values),
+        }
+    }
+
+    fn get(&self, path: &AbsolutePathBuf) -> Option<Arc<dyn ExecutorFileSystem>> {
+        self.values.get(path).map(Arc::clone)
+    }
+
+    fn retain_paths(&mut self, paths: &HashSet<AbsolutePathBuf>) {
+        self.values = Arc::new(
+            self.values
+                .iter()
+                .filter(|(path, _)| paths.contains(*path))
+                .map(|(path, fs)| (path.clone(), Arc::clone(fs)))
+                .collect(),
+        );
+    }
+}
+
+impl fmt::Debug for SkillFileSystemsByPath {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SkillFileSystemsByPath")
+            .field("len", &self.values.len())
+            .finish()
+    }
 }
 
 pub fn filter_skill_load_outcome_for_product(
@@ -122,6 +168,14 @@ pub fn filter_skill_load_outcome_for_product(
     outcome
         .skills
         .retain(|skill| skill.matches_product_restriction_for_product(restriction_product));
+    let retained_paths: HashSet<AbsolutePathBuf> = outcome
+        .skills
+        .iter()
+        .map(|skill| skill.path_to_skills_md.clone())
+        .collect();
+    outcome
+        .file_systems_by_skill_path
+        .retain_paths(&retained_paths);
     outcome.implicit_skills_by_scripts_dir = Arc::new(
         outcome
             .implicit_skills_by_scripts_dir

codex-rs/core/config.schema.json

@@ -810,7 +810,8 @@
     },
     "MarketplaceSourceType": {
       "enum": [
-        "git"
+        "git",
+        "local"
       ],
       "type": "string"
     },

codex-rs/core/src/agent/role_tests.rs

@@ -658,7 +658,12 @@ enabled = false
     let plugin_outcome = plugins_manager.plugins_for_config(&config).await;
     let effective_skill_roots = plugin_outcome.effective_skill_roots();
     let skills_input = skills_load_input_from_config(&config, effective_skill_roots);
-    let outcome = skills_manager.skills_for_config(&skills_input);
+    let outcome = skills_manager
+        .skills_for_config(
+            &skills_input,
+            Some(Arc::clone(&codex_exec_server::LOCAL_FS)),
+        )
+        .await;
     let skill = outcome
         .skills
         .iter()

codex-rs/core/src/codex.rs

@@ -496,10 +496,17 @@ impl Codex {
         let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
         let (tx_event, rx_event) = async_channel::unbounded();
 
+        let environment = environment_manager
+            .current()
+            .await
+            .map_err(|err| CodexErr::Fatal(format!("failed to create environment: {err}")))?;
+        let fs = environment
+            .as_ref()
+            .map(|environment| environment.get_filesystem());
         let plugin_outcome = plugins_manager.plugins_for_config(&config).await;
         let effective_skill_roots = plugin_outcome.effective_skill_roots();
         let skills_input = skills_load_input_from_config(&config, effective_skill_roots);
-        let loaded_skills = skills_manager.skills_for_config(&skills_input);
+        let loaded_skills = skills_manager.skills_for_config(&skills_input, fs).await;
 
         for err in &loaded_skills.errors {
             error!(
@@ -544,10 +551,6 @@ impl Codex {
             config.startup_warnings.push(message);
         }
 
-        let environment = environment_manager
-            .current()
-            .await
-            .map_err(|err| CodexErr::Fatal(format!("failed to create environment: {err}")))?;
         let user_instructions = get_user_instructions(&config, environment.as_deref()).await;
 
         let exec_policy = if crate::guardian::is_guardian_reviewer_source(&session_source) {
@@ -2670,10 +2673,16 @@ impl Session {
             .await;
         let effective_skill_roots = plugin_outcome.effective_skill_roots();
         let skills_input = skills_load_input_from_config(&per_turn_config, effective_skill_roots);
+        let fs = self
+            .services
+            .environment
+            .as_ref()
+            .map(|environment| environment.get_filesystem());
         let skills_outcome = Arc::new(
             self.services
                 .skills_manager
-                .skills_for_config(&skills_input),
+                .skills_for_config(&skills_input, fs)
+                .await,
         );
         let mut turn_context: TurnContext = Self::make_turn_context(
             self.conversation_id,
@@ -5398,6 +5407,11 @@ mod handlers {
 
         let skills_manager = &sess.services.skills_manager;
         let plugins_manager = &sess.services.plugins_manager;
+        let fs = sess
+            .services
+            .environment
+            .as_ref()
+            .map(|environment| environment.get_filesystem());
         let config = sess.get_config().await;
         let codex_home = sess.codex_home().await;
         let mut skills = Vec::new();
@@ -5454,7 +5468,7 @@ mod handlers {
                 config.bundled_skills_enabled(),
             );
             let outcome = skills_manager
-                .skills_for_cwd(&skills_input, force_reload)
+                .skills_for_cwd(&skills_input, force_reload, fs.clone())
                 .await;
             let errors = super::errors_to_info(&outcome.errors);
             let skills_metadata = super::skills_to_info(&outcome.skills, &outcome.disabled_paths);
@@ -6232,6 +6246,7 @@ pub(crate) async fn run_turn(
         warnings: skill_warnings,
     } = build_skill_injections(
         &mentioned_skills,
+        skills_outcome,
         Some(&session_telemetry),
         &sess.services.analytics_events_client,
         tracking.clone(),

codex-rs/core/src/codex_delegate.rs

@@ -75,7 +75,7 @@ pub(crate) async fn run_codex_thread_interactive(
     let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
     let (tx_ops, rx_ops) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
 
-    let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs {
+    let CodexSpawnOk { codex, .. } = Box::pin(Codex::spawn(CodexSpawnArgs {
         config,
         auth_manager,
         models_manager,
@@ -97,7 +97,7 @@ pub(crate) async fn run_codex_thread_interactive(
         inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)),
         parent_trace: None,
         analytics_events_client: Some(parent_session.services.analytics_events_client.clone()),
-    })
+    }))
     .await?;
     if parent_session.enabled(codex_features::Feature::GeneralAnalytics) {
         let thread_config = codex.thread_config_snapshot().await;
@@ -172,7 +172,7 @@ pub(crate) async fn run_codex_thread_one_shot(
     // Use a child token so we can stop the delegate after completion without
     // requiring the caller to cancel the parent token.
     let child_cancel = cancel_token.child_token();
-    let io = run_codex_thread_interactive(
+    let io = Box::pin(run_codex_thread_interactive(
         config,
         auth_manager,
         models_manager,
@@ -181,7 +181,7 @@ pub(crate) async fn run_codex_thread_one_shot(
         child_cancel.clone(),
         subagent_source,
         initial_history,
-    )
+    ))
     .await?;
 
     // Send the initial input to kick off the one-shot turn.

codex-rs/core/src/codex_tests.rs

@@ -2450,12 +2450,19 @@ async fn new_default_turn_uses_config_aware_skills_for_role_overrides() {
     )
     .expect("write skill");
 
+    let skill_fs = session
+        .services
+        .environment
+        .as_ref()
+        .map(|environment| environment.get_filesystem())
+        .unwrap_or_else(|| std::sync::Arc::clone(&codex_exec_server::LOCAL_FS));
     let parent_outcome = session
         .services
         .skills_manager
         .skills_for_cwd(
             &crate::skills_load_input_from_config(&parent_config, Vec::new()),
             /*force_reload*/ true,
+            Some(Arc::clone(&skill_fs)),
         )
         .await;
     let parent_skill = parent_outcome
@@ -2845,7 +2852,13 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
     let effective_skill_roots = plugin_outcome.effective_skill_roots();
     let skills_input =
         crate::skills_load_input_from_config(&per_turn_config, effective_skill_roots);
-    let skills_outcome = Arc::new(services.skills_manager.skills_for_config(&skills_input));
+    let skill_fs = environment.get_filesystem();
+    let skills_outcome = Arc::new(
+        services
+            .skills_manager
+            .skills_for_config(&skills_input, Some(Arc::clone(&skill_fs)))
+            .await,
+    );
     let turn_context = Session::make_turn_context(
         conversation_id,
         Some(Arc::clone(&auth_manager)),
@@ -3691,7 +3704,13 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
     let effective_skill_roots = plugin_outcome.effective_skill_roots();
     let skills_input =
         crate::skills_load_input_from_config(&per_turn_config, effective_skill_roots);
-    let skills_outcome = Arc::new(services.skills_manager.skills_for_config(&skills_input));
+    let skill_fs = environment.get_filesystem();
+    let skills_outcome = Arc::new(
+        services
+            .skills_manager
+            .skills_for_config(&skills_input, Some(Arc::clone(&skill_fs)))
+            .await,
+    );
     let turn_context = Arc::new(Session::make_turn_context(
         conversation_id,
         Some(Arc::clone(&auth_manager)),

codex-rs/core/src/file_watcher.rs

@@ -252,6 +252,16 @@ pub struct WatchRegistration {
     watched_paths: Vec<WatchPath>,
 }
 
+impl Default for WatchRegistration {
+    fn default() -> Self {
+        Self {
+            file_watcher: std::sync::Weak::new(),
+            subscriber_id: 0,
+            watched_paths: Vec::new(),
+        }
+    }
+}
+
 impl Drop for WatchRegistration {
     fn drop(&mut self) {
         if let Some(file_watcher) = self.file_watcher.upgrade() {

codex-rs/core/src/guardian/review.rs

@@ -163,14 +163,14 @@ async fn run_guardian_review(
 
     let schema = guardian_output_schema();
     let terminal_action = action_summary.clone();
-    let outcome = run_guardian_review_session(
+    let outcome = Box::pin(run_guardian_review_session(
         session.clone(),
         turn.clone(),
         request,
         retry_reason,
         schema,
         external_cancel,
-    )
+    ))
     .await;
 
     let assessment = match outcome {
@@ -303,14 +303,16 @@ pub(crate) async fn review_approval_request(
     request: GuardianApprovalRequest,
     retry_reason: Option<String>,
 ) -> ReviewDecision {
-    run_guardian_review(
+    // Box the delegated review future so callers do not inline the entire
+    // guardian session state machine into their own async stack.
+    Box::pin(run_guardian_review(
         Arc::clone(session),
         Arc::clone(turn),
         review_id,
         request,
         retry_reason,
         /*external_cancel*/ None,
-    )
+    ))
     .await
 }
 
@@ -322,14 +324,14 @@ pub(crate) async fn review_approval_request_with_cancel(
     retry_reason: Option<String>,
     cancel_token: CancellationToken,
 ) -> ReviewDecision {
-    run_guardian_review(
+    Box::pin(run_guardian_review(
         Arc::clone(session),
         Arc::clone(turn),
         review_id,
         request,
         retry_reason,
         Some(cancel_token),
-    )
+    ))
     .await
 }
 
@@ -411,22 +413,24 @@ pub(super) async fn run_guardian_review_session(
         Err(err) => return GuardianReviewOutcome::Completed(Err(err)),
     };
 
-    match session
-        .guardian_review_session
-        .run_review(GuardianReviewSessionParams {
-            parent_session: Arc::clone(&session),
-            parent_turn: turn.clone(),
-            spawn_config: guardian_config,
-            request,
-            retry_reason,
-            schema,
-            model: guardian_model,
-            reasoning_effort: guardian_reasoning_effort,
-            reasoning_summary: turn.reasoning_summary,
-            personality: turn.personality,
-            external_cancel,
-        })
-        .await
+    match Box::pin(
+        session
+            .guardian_review_session
+            .run_review(GuardianReviewSessionParams {
+                parent_session: Arc::clone(&session),
+                parent_turn: turn.clone(),
+                spawn_config: guardian_config,
+                request,
+                retry_reason,
+                schema,
+                model: guardian_model,
+                reasoning_effort: guardian_reasoning_effort,
+                reasoning_summary: turn.reasoning_summary,
+                personality: turn.personality,
+                external_cancel,
+            }),
+    )
+    .await
     {
         GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => {
             GuardianReviewOutcome::Completed(parse_guardian_assessment(

codex-rs/core/src/guardian/review_session.rs

@@ -327,32 +327,30 @@ impl GuardianReviewSessionManager {
         };
 
         if trunk.reuse_key != next_reuse_key {
-            return self
-                .run_ephemeral_review(
-                    params,
-                    next_reuse_key,
-                    deadline,
-                    /*fork_snapshot*/ None,
-                )
-                .await;
+            return Box::pin(self.run_ephemeral_review(
+                params,
+                next_reuse_key,
+                deadline,
+                /*fork_snapshot*/ None,
+            ))
+            .await;
         }
 
         let trunk_guard = match trunk.review_lock.try_lock() {
             Ok(trunk_guard) => trunk_guard,
             Err(_) => {
-                return self
-                    .run_ephemeral_review(
-                        params,
-                        next_reuse_key,
-                        deadline,
-                        trunk.fork_snapshot().await,
-                    )
-                    .await;
+                return Box::pin(self.run_ephemeral_review(
+                    params,
+                    next_reuse_key,
+                    deadline,
+                    trunk.fork_snapshot().await,
+                ))
+                .await;
             }
         };
 
         let (outcome, keep_review_session) =
-            run_review_on_session(trunk.as_ref(), &params, deadline).await;
+            Box::pin(run_review_on_session(trunk.as_ref(), &params, deadline)).await;
         if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) {
             trunk.refresh_last_committed_fork_snapshot().await;
         }
@@ -488,7 +486,12 @@ impl GuardianReviewSessionManager {
         let mut cleanup =
             EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session));
 
-        let (outcome, _) = run_review_on_session(review_session.as_ref(), &params, deadline).await;
+        let (outcome, _) = Box::pin(run_review_on_session(
+            review_session.as_ref(),
+            &params,
+            deadline,
+        ))
+        .await;
         if let Some(review_session) = self.take_active_ephemeral(&review_session).await {
             cleanup.disarm();
             review_session.shutdown_in_background();
@@ -512,7 +515,7 @@ async fn spawn_guardian_review_session(
         ),
         None => (None, 0, None),
     };
-    let codex = run_codex_thread_interactive(
+    let codex = Box::pin(run_codex_thread_interactive(
         spawn_config,
         params.parent_session.services.auth_manager.clone(),
         params.parent_session.services.models_manager.clone(),
@@ -521,7 +524,7 @@ async fn spawn_guardian_review_session(
         cancel_token.clone(),
         SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()),
         initial_history,
-    )
+    ))
     .await?;
 
     Ok(GuardianReviewSession {

codex-rs/core/src/guardian/tests.rs

@@ -1332,222 +1332,242 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() ->
     Ok(())
 }
 
-#[tokio::test(flavor = "current_thread")]
-async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> {
-    let first_assessment = serde_json::json!({
-        "risk_level": "low",
-        "user_authorization": "high",
-        "outcome": "allow",
-        "rationale": "first guardian rationale",
-    })
-    .to_string();
-    let second_assessment = serde_json::json!({
-        "risk_level": "low",
-        "user_authorization": "high",
-        "outcome": "allow",
-        "rationale": "second guardian rationale",
-    })
-    .to_string();
-    let third_assessment = serde_json::json!({
-        "risk_level": "low",
-        "user_authorization": "high",
-        "outcome": "allow",
-        "rationale": "third guardian rationale",
-    })
-    .to_string();
-    let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
-    let (server, _) = start_streaming_sse_server(vec![
-        vec![StreamingSseChunk {
-            gate: None,
-            body: sse(vec![
-                ev_response_created("resp-guardian-1"),
-                ev_assistant_message("msg-guardian-1", &first_assessment),
-                ev_completed("resp-guardian-1"),
-            ]),
-        }],
-        vec![
-            StreamingSseChunk {
+#[test]
+fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> {
+    const TEST_STACK_SIZE_BYTES: usize = 2 * 1024 * 1024;
+
+    let handle =
+        std::thread::Builder::new()
+            .name("guardian_parallel_reviews_fork_from_last_committed_trunk_history".to_string())
+            .stack_size(TEST_STACK_SIZE_BYTES)
+            .spawn(|| -> anyhow::Result<()> {
+                let runtime = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()?;
+                runtime.block_on(Box::pin(async {
+        let first_assessment = serde_json::json!({
+            "risk_level": "low",
+            "user_authorization": "high",
+            "outcome": "allow",
+            "rationale": "first guardian rationale",
+        })
+        .to_string();
+        let second_assessment = serde_json::json!({
+            "risk_level": "low",
+            "user_authorization": "high",
+            "outcome": "allow",
+            "rationale": "second guardian rationale",
+        })
+        .to_string();
+        let third_assessment = serde_json::json!({
+            "risk_level": "low",
+            "user_authorization": "high",
+            "outcome": "allow",
+            "rationale": "third guardian rationale",
+        })
+        .to_string();
+        let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
+        let (server, _) = start_streaming_sse_server(vec![
+            vec![StreamingSseChunk {
                 gate: None,
-                body: sse(vec![ev_response_created("resp-guardian-2")]),
-            },
-            StreamingSseChunk {
-                gate: Some(gate_rx),
                 body: sse(vec![
-                    ev_assistant_message("msg-guardian-2", &second_assessment),
-                    ev_completed("resp-guardian-2"),
+                    ev_response_created("resp-guardian-1"),
+                    ev_assistant_message("msg-guardian-1", &first_assessment),
+                    ev_completed("resp-guardian-1"),
                 ]),
-            },
-        ],
-        vec![StreamingSseChunk {
-            gate: None,
-            body: sse(vec![
-                ev_response_created("resp-guardian-3"),
-                ev_assistant_message("msg-guardian-3", &third_assessment),
-                ev_completed("resp-guardian-3"),
-            ]),
-        }],
-    ])
-    .await;
+            }],
+            vec![
+                StreamingSseChunk {
+                    gate: None,
+                    body: sse(vec![ev_response_created("resp-guardian-2")]),
+                },
+                StreamingSseChunk {
+                    gate: Some(gate_rx),
+                    body: sse(vec![
+                        ev_assistant_message("msg-guardian-2", &second_assessment),
+                        ev_completed("resp-guardian-2"),
+                    ]),
+                },
+            ],
+            vec![StreamingSseChunk {
+                gate: None,
+                body: sse(vec![
+                    ev_response_created("resp-guardian-3"),
+                    ev_assistant_message("msg-guardian-3", &third_assessment),
+                    ev_completed("resp-guardian-3"),
+                ]),
+            }],
+        ])
+        .await;
 
-    let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
-    seed_guardian_parent_history(&session, &turn).await;
+        let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
+        seed_guardian_parent_history(&session, &turn).await;
 
-    let initial_request = GuardianApprovalRequest::Shell {
-        id: "shell-guardian-1".to_string(),
-        command: vec!["git".to_string(), "status".to_string()],
-        cwd: test_path_buf("/repo/codex-rs/core").abs(),
-        sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
-        additional_permissions: None,
-        justification: Some("Inspect repo state before proceeding.".to_string()),
-    };
-    assert_eq!(
-        review_approval_request(
+        let initial_request = GuardianApprovalRequest::Shell {
+            id: "shell-guardian-1".to_string(),
+            command: vec!["git".to_string(), "status".to_string()],
+            cwd: test_path_buf("/repo/codex-rs/core").abs(),
+            sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
+            additional_permissions: None,
+            justification: Some("Inspect repo state before proceeding.".to_string()),
+        };
+        assert_eq!(
+            review_approval_request(
+                &session,
+                &turn,
+                "review-shell-guardian-1".to_string(),
+                initial_request,
+                /*retry_reason*/ None
+            )
+            .await,
+            ReviewDecision::Approved
+        );
+        session
+            .record_into_history(
+                &[
+                    ResponseItem::Message {
+                        id: None,
+                        role: "user".to_string(),
+                        content: vec![ContentItem::InputText {
+                            text: "Please inspect pending changes before pushing.".to_string(),
+                        }],
+                        end_turn: None,
+                        phase: None,
+                    },
+                    ResponseItem::Message {
+                        id: None,
+                        role: "assistant".to_string(),
+                        content: vec![ContentItem::OutputText {
+                            text: "I need approval to run git diff.".to_string(),
+                        }],
+                        end_turn: None,
+                        phase: None,
+                    },
+                ],
+                turn.as_ref(),
+            )
+            .await;
+
+        let second_request = GuardianApprovalRequest::Shell {
+            id: "shell-guardian-2".to_string(),
+            command: vec!["git".to_string(), "diff".to_string()],
+            cwd: test_path_buf("/repo/codex-rs/core").abs(),
+            sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
+            additional_permissions: None,
+            justification: Some("Inspect pending changes before proceeding.".to_string()),
+        };
+        let third_request = GuardianApprovalRequest::Shell {
+            id: "shell-guardian-3".to_string(),
+            command: vec!["git".to_string(), "push".to_string()],
+            cwd: test_path_buf("/repo/codex-rs/core").abs(),
+            sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
+            additional_permissions: None,
+            justification: Some("Inspect whether pushing is safe before proceeding.".to_string()),
+        };
+
+        let session_for_second = Arc::clone(&session);
+        let turn_for_second = Arc::clone(&turn);
+        let mut second_review = tokio::spawn(async move {
+            review_approval_request(
+                &session_for_second,
+                &turn_for_second,
+                "review-shell-guardian-2".to_string(),
+                second_request,
+                Some("trunk follow-up".to_string()),
+            )
+            .await
+        });
+
+        let second_request_observed = tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if server.requests().await.len() >= 2 {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await;
+        assert!(
+            second_request_observed.is_ok(),
+            "second guardian request was not observed"
+        );
+        session
+            .record_into_history(
+                &[
+                    ResponseItem::Message {
+                        id: None,
+                        role: "user".to_string(),
+                        content: vec![ContentItem::InputText {
+                            text: "Now inspect whether pushing is safe.".to_string(),
+                        }],
+                        end_turn: None,
+                        phase: None,
+                    },
+                    ResponseItem::Message {
+                        id: None,
+                        role: "assistant".to_string(),
+                        content: vec![ContentItem::OutputText {
+                            text: "I need approval to push after the diff check.".to_string(),
+                        }],
+                        end_turn: None,
+                        phase: None,
+                    },
+                ],
+                turn.as_ref(),
+            )
+            .await;
+
+        let third_decision = review_approval_request(
             &session,
             &turn,
-            "review-shell-guardian-1".to_string(),
-            initial_request,
-            /*retry_reason*/ None
-        )
-        .await,
-        ReviewDecision::Approved
-    );
-    session
-        .record_into_history(
-            &[
-                ResponseItem::Message {
-                    id: None,
-                    role: "user".to_string(),
-                    content: vec![ContentItem::InputText {
-                        text: "Please inspect pending changes before pushing.".to_string(),
-                    }],
-                    end_turn: None,
-                    phase: None,
-                },
-                ResponseItem::Message {
-                    id: None,
-                    role: "assistant".to_string(),
-                    content: vec![ContentItem::OutputText {
-                        text: "I need approval to run git diff.".to_string(),
-                    }],
-                    end_turn: None,
-                    phase: None,
-                },
-            ],
-            turn.as_ref(),
+            "review-shell-guardian-3".to_string(),
+            third_request,
+            Some("parallel follow-up".to_string()),
         )
         .await;
+        assert_eq!(third_decision, ReviewDecision::Approved);
+        let requests = server.requests().await;
+        assert_eq!(requests.len(), 3);
+        let third_request_body = serde_json::from_slice::<serde_json::Value>(&requests[2])?;
+        let third_request_body_text = third_request_body.to_string();
+        assert!(
+            third_request_body_text.contains("first guardian rationale"),
+            "forked guardian review should include the last committed trunk assessment"
+        );
+        let third_user_message = last_user_message_text_from_body(&third_request_body);
+        assert!(third_user_message.contains(">>> TRANSCRIPT DELTA START\n"));
+        assert!(
+            third_user_message.contains("[5] user: Please inspect pending changes before pushing.")
+        );
+        assert!(third_user_message.contains("[7] user: Now inspect whether pushing is safe."));
+        assert!(!third_user_message.contains("[1] user: Please check the repo visibility"));
+        assert!(
+            !third_request_body_text.contains("second guardian rationale"),
+            "forked guardian review should not include the still in-flight trunk assessment"
+        );
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), &mut second_review)
+                .await
+                .is_err(),
+            "the trunk guardian review should still be blocked on its gated response"
+        );
 
-    let second_request = GuardianApprovalRequest::Shell {
-        id: "shell-guardian-2".to_string(),
-        command: vec!["git".to_string(), "diff".to_string()],
-        cwd: test_path_buf("/repo/codex-rs/core").abs(),
-        sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
-        additional_permissions: None,
-        justification: Some("Inspect pending changes before proceeding.".to_string()),
-    };
-    let third_request = GuardianApprovalRequest::Shell {
-        id: "shell-guardian-3".to_string(),
-        command: vec!["git".to_string(), "push".to_string()],
-        cwd: test_path_buf("/repo/codex-rs/core").abs(),
-        sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
-        additional_permissions: None,
-        justification: Some("Inspect whether pushing is safe before proceeding.".to_string()),
-    };
+        gate_tx
+            .send(())
+            .expect("second guardian review gate should still be open");
+        assert_eq!(second_review.await?, ReviewDecision::Approved);
+        server.shutdown().await;
 
-    let session_for_second = Arc::clone(&session);
-    let turn_for_second = Arc::clone(&turn);
-    let mut second_review = tokio::spawn(async move {
-        review_approval_request(
-            &session_for_second,
-            &turn_for_second,
-            "review-shell-guardian-2".to_string(),
-            second_request,
-            Some("trunk follow-up".to_string()),
-        )
-        .await
-    });
+        Ok(())
+                }))
+            })?;
 
-    let second_request_observed = tokio::time::timeout(Duration::from_secs(5), async {
-        loop {
-            if server.requests().await.len() >= 2 {
-                break;
-            }
-            tokio::task::yield_now().await;
-        }
-    })
-    .await;
-    assert!(
-        second_request_observed.is_ok(),
-        "second guardian request was not observed"
-    );
-    session
-        .record_into_history(
-            &[
-                ResponseItem::Message {
-                    id: None,
-                    role: "user".to_string(),
-                    content: vec![ContentItem::InputText {
-                        text: "Now inspect whether pushing is safe.".to_string(),
-                    }],
-                    end_turn: None,
-                    phase: None,
-                },
-                ResponseItem::Message {
-                    id: None,
-                    role: "assistant".to_string(),
-                    content: vec![ContentItem::OutputText {
-                        text: "I need approval to push after the diff check.".to_string(),
-                    }],
-                    end_turn: None,
-                    phase: None,
-                },
-            ],
-            turn.as_ref(),
-        )
-        .await;
-
-    let third_decision = review_approval_request(
-        &session,
-        &turn,
-        "review-shell-guardian-3".to_string(),
-        third_request,
-        Some("parallel follow-up".to_string()),
-    )
-    .await;
-    assert_eq!(third_decision, ReviewDecision::Approved);
-    let requests = server.requests().await;
-    assert_eq!(requests.len(), 3);
-    let third_request_body = serde_json::from_slice::<serde_json::Value>(&requests[2])?;
-    let third_request_body_text = third_request_body.to_string();
-    assert!(
-        third_request_body_text.contains("first guardian rationale"),
-        "forked guardian review should include the last committed trunk assessment"
-    );
-    let third_user_message = last_user_message_text_from_body(&third_request_body);
-    assert!(third_user_message.contains(">>> TRANSCRIPT DELTA START\n"));
-    assert!(
-        third_user_message.contains("[5] user: Please inspect pending changes before pushing.")
-    );
-    assert!(third_user_message.contains("[7] user: Now inspect whether pushing is safe."));
-    assert!(!third_user_message.contains("[1] user: Please check the repo visibility"));
-    assert!(
-        !third_request_body_text.contains("second guardian rationale"),
-        "forked guardian review should not include the still in-flight trunk assessment"
-    );
-    assert!(
-        tokio::time::timeout(Duration::from_millis(100), &mut second_review)
-            .await
-            .is_err(),
-        "the trunk guardian review should still be blocked on its gated response"
-    );
-
-    gate_tx
-        .send(())
-        .expect("second guardian review gate should still be open");
-    assert_eq!(second_review.await?, ReviewDecision::Approved);
-    server.shutdown().await;
-
-    Ok(())
+    match handle.join() {
+        Ok(result) => result,
+        Err(_) => Err(anyhow::anyhow!(
+            "guardian_parallel_reviews_fork_from_last_committed_trunk_history thread panicked"
+        )),
+    }
 }
 #[test]
 fn guardian_review_session_config_preserves_parent_network_proxy() {

codex-rs/core/src/plugins/installed_marketplaces.rs

@@ -45,7 +45,11 @@ pub(crate) fn installed_marketplace_roots_from_config(
                 );
                 return None;
             }
-            let path = default_install_root.join(marketplace_name);
+            let path = resolve_configured_marketplace_root(
+                marketplace_name,
+                marketplace,
+                &default_install_root,
+            )?;
             path.join(".agents/plugins/marketplace.json")
                 .is_file()
                 .then_some(path)
@@ -55,3 +59,18 @@ pub(crate) fn installed_marketplace_roots_from_config(
     roots.sort_unstable_by(|left, right| left.as_path().cmp(right.as_path()));
     roots
 }
+
+pub(crate) fn resolve_configured_marketplace_root(
+    marketplace_name: &str,
+    marketplace: &toml::Value,
+    default_install_root: &Path,
+) -> Option<PathBuf> {
+    match marketplace.get("source_type").and_then(toml::Value::as_str) {
+        Some("local") => marketplace
+            .get("source")
+            .and_then(toml::Value::as_str)
+            .filter(|source| !source.is_empty())
+            .map(PathBuf::from),
+        _ => Some(default_install_root.join(marketplace_name)),
+    }
+}

codex-rs/core/src/plugins/manager.rs

@@ -46,6 +46,7 @@ use codex_app_server_protocol::ConfigValueWriteParams;
 use codex_app_server_protocol::MergeStrategy;
 use codex_config::types::McpServerConfig;
 use codex_config::types::PluginConfig;
+use codex_exec_server::LOCAL_FS;
 use codex_features::Feature;
 use codex_login::AuthManager;
 use codex_login::CodexAuth;
@@ -1031,7 +1032,8 @@ impl PluginsManager {
             manifest_paths,
             self.restriction_product,
             &skill_config_rules,
-        );
+        )
+        .await;
         let apps = load_apps_from_paths(
             source_path.as_path(),
             plugin_app_config_paths(source_path.as_path(), manifest_paths),
@@ -1751,7 +1753,8 @@ async fn load_plugin(
         manifest_paths,
         restriction_product,
         skill_config_rules,
-    );
+    )
+    .await;
     let has_enabled_skills = resolved_skills.has_enabled_skills();
     loaded_plugin.disabled_skill_paths = resolved_skills.disabled_skill_paths;
     loaded_plugin.has_enabled_skills = has_enabled_skills;
@@ -1795,20 +1798,21 @@ impl ResolvedPluginSkills {
     }
 }
 
-fn load_plugin_skills(
+async fn load_plugin_skills(
     plugin_root: &AbsolutePathBuf,
     manifest_paths: &PluginManifestPaths,
     restriction_product: Option<Product>,
     skill_config_rules: &SkillConfigRules,
 ) -> ResolvedPluginSkills {
-    let outcome = load_skills_from_roots(
-        plugin_skill_roots(plugin_root, manifest_paths)
-            .into_iter()
-            .map(|path| SkillRoot {
-                path,
-                scope: SkillScope::User,
-            }),
-    );
+    let roots = plugin_skill_roots(plugin_root, manifest_paths)
+        .into_iter()
+        .map(|path| SkillRoot {
+            path,
+            scope: SkillScope::User,
+            file_system: Arc::clone(&LOCAL_FS),
+        })
+        .collect::<Vec<_>>();
+    let outcome = load_skills_from_roots(roots).await;
     let had_errors = !outcome.errors.is_empty();
     let skills = outcome
         .skills

codex-rs/core/src/plugins/marketplace_add.rs

@@ -16,10 +16,12 @@ use install::marketplace_staging_root;
 use install::replace_marketplace_root;
 use install::safe_marketplace_dir_name;
 use metadata::MarketplaceInstallMetadata;
+use metadata::find_marketplace_root_by_name;
 use metadata::installed_marketplace_root_for_source;
 use metadata::record_added_marketplace_entry;
 use source::MarketplaceSource;
 use source::parse_marketplace_source;
+use source::stage_marketplace_source;
 use source::validate_marketplace_source_root;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -102,6 +104,33 @@ where
         });
     }
 
+    if let MarketplaceSource::Local { path } = &source {
+        let marketplace_name = validate_marketplace_source_root(path)?;
+        if marketplace_name == OPENAI_CURATED_MARKETPLACE_NAME {
+            return Err(MarketplaceAddError::InvalidRequest(format!(
+                "marketplace '{OPENAI_CURATED_MARKETPLACE_NAME}' is reserved and cannot be added from {}",
+                source.display()
+            )));
+        }
+        if find_marketplace_root_by_name(codex_home, &install_root, &marketplace_name)?.is_some() {
+            return Err(MarketplaceAddError::InvalidRequest(format!(
+                "marketplace '{marketplace_name}' is already added from a different source; remove it before adding {}",
+                source.display()
+            )));
+        }
+        record_added_marketplace_entry(codex_home, &marketplace_name, &install_metadata)?;
+        return Ok(MarketplaceAddOutcome {
+            marketplace_name,
+            source_display: source.display(),
+            installed_root: AbsolutePathBuf::try_from(path.clone()).map_err(|err| {
+                MarketplaceAddError::Internal(format!(
+                    "failed to resolve installed marketplace root: {err}"
+                ))
+            })?,
+            already_added: false,
+        });
+    }
+
     let staging_root = marketplace_staging_root(&install_root);
     fs::create_dir_all(&staging_root).map_err(|err| {
         MarketplaceAddError::Internal(format!(
@@ -120,8 +149,7 @@ where
         })?;
     let staged_root = staged_root.keep();
 
-    let MarketplaceSource::Git { url, ref_name } = &source;
-    clone_source(url, ref_name.as_deref(), &sparse_paths, &staged_root)?;
+    stage_marketplace_source(&source, &sparse_paths, &staged_root, clone_source)?;
 
     let marketplace_name = validate_marketplace_source_root(&staged_root)?;
     if marketplace_name == OPENAI_CURATED_MARKETPLACE_NAME {
@@ -214,6 +242,75 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn add_marketplace_sync_installs_local_directory_source_and_updates_config() -> Result<()> {
+        let codex_home = TempDir::new()?;
+        let source_root = TempDir::new()?;
+        write_marketplace_source(source_root.path(), "local copy")?;
+
+        let result = add_marketplace_sync_with_cloner(
+            codex_home.path(),
+            MarketplaceAddRequest {
+                source: source_root.path().display().to_string(),
+                ref_name: None,
+                sparse_paths: Vec::new(),
+            },
+            |_url, _ref_name, _sparse_paths, _destination| {
+                panic!("git cloner should not be called for local marketplace sources")
+            },
+        )?;
+
+        let expected_source = source_root.path().canonicalize()?.display().to_string();
+        assert_eq!(result.marketplace_name, "debug");
+        assert_eq!(result.source_display, expected_source);
+        assert_eq!(
+            result.installed_root.as_path(),
+            source_root.path().canonicalize()?
+        );
+        assert!(!result.already_added);
+        assert!(
+            !marketplace_install_root(codex_home.path())
+                .join("debug")
+                .exists()
+        );
+
+        let config = fs::read_to_string(codex_home.path().join(codex_config::CONFIG_TOML_FILE))?;
+        assert!(config.contains("[marketplaces.debug]"));
+        assert!(config.contains("source_type = \"local\""));
+        assert!(config.contains(&format!("source = \"{expected_source}\"")));
+        Ok(())
+    }
+
+    #[test]
+    fn add_marketplace_sync_treats_existing_local_directory_source_as_already_added() -> Result<()>
+    {
+        let codex_home = TempDir::new()?;
+        let source_root = TempDir::new()?;
+        write_marketplace_source(source_root.path(), "local copy")?;
+
+        let request = MarketplaceAddRequest {
+            source: source_root.path().display().to_string(),
+            ref_name: None,
+            sparse_paths: Vec::new(),
+        };
+        let first_result = add_marketplace_sync_with_cloner(codex_home.path(), request.clone(), {
+            |_url, _ref_name, _sparse_paths, _destination| {
+                panic!("git cloner should not be called for local marketplace sources")
+            }
+        })?;
+        let second_result = add_marketplace_sync_with_cloner(codex_home.path(), request, {
+            |_url, _ref_name, _sparse_paths, _destination| {
+                panic!("git cloner should not be called for local marketplace sources")
+            }
+        })?;
+
+        assert!(!first_result.already_added);
+        assert!(second_result.already_added);
+        assert_eq!(second_result.installed_root, first_result.installed_root);
+
+        Ok(())
+    }
+
     fn write_marketplace_source(source: &Path, marker: &str) -> std::io::Result<()> {
         fs::create_dir_all(source.join(".agents/plugins"))?;
         fs::create_dir_all(source.join("plugins/sample/.codex-plugin"))?;

codex-rs/core/src/plugins/marketplace_add/metadata.rs

@@ -1,5 +1,6 @@
 use super::MarketplaceAddError;
 use super::MarketplaceSource;
+use crate::plugins::installed_marketplaces::resolve_configured_marketplace_root;
 use crate::plugins::validate_marketplace_root;
 use codex_config::CONFIG_TOML_FILE;
 use codex_config::MarketplaceConfigUpdate;
@@ -23,6 +24,9 @@ enum InstalledMarketplaceSource {
         ref_name: Option<String>,
         sparse_paths: Vec<String>,
     },
+    Local {
+        path: String,
+    },
 }
 
 pub(super) fn record_added_marketplace_entry(
@@ -77,7 +81,11 @@ pub(super) fn installed_marketplace_root_for_source(
         if !install_metadata.matches_config(marketplace) {
             continue;
         }
-        let root = install_root.join(marketplace_name);
+        let Some(root) =
+            resolve_configured_marketplace_root(marketplace_name, marketplace, install_root)
+        else {
+            continue;
+        };
         if validate_marketplace_root(&root).is_ok() {
             return Ok(Some(root));
         }
@@ -86,6 +94,48 @@ pub(super) fn installed_marketplace_root_for_source(
     Ok(None)
 }
 
+pub(super) fn find_marketplace_root_by_name(
+    codex_home: &Path,
+    install_root: &Path,
+    marketplace_name: &str,
+) -> Result<Option<PathBuf>, MarketplaceAddError> {
+    let config_path = codex_home.join(CONFIG_TOML_FILE);
+    let config = match fs::read_to_string(&config_path) {
+        Ok(config) => config,
+        Err(err) if err.kind() == ErrorKind::NotFound => return Ok(None),
+        Err(err) => {
+            return Err(MarketplaceAddError::Internal(format!(
+                "failed to read user config {}: {err}",
+                config_path.display()
+            )));
+        }
+    };
+    let config: toml::Value = toml::from_str(&config).map_err(|err| {
+        MarketplaceAddError::Internal(format!(
+            "failed to parse user config {}: {err}",
+            config_path.display()
+        ))
+    })?;
+    let Some(marketplace) = config
+        .get("marketplaces")
+        .and_then(toml::Value::as_table)
+        .and_then(|marketplaces| marketplaces.get(marketplace_name))
+    else {
+        return Ok(None);
+    };
+
+    let Some(root) =
+        resolve_configured_marketplace_root(marketplace_name, marketplace, install_root)
+    else {
+        return Ok(None);
+    };
+    if validate_marketplace_root(&root).is_ok() {
+        Ok(Some(root))
+    } else {
+        Ok(None)
+    }
+}
+
 impl MarketplaceInstallMetadata {
     pub(super) fn from_source(source: &MarketplaceSource, sparse_paths: &[String]) -> Self {
         let source = match source {
@@ -94,6 +144,9 @@ impl MarketplaceInstallMetadata {
                 ref_name: ref_name.clone(),
                 sparse_paths: sparse_paths.to_vec(),
             },
+            MarketplaceSource::Local { path } => InstalledMarketplaceSource::Local {
+                path: path.display().to_string(),
+            },
         };
         Self { source }
     }
@@ -101,24 +154,28 @@ impl MarketplaceInstallMetadata {
     fn config_source_type(&self) -> &'static str {
         match &self.source {
             InstalledMarketplaceSource::Git { .. } => "git",
+            InstalledMarketplaceSource::Local { .. } => "local",
         }
     }
 
     fn config_source(&self) -> String {
         match &self.source {
             InstalledMarketplaceSource::Git { url, .. } => url.clone(),
+            InstalledMarketplaceSource::Local { path } => path.clone(),
         }
     }
 
     fn ref_name(&self) -> Option<&str> {
         match &self.source {
             InstalledMarketplaceSource::Git { ref_name, .. } => ref_name.as_deref(),
+            InstalledMarketplaceSource::Local { .. } => None,
         }
     }
 
     fn sparse_paths(&self) -> &[String] {
         match &self.source {
             InstalledMarketplaceSource::Git { sparse_paths, .. } => sparse_paths,
+            InstalledMarketplaceSource::Local { .. } => &[],
         }
     }
 
@@ -227,4 +284,39 @@ mod tests {
             "unexpected error: {err}"
         );
     }
+
+    #[test]
+    fn installed_marketplace_root_for_source_uses_local_source_root() {
+        let codex_home = TempDir::new().unwrap();
+        let install_root = codex_home.path().join("marketplaces");
+        let source_root = codex_home.path().join("source");
+        fs::create_dir_all(source_root.join(".agents/plugins")).unwrap();
+        fs::write(
+            source_root.join(".agents/plugins/marketplace.json"),
+            r#"{"name":"debug","plugins":[]}"#,
+        )
+        .unwrap();
+        fs::write(
+            codex_home.path().join(CONFIG_TOML_FILE),
+            format!(
+                "[marketplaces.debug]\nsource_type = \"local\"\nsource = \"{}\"\n",
+                source_root.display()
+            ),
+        )
+        .unwrap();
+
+        let source = MarketplaceSource::Local {
+            path: source_root.clone(),
+        };
+        let install_metadata = MarketplaceInstallMetadata::from_source(&source, &[]);
+
+        let root = installed_marketplace_root_for_source(
+            codex_home.path(),
+            &install_root,
+            &install_metadata,
+        )
+        .unwrap();
+
+        assert_eq!(root, Some(source_root));
+    }
 }

codex-rs/core/src/plugins/marketplace_add/source.rs

@@ -2,6 +2,7 @@ use super::MarketplaceAddError;
 use crate::plugins::validate_marketplace_root;
 use crate::plugins::validate_plugin_segment;
 use std::path::Path;
+use std::path::PathBuf;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) enum MarketplaceSource {
@@ -9,6 +10,9 @@ pub(super) enum MarketplaceSource {
         url: String,
         ref_name: Option<String>,
     },
+    Local {
+        path: PathBuf,
+    },
 }
 
 pub(super) fn parse_marketplace_source(
@@ -26,9 +30,18 @@ pub(super) fn parse_marketplace_source(
     let ref_name = explicit_ref.or(parsed_ref);
 
     if looks_like_local_path(&base_source) {
-        return Err(MarketplaceAddError::InvalidRequest(
-            "local marketplace sources are not supported yet; use an HTTP(S) Git URL, SSH Git URL, or GitHub owner/repo".to_string(),
-        ));
+        if ref_name.is_some() {
+            return Err(MarketplaceAddError::InvalidRequest(
+                "--ref is only supported for git marketplace sources".to_string(),
+            ));
+        }
+        let path = resolve_local_source_path(&base_source)?;
+        if path.is_file() {
+            return Err(MarketplaceAddError::InvalidRequest(
+                "local marketplace source must be a directory, not a file".to_string(),
+            ));
+        }
+        return Ok(MarketplaceSource::Local { path });
     }
 
     if is_ssh_git_url(&base_source) || is_git_url(&base_source) {
@@ -50,6 +63,31 @@ pub(super) fn parse_marketplace_source(
     )))
 }
 
+pub(super) fn stage_marketplace_source<F>(
+    source: &MarketplaceSource,
+    sparse_paths: &[String],
+    staged_root: &Path,
+    clone_source: F,
+) -> Result<(), MarketplaceAddError>
+where
+    F: Fn(&str, Option<&str>, &[String], &Path) -> Result<(), MarketplaceAddError>,
+{
+    if !sparse_paths.is_empty() && !matches!(source, MarketplaceSource::Git { .. }) {
+        return Err(MarketplaceAddError::InvalidRequest(
+            "--sparse is only supported for git marketplace sources".to_string(),
+        ));
+    }
+
+    match source {
+        MarketplaceSource::Git { url, ref_name } => {
+            clone_source(url, ref_name.as_deref(), sparse_paths, staged_root)
+        }
+        MarketplaceSource::Local { .. } => unreachable!(
+            "local marketplace sources are added without staging a copied install root"
+        ),
+    }
+}
+
 pub(super) fn validate_marketplace_source_root(root: &Path) -> Result<String, MarketplaceAddError> {
     let marketplace_name = validate_marketplace_root(root)
         .map_err(|err| MarketplaceAddError::InvalidRequest(err.to_string()))?;
@@ -94,6 +132,38 @@ fn looks_like_local_path(source: &str) -> bool {
         || source == ".."
 }
 
+fn resolve_local_source_path(source: &str) -> Result<PathBuf, MarketplaceAddError> {
+    let path = expand_tilde_path(source);
+    let path = if path.is_absolute() {
+        path
+    } else {
+        std::env::current_dir()
+            .map_err(|err| {
+                MarketplaceAddError::Internal(format!(
+                    "failed to read current working directory for local marketplace source: {err}"
+                ))
+            })?
+            .join(path)
+    };
+
+    path.canonicalize().map_err(|err| {
+        MarketplaceAddError::InvalidRequest(format!(
+            "failed to resolve local marketplace source {}: {err}",
+            path.display()
+        ))
+    })
+}
+
+fn expand_tilde_path(source: &str) -> PathBuf {
+    let Some(rest) = source.strip_prefix("~/") else {
+        return PathBuf::from(source);
+    };
+    let Some(home) = std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE")) else {
+        return PathBuf::from(source);
+    };
+    PathBuf::from(home).join(rest)
+}
+
 fn is_ssh_git_url(source: &str) -> bool {
     source.starts_with("ssh://") || source.starts_with("git@") && source.contains(':')
 }
@@ -126,6 +196,7 @@ impl MarketplaceSource {
                 Some(ref_name) => format!("{url}#{ref_name}"),
                 None => url.clone(),
             },
+            Self::Local { path } => path.display().to_string(),
         }
     }
 }
@@ -134,6 +205,7 @@ impl MarketplaceSource {
 mod tests {
     use super::*;
     use pretty_assertions::assert_eq;
+    use tempfile::TempDir;
 
     #[test]
     fn github_shorthand_parses_ref_suffix() {
@@ -229,12 +301,57 @@ mod tests {
     }
 
     #[test]
-    fn parse_marketplace_source_rejects_local_directory_source() {
-        let err = parse_marketplace_source("./marketplace", /*explicit_ref*/ None).unwrap_err();
+    fn local_path_source_parses() {
+        let source = parse_marketplace_source(".", /*explicit_ref*/ None).unwrap();
 
-        assert_eq!(
-            err.to_string(),
-            "local marketplace sources are not supported yet; use an HTTP(S) Git URL, SSH Git URL, or GitHub owner/repo"
+        let MarketplaceSource::Local { path } = source else {
+            panic!("expected local path source");
+        };
+        assert!(path.is_absolute());
+    }
+
+    #[test]
+    fn local_file_source_is_rejected() {
+        let tempdir = TempDir::new().unwrap();
+        let file = tempdir.path().join("marketplace.json");
+        std::fs::write(&file, "{}").unwrap();
+
+        let err =
+            parse_marketplace_source(file.to_str().unwrap(), /*explicit_ref*/ None).unwrap_err();
+
+        assert!(
+            err.to_string()
+                .contains("local marketplace source must be a directory, not a file"),
+            "unexpected error: {err}"
+        );
+    }
+
+    #[test]
+    fn non_git_sources_reject_ref_override() {
+        let err = parse_marketplace_source("./marketplace", Some("main".to_string())).unwrap_err();
+
+        assert!(
+            err.to_string()
+                .contains("--ref is only supported for git marketplace sources"),
+            "unexpected error: {err}"
+        );
+    }
+
+    #[test]
+    fn non_git_sources_reject_sparse_checkout() {
+        let path = std::env::current_dir().unwrap();
+        let err = stage_marketplace_source(
+            &MarketplaceSource::Local { path },
+            &["plugins/foo".to_string()],
+            Path::new("/tmp"),
+            |_url, _ref_name, _sparse_paths, _staged_root| Ok(()),
+        )
+        .unwrap_err();
+
+        assert!(
+            err.to_string()
+                .contains("--sparse is only supported for git marketplace sources"),
+            "unexpected error: {err}"
         );
     }
 

codex-rs/core/src/skills_watcher.rs

@@ -59,12 +59,14 @@ impl SkillsWatcher {
         config: &Config,
         skills_manager: &SkillsManager,
         plugins_manager: &PluginsManager,
+        fs: Option<Arc<dyn codex_exec_server::ExecutorFileSystem>>,
     ) -> WatchRegistration {
         let plugin_outcome = plugins_manager.plugins_for_config(config).await;
         let effective_skill_roots = plugin_outcome.effective_skill_roots();
         let skills_input = skills_load_input_from_config(config, effective_skill_roots);
         let roots = skills_manager
-            .skill_roots_for_config(&skills_input)
+            .skill_roots_for_config(&skills_input, fs)
+            .await
             .into_iter()
             .map(|root| WatchPath {
                 path: root.path.into_path_buf(),

codex-rs/core/src/thread_manager.rs

@@ -365,6 +365,10 @@ impl ThreadManager {
         self.state.mcp_manager.clone()
     }
 
+    pub fn environment_manager(&self) -> Arc<EnvironmentManager> {
+        self.state.environment_manager.clone()
+    }
+
     pub fn get_models_manager(&self) -> Arc<ModelsManager> {
         self.state.models_manager.clone()
     }
@@ -903,14 +907,24 @@ impl ThreadManagerState {
         parent_trace: Option<W3cTraceContext>,
         user_shell_override: Option<crate::shell::Shell>,
     ) -> CodexResult<NewThread> {
-        let watch_registration = self
-            .skills_watcher
-            .register_config(
-                &config,
-                self.skills_manager.as_ref(),
-                self.plugins_manager.as_ref(),
-            )
-            .await;
+        let environment = self
+            .environment_manager
+            .current()
+            .await
+            .map_err(|err| CodexErr::Fatal(format!("failed to create environment: {err}")))?;
+        let watch_registration = match environment.as_ref() {
+            Some(environment) if !environment.is_remote() => {
+                self.skills_watcher
+                    .register_config(
+                        &config,
+                        self.skills_manager.as_ref(),
+                        self.plugins_manager.as_ref(),
+                        Some(environment.get_filesystem()),
+                    )
+                    .await
+            }
+            Some(_) | None => crate::file_watcher::WatchRegistration::default(),
+        };
         let CodexSpawnOk {
             codex, thread_id, ..
         } = Codex::spawn(CodexSpawnArgs {

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

@@ -2,10 +2,19 @@
 #![allow(clippy::unwrap_used, clippy::expect_used)]
 
 use anyhow::Result;
+use codex_core::ThreadManager;
+use codex_exec_server::CreateDirectoryOptions;
+use codex_exec_server::EnvironmentManager;
+use codex_exec_server::ExecutorFileSystem;
+use codex_login::CodexAuth;
+use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::SandboxPolicy;
+use codex_protocol::protocol::SessionSource;
 use codex_protocol::user_input::UserInput;
+use codex_utils_absolute_path::AbsolutePathBuf;
+use core_test_support::load_default_config_for_test;
 use core_test_support::responses::ev_assistant_message;
 use core_test_support::responses::ev_completed;
 use core_test_support::responses::ev_response_created;
@@ -14,16 +23,39 @@ use core_test_support::responses::sse;
 use core_test_support::responses::start_mock_server;
 use core_test_support::skip_if_no_network;
 use core_test_support::test_codex::test_codex;
+use pretty_assertions::assert_eq;
 use std::fs;
 use std::path::Path;
+use std::sync::Arc;
+use tempfile::TempDir;
 
-fn write_skill(home: &Path, name: &str, description: &str, body: &str) -> std::path::PathBuf {
-    let skill_dir = home.join("skills").join(name);
-    fs::create_dir_all(&skill_dir).unwrap();
+async fn write_repo_skill(
+    cwd: AbsolutePathBuf,
+    fs: Arc<dyn ExecutorFileSystem>,
+    name: &str,
+    description: &str,
+    body: &str,
+) -> Result<()> {
+    let skill_dir = cwd.join(".agents").join("skills").join(name);
+    fs.create_directory(
+        &skill_dir,
+        CreateDirectoryOptions { recursive: true },
+        /*sandbox*/ None,
+    )
+    .await?;
     let contents = format!("---\nname: {name}\ndescription: {description}\n---\n\n{body}\n");
     let path = skill_dir.join("SKILL.md");
-    fs::write(&path, contents).unwrap();
-    path
+    fs.write_file(&path, contents.into_bytes(), /*sandbox*/ None)
+        .await?;
+    Ok(())
+}
+
+fn write_home_skill(codex_home: &Path, dir: &str, name: &str, description: &str) -> Result<()> {
+    let skill_dir = codex_home.join("skills").join(dir);
+    fs::create_dir_all(&skill_dir)?;
+    let contents = format!("---\nname: {name}\ndescription: {description}\n---\n\n# Body\n");
+    fs::write(skill_dir.join("SKILL.md"), contents)?;
+    Ok(())
 }
 
 fn system_skill_md_path(home: impl AsRef<Path>, name: &str) -> std::path::PathBuf {
@@ -40,13 +72,18 @@ async fn user_turn_includes_skill_instructions() -> Result<()> {
 
     let server = start_mock_server().await;
     let skill_body = "skill body";
-    let mut builder = test_codex().with_pre_build_hook(|home| {
-        write_skill(home, "demo", "demo skill", skill_body);
+    let mut builder = test_codex().with_workspace_setup(move |cwd, fs| async move {
+        write_repo_skill(cwd, fs, "demo", "demo skill", skill_body).await
     });
-    let test = builder.build(&server).await?;
+    let test = builder.build_remote_aware(&server).await?;
 
-    let skill_path = test.codex_home_path().join("skills/demo/SKILL.md");
-    let skill_path = std::fs::canonicalize(skill_path)?;
+    let skill_path = test
+        .config
+        .cwd
+        .join(".agents/skills/demo/SKILL.md")
+        .canonicalize()
+        .unwrap_or_else(|_| test.config.cwd.join(".agents/skills/demo/SKILL.md"))
+        .to_path_buf();
 
     let mock = mount_sse_once(
         &server,
@@ -72,7 +109,7 @@ async fn user_turn_includes_skill_instructions() -> Result<()> {
                 },
             ],
             final_output_json_schema: None,
-            cwd: test.cwd_path().to_path_buf(),
+            cwd: test.config.cwd.to_path_buf(),
             approval_policy: AskForApproval::Never,
             approvals_reviewer: None,
             sandbox_policy: SandboxPolicy::DangerFullAccess,
@@ -106,6 +143,138 @@ async fn user_turn_includes_skill_instructions() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn list_skills_includes_repo_and_home_skills_remote_aware() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = start_mock_server().await;
+    let mut builder = test_codex()
+        .with_pre_build_hook(|home| {
+            write_home_skill(home, "home-demo", "home-demo", "from home")
+                .expect("write home skill");
+        })
+        .with_workspace_setup(|cwd, fs| async move {
+            write_repo_skill(cwd, fs, "repo-demo", "from repo", "# Body").await
+        });
+    let test = builder.build_remote_aware(&server).await?;
+
+    test.codex
+        .submit(Op::ListSkills {
+            cwds: Vec::new(),
+            force_reload: true,
+        })
+        .await?;
+    let response =
+        core_test_support::wait_for_event_match(test.codex.as_ref(), |event| match event {
+            codex_protocol::protocol::EventMsg::ListSkillsResponse(response) => {
+                Some(response.clone())
+            }
+            _ => None,
+        })
+        .await;
+
+    let cwd = test.config.cwd.as_path();
+    let skills = response
+        .skills
+        .iter()
+        .find(|entry| entry.cwd.as_path() == cwd)
+        .map(|entry| entry.skills.clone())
+        .unwrap_or_default();
+
+    let repo_skill = skills
+        .iter()
+        .find(|skill| skill.name == "repo-demo")
+        .expect("expected repo skill");
+    assert_eq!(repo_skill.scope, codex_protocol::protocol::SkillScope::Repo);
+    let repo_path = repo_skill.path.to_string_lossy().replace('\\', "/");
+    assert!(
+        repo_path.ends_with("/.agents/skills/repo-demo/SKILL.md"),
+        "unexpected repo skill path: {repo_path}"
+    );
+
+    let home_skill = skills
+        .iter()
+        .find(|skill| skill.name == "home-demo")
+        .expect("expected home skill");
+    assert_eq!(home_skill.scope, codex_protocol::protocol::SkillScope::User);
+    let home_path = home_skill.path.to_string_lossy().replace('\\', "/");
+    assert!(
+        home_path.ends_with("/skills/home-demo/SKILL.md"),
+        "unexpected home skill path: {home_path}"
+    );
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn list_skills_skips_cwd_roots_when_environment_disabled() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let cwd = TempDir::new()?;
+    write_home_skill(
+        codex_home.path(),
+        "home-disabled",
+        "home-disabled",
+        "from home",
+    )?;
+    let repo_skill_dir = cwd
+        .path()
+        .join(".agents")
+        .join("skills")
+        .join("repo-disabled");
+    fs::create_dir_all(&repo_skill_dir)?;
+    fs::write(
+        repo_skill_dir.join("SKILL.md"),
+        "---\nname: repo-disabled\ndescription: from repo\n---\n\n# Body\n",
+    )?;
+    let mut config = load_default_config_for_test(&codex_home).await;
+    config.cwd = AbsolutePathBuf::from_absolute_path_checked(cwd.path())?;
+
+    let thread_manager = ThreadManager::new(
+        &config,
+        codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
+        SessionSource::Exec,
+        CollaborationModesConfig::default(),
+        Arc::new(EnvironmentManager::new(Some("none".to_string()))),
+        /*analytics_events_client*/ None,
+    );
+    let new_thread = thread_manager.start_thread(config.clone()).await?;
+    let cwd = config.cwd.to_path_buf();
+
+    new_thread
+        .thread
+        .submit(Op::ListSkills {
+            cwds: vec![cwd.clone()],
+            force_reload: true,
+        })
+        .await?;
+    let response =
+        core_test_support::wait_for_event_match(new_thread.thread.as_ref(), |event| match event {
+            codex_protocol::protocol::EventMsg::ListSkillsResponse(response) => {
+                Some(response.clone())
+            }
+            _ => None,
+        })
+        .await;
+
+    assert_eq!(response.skills.len(), 1);
+    assert_eq!(response.skills[0].cwd, cwd);
+    assert_eq!(response.skills[0].errors.len(), 0);
+    assert!(
+        response.skills[0]
+            .skills
+            .iter()
+            .any(|skill| skill.name == "home-disabled")
+    );
+    assert!(
+        response.skills[0]
+            .skills
+            .iter()
+            .all(|skill| skill.name != "repo-disabled")
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn skill_load_errors_surface_in_session_configured() -> Result<()> {
     skip_if_no_network!(Ok(()));

codex-rs/utils/plugins/Cargo.toml

@@ -14,9 +14,12 @@ path = "src/lib.rs"
 workspace = true
 
 [dependencies]
+codex-exec-server = { workspace = true }
 codex-login = { workspace = true }
+codex-utils-absolute-path = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt"] }

codex-rs/utils/plugins/src/plugin_namespace.rs

@@ -1,7 +1,7 @@
 //! Resolve plugin namespace from skill file paths by walking ancestors for `plugin.json`.
 
-use std::fs;
-use std::path::Path;
+use codex_exec_server::ExecutorFileSystem;
+use codex_utils_absolute_path::AbsolutePathBuf;
 
 /// Relative path from a plugin root to its manifest file.
 pub const PLUGIN_MANIFEST_PATH: &str = ".codex-plugin/plugin.json";
@@ -13,12 +13,19 @@ struct RawPluginManifestName {
     name: String,
 }
 
-fn plugin_manifest_name(plugin_root: &Path) -> Option<String> {
+async fn plugin_manifest_name(
+    fs: &dyn ExecutorFileSystem,
+    plugin_root: &AbsolutePathBuf,
+) -> Option<String> {
     let manifest_path = plugin_root.join(PLUGIN_MANIFEST_PATH);
-    if !manifest_path.is_file() {
-        return None;
+    match fs.get_metadata(&manifest_path, /*sandbox*/ None).await {
+        Ok(metadata) if metadata.is_file => {}
+        Ok(_) | Err(_) => return None,
     }
-    let contents = fs::read_to_string(&manifest_path).ok()?;
+    let contents = fs
+        .read_file_text(&manifest_path, /*sandbox*/ None)
+        .await
+        .ok()?;
     let RawPluginManifestName { name: raw_name } = serde_json::from_str(&contents).ok()?;
     Some(
         plugin_root
@@ -32,9 +39,12 @@ fn plugin_manifest_name(plugin_root: &Path) -> Option<String> {
 
 /// Returns the plugin manifest `name` for the nearest ancestor of `path` that contains a valid
 /// plugin manifest (same `name` rules as full manifest loading in codex-core).
-pub fn plugin_namespace_for_skill_path(path: &Path) -> Option<String> {
+pub async fn plugin_namespace_for_skill_path(
+    fs: &dyn ExecutorFileSystem,
+    path: &AbsolutePathBuf,
+) -> Option<String> {
     for ancestor in path.ancestors() {
-        if let Some(name) = plugin_manifest_name(ancestor) {
+        if let Some(name) = plugin_manifest_name(fs, &ancestor).await {
             return Some(name);
         }
     }
@@ -44,11 +54,13 @@ pub fn plugin_namespace_for_skill_path(path: &Path) -> Option<String> {
 #[cfg(test)]
 mod tests {
     use super::plugin_namespace_for_skill_path;
+    use codex_exec_server::LOCAL_FS;
+    use codex_utils_absolute_path::test_support::PathBufExt;
     use std::fs;
     use tempfile::tempdir;
 
-    #[test]
-    fn uses_manifest_name() {
+    #[tokio::test]
+    async fn uses_manifest_name() {
         let tmp = tempdir().expect("tempdir");
         let plugin_root = tmp.path().join("plugins/sample");
         let skill_path = plugin_root.join("skills/search/SKILL.md");
@@ -63,7 +75,7 @@ mod tests {
         fs::write(&skill_path, "---\ndescription: search\n---\n").expect("write skill");
 
         assert_eq!(
-            plugin_namespace_for_skill_path(&skill_path),
+            plugin_namespace_for_skill_path(LOCAL_FS.as_ref(), &skill_path.abs()).await,
             Some("sample".to_string())
         );
     }