Add executor capability discovery V2 infrastructure (#47683)

## What changed

- Define `capabilities/discoverV2` request and inventory types and advertise `capabilityDiscoveryV2` in executor metadata.
- Prewarm installed plugin and global skill locations at server startup, with nonfatal scan failures.
- Add sandbox-scoped discovery caching with file watcher invalidation and request-time reloads when watching is unavailable.
- Batch metadata reads through the filesystem sandbox helper and bound discovery responses to 4 MiB and 2,048 capabilities.
- Share plugin identity and installed-version selection in `codex-core-plugin-common`, and fall back to `.app.json` when a plugin does not declare an apps configuration.

The V2 discovery manager is not yet wired into RPC request dispatch.

## Testing

Add coverage for inventory metadata, sandbox cache isolation, watcher invalidation, startup scan failures, response limits, and compatibility with older executor metadata.

GitOrigin-RevId: a223a5bc15fdf251023c0c71f468fd92c3b2d847

codex-rs/Cargo.lock

@@ -3020,6 +3020,17 @@ dependencies = [
  "codex-utils-absolute-path",
 ]
 
+[[package]]
+name = "codex-core-plugin-common"
+version = "0.0.0"
+dependencies = [
+ "codex-utils-absolute-path",
+ "semver",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "codex-core-plugins"
 version = "0.0.0"
@@ -3030,6 +3041,7 @@ dependencies = [
  "codex-app-server-protocol",
  "codex-config",
  "codex-connectors",
+ "codex-core-plugin-common",
  "codex-exec-server",
  "codex-exec-server-test-support",
  "codex-git-utils",
@@ -3055,7 +3067,6 @@ dependencies = [
  "libc",
  "pretty_assertions",
  "regex",
- "semver",
  "serde",
  "serde_json",
  "serde_with",
@@ -3148,15 +3159,18 @@ dependencies = [
  "codex-api",
  "codex-build-info",
  "codex-config",
+ "codex-core-plugin-common",
  "codex-exec-server-protocol",
  "codex-exec-server-test-support",
  "codex-file-system",
+ "codex-file-watcher",
  "codex-http-client",
  "codex-network-proxy",
  "codex-otel",
  "codex-protocol",
  "codex-sandboxing",
  "codex-shell-command",
+ "codex-skills",
  "codex-test-binary-support",
  "codex-utils-absolute-path",
  "codex-utils-cargo-bin",
@@ -4152,6 +4166,7 @@ name = "codex-plugin"
 version = "0.0.0"
 dependencies = [
  "codex-config",
+ "codex-core-plugin-common",
  "codex-protocol",
  "codex-utils-absolute-path",
  "codex-utils-path-uri",

codex-rs/Cargo.toml

@@ -44,6 +44,7 @@ members = [
     "skills",
     "core",
     "core-api",
+    "core-plugin-common",
     "core-plugins",
     "diagnostics",
     "guardian-context",
@@ -208,6 +209,7 @@ codex-connectors-extension = { path = "ext/connectors" }
 codex-context-fragments = { path = "context-fragments" }
 codex-core = { path = "core" }
 codex-core-api = { path = "core-api" }
+codex-core-plugin-common = { path = "core-plugin-common" }
 codex-core-plugins = { path = "core-plugins" }
 codex-diagnostics = { path = "diagnostics" }
 codex-exec = { path = "exec" }

codex-rs/core-plugin-common/BUILD.bazel

@@ -0,0 +1,15 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "core-plugin-common",
+    compile_data = glob(
+        include = ["**"],
+        allow_empty = True,
+        exclude = [
+            "**/* *",
+            "BUILD.bazel",
+            "Cargo.toml",
+        ],
+    ),
+    crate_name = "codex_core_plugin_common",
+)

codex-rs/core-plugin-common/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "codex-core-plugin-common"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[lib]
+name = "codex_core_plugin_common"
+path = "src/lib.rs"
+doctest = false
+
+[lints]
+workspace = true
+
+[dependencies]
+codex-utils-absolute-path = { workspace = true }
+semver = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }

codex-rs/core-plugin-common/src/installed.rs

@@ -0,0 +1,88 @@
+//! Read-only installed-version selection and persisted installation identity.
+
+use codex_utils_absolute_path::AbsolutePathBuf;
+use semver::Version;
+use serde::Deserialize;
+use serde::Serialize;
+use std::cmp::Ordering;
+use std::fs;
+
+pub const DEFAULT_PLUGIN_VERSION: &str = "local";
+pub const PLUGINS_CACHE_DIR: &str = "plugins/cache";
+pub const REMOTE_PLUGIN_INSTALL_METADATA_FILE: &str = ".codex-remote-plugin-install.json";
+pub const REMOTE_PLUGIN_INSTALL_METADATA_SCHEMA_VERSION: u8 = 1;
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct RemotePluginInstallMetadata {
+    pub schema_version: u8,
+    pub remote_plugin_id: String,
+}
+
+pub fn validate_plugin_version_segment(plugin_version: &str) -> Result<(), String> {
+    if plugin_version.is_empty() {
+        return Err("invalid plugin version: must not be empty".to_string());
+    }
+    if matches!(plugin_version, "." | "..") {
+        return Err("invalid plugin version: path traversal is not allowed".to_string());
+    }
+    if !plugin_version
+        .chars()
+        .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.' | '+'))
+    {
+        return Err(
+            "invalid plugin version: only ASCII letters, digits, `.`, `+`, `_`, and `-` are allowed"
+                .to_string(),
+        );
+    }
+    Ok(())
+}
+
+pub fn compare_plugin_versions(left: &str, right: &str) -> Ordering {
+    match (Version::parse(left), Version::parse(right)) {
+        (Ok(left), Ok(right)) => left.cmp(&right),
+        _ => left.cmp(right),
+    }
+}
+
+pub fn active_plugin_version(plugin_base_root: &AbsolutePathBuf) -> Option<String> {
+    let mut discovered_versions = fs::read_dir(plugin_base_root.as_path())
+        .ok()?
+        .filter_map(Result::ok)
+        .filter_map(|entry| {
+            entry.file_type().ok().filter(std::fs::FileType::is_dir)?;
+            entry.file_name().into_string().ok()
+        })
+        .filter(|version| validate_plugin_version_segment(version).is_ok())
+        .collect::<Vec<_>>();
+    discovered_versions.sort_unstable_by(|left, right| compare_plugin_versions(left, right));
+    if discovered_versions.is_empty() {
+        None
+    } else if discovered_versions
+        .iter()
+        .any(|version| version == DEFAULT_PLUGIN_VERSION)
+    {
+        Some(DEFAULT_PLUGIN_VERSION.to_string())
+    } else {
+        discovered_versions.pop()
+    }
+}
+
+/// Validates persisted installation metadata and returns its normalized remote identity.
+pub fn parse_remote_plugin_id(contents: &str) -> Result<String, String> {
+    let metadata: RemotePluginInstallMetadata = serde_json::from_str(contents)
+        .map_err(|err| format!("failed to parse remote plugin install metadata: {err}"))?;
+    if metadata.schema_version != REMOTE_PLUGIN_INSTALL_METADATA_SCHEMA_VERSION {
+        return Err(format!(
+            "unsupported remote plugin install metadata schema version: {}",
+            metadata.schema_version
+        ));
+    }
+    let remote_plugin_id = metadata.remote_plugin_id.trim();
+    if remote_plugin_id.is_empty() {
+        return Err(
+            "invalid remote plugin install metadata: remote plugin id must not be blank"
+                .to_string(),
+        );
+    }
+    Ok(remote_plugin_id.to_string())
+}

codex-rs/core-plugin-common/src/lib.rs

@@ -0,0 +1,4 @@
+//! Shared plugin identities and read-only installed-package rules.
+
+pub mod installed;
+pub mod plugin_id;

codex-rs/core-plugins/Cargo.toml

@@ -18,6 +18,7 @@ codex-analytics = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-config = { workspace = true }
 codex-connectors = { workspace = true }
+codex-core-plugin-common = { workspace = true }
 codex-exec-server = { workspace = true }
 codex-git-utils = { workspace = true }
 codex-hooks = { workspace = true }
@@ -41,7 +42,6 @@ flate2 = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
 regex = { workspace = true }
-semver = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 serde_with = { workspace = true }

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

@@ -9,9 +9,7 @@ use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_plugins::AgentPluginSchemaStatus;
 use codex_utils_plugins::agent_plugin_schema_status;
 use codex_utils_plugins::find_plugin_manifest_path;
-use semver::Version;
 use serde::Deserialize;
-use serde::Serialize;
 use serde_json::Value as JsonValue;
 use sha2::Digest;
 use sha2::Sha256;
@@ -22,19 +20,15 @@ use std::io::Write;
 use std::path::Path;
 use std::path::PathBuf;
 
-pub const DEFAULT_PLUGIN_VERSION: &str = "local";
-pub const PLUGINS_CACHE_DIR: &str = "plugins/cache";
+pub use codex_core_plugin_common::installed::DEFAULT_PLUGIN_VERSION;
+pub use codex_core_plugin_common::installed::PLUGINS_CACHE_DIR;
 pub const PLUGINS_DATA_DIR: &str = "plugins/data";
 const AGENT_PLUGINS_DATA_DIR: &str = "agent-plugins";
-const REMOTE_PLUGIN_INSTALL_METADATA_FILE: &str = ".codex-remote-plugin-install.json";
-const REMOTE_PLUGIN_INSTALL_METADATA_SCHEMA_VERSION: u8 = 1;
+use codex_core_plugin_common::installed::REMOTE_PLUGIN_INSTALL_METADATA_FILE;
+use codex_core_plugin_common::installed::REMOTE_PLUGIN_INSTALL_METADATA_SCHEMA_VERSION;
 const DEFAULT_AGENT_PLUGIN_VERSION: &str = "1.0.0";
 
-#[derive(Debug, Deserialize, Serialize)]
-struct RemotePluginInstallMetadata {
-    schema_version: u8,
-    remote_plugin_id: String,
-}
+use codex_core_plugin_common::installed::RemotePluginInstallMetadata;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PluginInstallResult {
@@ -69,26 +63,9 @@ impl ActivePluginInstallation {
                 ));
             }
         };
-        let metadata: RemotePluginInstallMetadata =
-            serde_json::from_str(&contents).map_err(|err| {
-                PluginStoreError::Invalid(format!(
-                    "failed to parse remote plugin install metadata: {err}"
-                ))
-            })?;
-        if metadata.schema_version != REMOTE_PLUGIN_INSTALL_METADATA_SCHEMA_VERSION {
-            return Err(PluginStoreError::Invalid(format!(
-                "unsupported remote plugin install metadata schema version: {}",
-                metadata.schema_version
-            )));
-        }
-        let remote_plugin_id = metadata.remote_plugin_id.trim();
-        if remote_plugin_id.is_empty() {
-            return Err(PluginStoreError::Invalid(
-                "invalid remote plugin install metadata: remote plugin id must not be blank"
-                    .to_string(),
-            ));
-        }
-        Ok(Some(remote_plugin_id.to_string()))
+        codex_core_plugin_common::installed::parse_remote_plugin_id(&contents)
+            .map(Some)
+            .map_err(PluginStoreError::Invalid)
     }
 }
 
@@ -167,26 +144,9 @@ impl PluginStore {
     }
 
     pub fn active_plugin_version(&self, plugin_id: &PluginId) -> Option<String> {
-        let mut discovered_versions = fs::read_dir(self.plugin_base_root(plugin_id).as_path())
-            .ok()?
-            .filter_map(Result::ok)
-            .filter_map(|entry| {
-                entry.file_type().ok().filter(std::fs::FileType::is_dir)?;
-                entry.file_name().into_string().ok()
-            })
-            .filter(|version| validate_plugin_version_segment(version).is_ok())
-            .collect::<Vec<_>>();
-        discovered_versions.sort_unstable_by(|left, right| compare_plugin_versions(left, right));
-        if discovered_versions.is_empty() {
-            None
-        } else if discovered_versions
-            .iter()
-            .any(|version| version == DEFAULT_PLUGIN_VERSION)
-        {
-            Some(DEFAULT_PLUGIN_VERSION.to_string())
-        } else {
-            discovered_versions.pop()
-        }
+        codex_core_plugin_common::installed::active_plugin_version(
+            &self.plugin_base_root(plugin_id),
+        )
     }
 
     pub fn active_plugin_root(&self, plugin_id: &PluginId) -> Option<AbsolutePathBuf> {
@@ -495,22 +455,7 @@ fn hex_prefix(bytes: &[u8], count: usize) -> String {
 }
 
 pub fn validate_plugin_version_segment(plugin_version: &str) -> Result<(), String> {
-    if plugin_version.is_empty() {
-        return Err("invalid plugin version: must not be empty".to_string());
-    }
-    if matches!(plugin_version, "." | "..") {
-        return Err("invalid plugin version: path traversal is not allowed".to_string());
-    }
-    if !plugin_version
-        .chars()
-        .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.' | '+'))
-    {
-        return Err(
-            "invalid plugin version: only ASCII letters, digits, `.`, `+`, `_`, and `-` are allowed"
-                .to_string(),
-        );
-    }
-    Ok(())
+    codex_core_plugin_common::installed::validate_plugin_version_segment(plugin_version)
 }
 
 fn plugin_manifest_for_source(
@@ -781,10 +726,7 @@ fn old_plugin_version_would_stay_active(old_version: &str, new_version: &str) ->
 }
 
 fn compare_plugin_versions(left: &str, right: &str) -> Ordering {
-    match (Version::parse(left), Version::parse(right)) {
-        (Ok(left), Ok(right)) => left.cmp(&right),
-        _ => left.cmp(right),
-    }
+    codex_core_plugin_common::installed::compare_plugin_versions(left, right)
 }
 
 fn copy_dir_recursive(source: &Path, target: &Path) -> Result<(), PluginStoreError> {

codex-rs/exec-server-protocol/src/capabilities.rs

@@ -0,0 +1,102 @@
+//! Executor capability inventory; callers apply identity and enablement.
+
+use codex_file_system::FileSystemSandboxContext;
+use codex_utils_path_uri::PathUri;
+use serde::Deserialize;
+use serde::Serialize;
+use std::fmt;
+
+use crate::CapabilityTextFile;
+use crate::DiscoveredPluginFiles;
+
+pub const CAPABILITIES_DISCOVER_V2_METHOD: &str = "capabilities/discoverV2";
+
+/// Discovers installed capabilities on the executor.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DiscoverV2CapabilitiesRequest {
+    pub cwd: PathUri,
+    #[serde(default)]
+    pub sandbox: Option<FileSystemSandboxContext>,
+}
+
+/// Discovered metadata before caller policy is applied.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DiscoverV2CapabilitiesResponse {
+    pub plugins: Vec<ExecutorPlugin>,
+    /// Skills outside installed packages.
+    pub skills: Vec<ExecutorSkill>,
+    #[serde(default)]
+    pub warnings: Vec<String>,
+}
+
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecutorPlugin {
+    /// Environment-local ID: `<name>@<marketplace>`.
+    pub id: String,
+    pub remote_plugin_id: Option<String>,
+    pub version: String,
+    pub root: PathUri,
+    pub files: DiscoveredPluginFiles,
+    pub skills: Vec<ExecutorSkill>,
+}
+
+/// Skill metadata, excluding instructions and resources.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecutorSkill {
+    pub name: String,
+    /// Nearest enclosing legacy plugin's name, following V1 precedence.
+    #[serde(default)]
+    pub namespace: Option<String>,
+    pub description: String,
+    #[serde(default)]
+    pub short_description: Option<String>,
+    /// Path to SKILL.md, not its containing directory.
+    pub path: PathUri,
+    pub metadata: Option<CapabilityTextFile>,
+}
+
+impl fmt::Debug for ExecutorPlugin {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ExecutorPlugin")
+            .field("id", &self.id)
+            .field("remote_plugin_id", &self.remote_plugin_id)
+            .field("version", &self.version)
+            .field("root", &self.root)
+            .field("manifest_path", &self.files.manifest.path)
+            .field(
+                "mcp_config_path",
+                &self.files.mcp_config.as_ref().map(|file| &file.path),
+            )
+            .field(
+                "apps_config_path",
+                &self.files.apps_config.as_ref().map(|file| &file.path),
+            )
+            .field("skills", &self.skills)
+            .finish()
+    }
+}
+
+impl fmt::Debug for ExecutorSkill {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Omit potentially sensitive descriptions and file contents.
+        formatter
+            .debug_struct("ExecutorSkill")
+            .field("name", &self.name)
+            .field("namespace", &self.namespace)
+            .field("path", &self.path)
+            .field(
+                "metadata_path",
+                &self.metadata.as_ref().map(|file| &file.path),
+            )
+            .finish()
+    }
+}
+
+#[cfg(test)]
+#[path = "capabilities_tests.rs"]
+mod tests;

codex-rs/exec-server-protocol/src/capabilities_tests.rs

@@ -0,0 +1,51 @@
+//! V2 discovery requests and responses round-trip through JSON.
+
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[test]
+fn discovery_contract_round_trips_json() -> Result<(), Box<dyn std::error::Error>> {
+    let request = DiscoverV2CapabilitiesRequest {
+        cwd: PathUri::parse("file:///workspace")?,
+        sandbox: None,
+    };
+    let wire = serde_json::to_value(&request)?;
+    assert_eq!(
+        serde_json::from_value::<DiscoverV2CapabilitiesRequest>(wire)?,
+        request
+    );
+    let text_file = CapabilityTextFile {
+        path: PathUri::parse("file:///plugin/config.json")?,
+        contents: "{}".to_string(),
+    };
+    let skill = ExecutorSkill {
+        name: "example".to_string(),
+        namespace: Some("nested".to_string()),
+        description: "Example skill".to_string(),
+        short_description: None,
+        path: PathUri::parse("file:///plugin/skills/example/SKILL.md")?,
+        metadata: Some(text_file.clone()),
+    };
+    let response = DiscoverV2CapabilitiesResponse {
+        plugins: vec![ExecutorPlugin {
+            id: "example@company".to_string(),
+            remote_plugin_id: None,
+            version: "local".to_string(),
+            root: PathUri::parse("file:///plugin")?,
+            files: DiscoveredPluginFiles {
+                manifest: text_file.clone(),
+                mcp_config: Some(text_file.clone()),
+                apps_config: Some(text_file),
+            },
+            skills: vec![skill.clone()],
+        }],
+        skills: vec![skill],
+        warnings: Vec::new(),
+    };
+    let wire = serde_json::to_value(&response)?;
+    assert_eq!(
+        serde_json::from_value::<DiscoverV2CapabilitiesResponse>(wire)?,
+        response
+    );
+    Ok(())
+}

codex-rs/exec-server-protocol/src/lib.rs

@@ -1,9 +1,11 @@
+mod capabilities;
 mod environment_config;
 mod network_policy;
 mod process_id;
 mod protocol;
 pub mod rpc;
 
+pub use capabilities::*;
 pub use environment_config::*;
 pub use network_policy::*;
 pub use process_id::ProcessId;

codex-rs/exec-server-protocol/src/protocol.rs

@@ -142,6 +142,9 @@ pub struct EnvironmentCapabilities {
     /// Whether capability discovery applies the filesystem sandbox sent with each root.
     #[serde(default)]
     pub capability_discovery_sandbox: bool,
+    /// Whether this executor supports V2 capability discovery.
+    #[serde(default)]
+    pub capability_discovery_v2: bool,
     /// Whether this executor supports the `environmentConfig/read` request.
     #[serde(default)]
     pub environment_config_read: bool,
@@ -248,6 +251,7 @@ impl EnvironmentInfo {
             capabilities: EnvironmentCapabilities {
                 network_proxy_launch: true,
                 capability_discovery_sandbox: true,
+                capability_discovery_v2: true,
                 environment_config_read: true,
                 http_header_env_vars: true,
                 sandboxed_file_streaming: true,
@@ -1158,6 +1162,20 @@ mod base64_bytes {
 #[cfg(test)]
 mod tests {
     use super::CapabilityRootDiscoverRequest;
+    #[test]
+    fn discovery_v2_support_defaults_off_for_older_executors() -> serde_json::Result<()> {
+        let legacy: super::EnvironmentCapabilities = serde_json::from_value(serde_json::json!({}))?;
+        assert!(!legacy.capability_discovery_v2);
+        let capabilities = super::EnvironmentInfo::local().capabilities;
+        assert_eq!(
+            serde_json::from_value::<super::EnvironmentCapabilities>(serde_json::to_value(
+                &capabilities
+            )?)?,
+            capabilities
+        );
+        Ok(())
+    }
+
     use super::EnvironmentCapabilities;
     use super::EnvironmentInfo;
     use super::ExecExitedNotification;
@@ -1366,6 +1384,7 @@ mod tests {
             EnvironmentCapabilities {
                 network_proxy_launch: true,
                 capability_discovery_sandbox: true,
+                capability_discovery_v2: false,
                 environment_config_read: false,
                 http_header_env_vars: false,
                 sandboxed_file_streaming: false,
@@ -1388,6 +1407,7 @@ mod tests {
             "capabilities": {
                 "networkProxyLaunch": false,
                 "capabilityDiscoverySandbox": false,
+                "capabilityDiscoveryV2": false,
                 "environmentConfigRead": false,
                 "httpHeaderEnvVars": false,
                 "sandboxedFileStreaming": false,

codex-rs/exec-server/Cargo.toml

@@ -7,10 +7,13 @@ license.workspace = true
 [lib]
 doctest = false
 
-# cargo-shear 1.11.2 misresolves this #[path] module in src/client.rs.
-# The module is compiled and exercised by the exec-server unit test target.
+# cargo-shear 1.11.2 misresolves these #[path] modules containing parent directories.
+# Both modules are compiled and exercised by the exec-server unit test target.
 [package.metadata.cargo-shear]
-ignored-paths = ["tests/unit/client_provisioning_tests.rs"]
+ignored-paths = [
+    "tests/unit/client_provisioning_tests.rs",
+    "src/discoverV2/tests/capability_watchers_tests.rs",
+]
 
 [lints]
 workspace = true
@@ -24,14 +27,17 @@ clatter = { workspace = true }
 codex-api = { workspace = true }
 codex-build-info = { workspace = true }
 codex-config = { workspace = true }
+codex-core-plugin-common = { workspace = true }
 codex-http-client = { workspace = true }
 codex-exec-server-protocol = { workspace = true }
 codex-file-system = { workspace = true }
+codex-file-watcher = { workspace = true }
 codex-network-proxy = { workspace = true }
 codex-otel = { workspace = true }
 codex-protocol = { workspace = true }
 codex-sandboxing = { workspace = true }
 codex-shell-command = { workspace = true }
+codex-skills = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-home-dir = { workspace = true }
 codex-utils-path-uri = { workspace = true }

codex-rs/exec-server/src/capability_discovery.rs

@@ -24,10 +24,11 @@ const MAX_DIRECTORIES_PER_ROOT: usize = 2_000;
 const MAX_ENTRIES_PER_ROOT: usize = 20_000;
 const MAX_FILE_BYTES: usize = 1024 * 1024;
 const MAX_BUNDLE_BYTES_PER_ROOT: usize = 16 * 1024 * 1024;
-const MAX_CONCURRENT_ROOTS: usize = 8;
+pub(crate) const MAX_CONCURRENT_ROOTS: usize = 8;
 const SKILL_FILE_NAME: &str = "SKILL.md";
 const SKILL_METADATA_PATH: &str = "agents/openai.yaml";
 const DEFAULT_MCP_CONFIG_PATH: &str = ".mcp.json";
+const DEFAULT_APP_CONFIG_PATH: &str = ".app.json";
 
 #[derive(Debug, thiserror::Error)]
 pub enum CapabilityDiscoveryError {
@@ -60,7 +61,7 @@ pub async fn discover_capability_roots(
     Ok(CapabilityRootsDiscoverResponse { roots })
 }
 
-async fn discover_root(
+pub(crate) async fn discover_root(
     file_system: &dyn ExecutorFileSystem,
     request: CapabilityRootDiscoverRequest,
 ) -> CapabilityRootDiscovery {
@@ -234,7 +235,10 @@ async fn discover_root(
             }
             None => None,
         };
-        let apps_config = match declarations.apps_config {
+        let apps_path = declarations
+            .apps_config
+            .or_else(|| path.join(DEFAULT_APP_CONFIG_PATH).ok());
+        let apps_config = match apps_path {
             Some(path) => {
                 read_optional_text_file(
                     file_system,

codex-rs/exec-server/src/discoverV2/capability_discoveries.rs

@@ -0,0 +1,377 @@
+//! Loads metadata within the request's sandbox and builds a bounded response.
+
+use std::collections::HashMap;
+use std::io;
+use std::sync::Arc;
+
+use codex_file_system::ExecutorFileSystem;
+use codex_skills::parse_skill_frontmatter_metadata;
+use futures::StreamExt;
+use serde::Deserialize;
+use serde::Serialize;
+
+use crate::CapabilityRootDiscoverRequest;
+use crate::CapabilityRootDiscovery;
+use crate::DiscoverV2CapabilitiesResponse;
+use crate::ExecutorPlugin;
+use crate::ExecutorSkill;
+use crate::LocalFileSystem;
+use crate::capability_discovery::MAX_CONCURRENT_ROOTS;
+use crate::capability_discovery::discover_root;
+use crate::discover_v2::capability_locations::CapabilityLocation;
+use crate::discover_v2::capability_locations::CapabilityLocations;
+
+const MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
+const MAX_CAPABILITIES: usize = 2_048;
+const MAX_WARNINGS: usize = 64;
+// Match V1: namespace, separator, and skill name.
+const MAX_QUALIFIED_NAME_LEN: usize = 64 * 2 + 1;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+enum DiscoveredSource {
+    Plugin(Box<ExecutorPlugin>),
+    Skills(Vec<ExecutorSkill>),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+struct CapabilityDiscovery {
+    content: DiscoveredSource,
+    warnings: Vec<String>,
+}
+
+impl Default for CapabilityDiscovery {
+    fn default() -> Self {
+        Self {
+            content: DiscoveredSource::Skills(Vec::new()),
+            warnings: Vec::new(),
+        }
+    }
+}
+
+pub(super) async fn load_capability_discoveries(
+    file_system: &LocalFileSystem,
+    locations: &CapabilityLocations,
+    sandbox: Option<&crate::FileSystemSandboxContext>,
+) -> io::Result<Arc<DiscoverV2CapabilitiesResponse>> {
+    let sandbox = sandbox.filter(|sandbox| sandbox.should_read_from_sandbox());
+    let warnings = if !locations.warnings.is_empty() {
+        vec!["some capability locations could not be resolved; discovery may be incomplete".into()]
+    } else {
+        Vec::new()
+    };
+    let response = match sandbox {
+        Some(sandbox) if !locations.locations.is_empty() => {
+            file_system
+                .sandboxed()?
+                .load_sandboxed_capability_discoveries(
+                    locations.locations.clone(),
+                    warnings,
+                    sandbox,
+                )
+                .await?
+        }
+        _ => load_capability_discovery_batch(file_system, &locations.locations, warnings).await?,
+    };
+    Ok(Arc::new(response))
+}
+
+/// Loads roots concurrently and bounds the combined response.
+pub(crate) async fn load_capability_discovery_batch(
+    file_system: &LocalFileSystem,
+    locations: &[CapabilityLocation],
+    warnings: Vec<String>,
+) -> io::Result<DiscoverV2CapabilitiesResponse> {
+    let mut response = DiscoverV2CapabilitiesResponse {
+        warnings,
+        ..Default::default()
+    };
+    let mut budget = ResponseBudget::default();
+    // Use V1 concurrency; admit results in location order.
+    let mut discoveries = futures::stream::iter(locations)
+        .map(|location| load_capability_discovery(file_system, location.clone()))
+        .buffered(MAX_CONCURRENT_ROOTS)
+        .enumerate();
+    while let Some((index, discovery)) = discoveries.next().await {
+        let mut discovery = discovery?;
+        if !discovery.warnings.is_empty() {
+            discovery.warnings = vec![
+                "some capability metadata could not be loaded; discovery may be incomplete".into(),
+            ];
+        }
+        append_capability_discovery(&mut response, discovery, &mut budget)?;
+        // A rejected oversized entry alone does not rule out smaller entries later.
+        if (budget.capabilities == MAX_CAPABILITIES || budget.bytes == MAX_RESPONSE_BYTES)
+            && index + 1 < locations.len()
+        {
+            budget.truncated = true;
+            break;
+        }
+    }
+    finish_response(response, budget.truncated)
+}
+
+async fn load_capability_discovery(
+    file_system: &LocalFileSystem,
+    location: CapabilityLocation,
+) -> io::Result<CapabilityDiscovery> {
+    // Missing roots are normal.
+    match file_system
+        .get_metadata(&location.root, Default::default(), /*sandbox*/ None)
+        .await
+    {
+        Ok(metadata) if metadata.is_directory => {}
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            return Ok(CapabilityDiscovery::default());
+        }
+        Ok(_) => return Ok(CapabilityDiscovery::default()),
+        Err(error) if error.kind() == io::ErrorKind::PermissionDenied => {
+            return Ok(CapabilityDiscovery::default());
+        }
+        Err(error) => {
+            return Ok(CapabilityDiscovery {
+                warnings: vec![format!(
+                    "could not inspect capability root {}: {error}",
+                    location.root
+                )],
+                ..Default::default()
+            });
+        }
+    }
+    let discovery = discover_root(
+        file_system,
+        CapabilityRootDiscoverRequest {
+            id: location
+                .plugin
+                .as_ref()
+                .map(|plugin| plugin.id.clone())
+                .unwrap_or_default(),
+            path: location.root.clone(),
+            sandbox: None,
+        },
+    )
+    .await;
+    Ok(parse_root(location, discovery))
+}
+
+fn parse_root(
+    location: CapabilityLocation,
+    discovery: CapabilityRootDiscovery,
+) -> CapabilityDiscovery {
+    let mut warnings = discovery.warnings;
+    warnings.extend(discovery.error);
+    if location.plugin.is_some() && discovery.plugin.is_none() {
+        warnings.push(format!(
+            "installed plugin at {} has no readable manifest",
+            location.root
+        ));
+        return CapabilityDiscovery {
+            warnings,
+            ..Default::default()
+        };
+    }
+
+    // V1 precedence: first manifest per root, nearest root per skill.
+    let mut plugin_namespaces = HashMap::new();
+    for manifest in &discovery.namespace_manifests {
+        #[derive(Deserialize)]
+        struct ManifestName {
+            #[serde(default)]
+            name: String,
+        }
+        let Some(root) = manifest.path.parent().and_then(|path| path.parent()) else {
+            continue;
+        };
+        let Ok(parsed) = serde_json::from_str::<ManifestName>(&manifest.contents) else {
+            continue;
+        };
+        let name = if parsed.name.trim().is_empty() {
+            let Some(name) = root.basename() else {
+                continue;
+            };
+            name
+        } else {
+            parsed.name
+        };
+        plugin_namespaces.entry(root).or_insert(name);
+    }
+
+    // Reuse the contents already read by V1.
+    let mut skills = Vec::new();
+    'skills: for skill in discovery.skills {
+        let parsed = parse_skill_frontmatter_metadata(&skill.instructions.contents, || {
+            skill
+                .instructions
+                .path
+                .parent()
+                .and_then(|path| path.basename())
+                .map(|name| name.split_whitespace().collect::<Vec<_>>().join(" "))
+                .filter(|name| !name.is_empty())
+                .unwrap_or_else(|| "skill".to_string())
+        });
+        let Ok(parsed) = parsed else {
+            warnings.push(format!(
+                "invalid skill metadata at {}",
+                skill.instructions.path
+            ));
+            continue;
+        };
+        let path = skill.instructions.path;
+        let mut ancestor = path.parent();
+        let mut namespace = None;
+        while let Some(root) = ancestor {
+            if let Some(name) = plugin_namespaces.get(&root) {
+                // Check V1's name limit before cloning the namespace.
+                if name.chars().take(MAX_QUALIFIED_NAME_LEN + 1).count()
+                    + 1
+                    + parsed.name.chars().count()
+                    > MAX_QUALIFIED_NAME_LEN
+                {
+                    warnings.push(format!(
+                        "invalid skill metadata at {path}: qualified name exceeds {MAX_QUALIFIED_NAME_LEN} characters"
+                    ));
+                    continue 'skills;
+                }
+                namespace = Some(name.clone());
+                break;
+            }
+            ancestor = root.parent();
+        }
+        skills.push(ExecutorSkill {
+            name: parsed.name,
+            namespace,
+            description: parsed.description,
+            short_description: parsed.short_description,
+            path,
+            metadata: skill.metadata,
+        });
+    }
+    let content = if let (Some(plugin), Some(files)) = (location.plugin, discovery.plugin) {
+        DiscoveredSource::Plugin(Box::new(ExecutorPlugin {
+            id: plugin.id,
+            remote_plugin_id: plugin.remote_plugin_id,
+            version: plugin.version,
+            root: location.root,
+            files,
+            skills,
+        }))
+    } else {
+        DiscoveredSource::Skills(skills)
+    };
+    CapabilityDiscovery { content, warnings }
+}
+
+#[derive(Default)]
+struct ResponseBudget {
+    capabilities: usize,
+    bytes: usize,
+    truncated: bool,
+}
+
+impl ResponseBudget {
+    fn admit(&mut self, capabilities: usize, bytes: usize) -> bool {
+        if self.capabilities + capabilities > MAX_CAPABILITIES
+            || self.bytes + bytes > MAX_RESPONSE_BYTES
+        {
+            self.truncated = true;
+            return false;
+        }
+        self.capabilities += capabilities;
+        self.bytes += bytes;
+        true
+    }
+}
+
+/// Appends within budget; duplicate skills count until consumers deduplicate them.
+fn append_capability_discovery(
+    response: &mut DiscoverV2CapabilitiesResponse,
+    discovery: CapabilityDiscovery,
+    budget: &mut ResponseBudget,
+) -> io::Result<()> {
+    response.warnings.extend(
+        discovery
+            .warnings
+            .into_iter()
+            .take(MAX_WARNINGS.saturating_sub(response.warnings.len())),
+    );
+
+    match discovery.content {
+        DiscoveredSource::Plugin(plugin) => {
+            let remaining = MAX_CAPABILITIES - budget.capabilities;
+            if remaining == 0 {
+                budget.truncated = true;
+                return Ok(());
+            }
+            let mut plugin = *plugin;
+            budget.truncated |= plugin.skills.len() > remaining - 1;
+            plugin.skills.truncate(remaining - 1); // The plugin itself also counts.
+            let size = serde_json::to_vec(&plugin).map_err(io::Error::other)?.len();
+            if budget.admit(1 + plugin.skills.len(), size) {
+                response.plugins.push(plugin);
+            }
+        }
+        DiscoveredSource::Skills(skills) => {
+            for skill in skills {
+                if budget.capabilities == MAX_CAPABILITIES {
+                    budget.truncated = true;
+                    break;
+                }
+                let size = serde_json::to_vec(&skill).map_err(io::Error::other)?.len();
+                if budget.admit(/*capabilities*/ 1, size) {
+                    response.skills.push(skill);
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+fn finish_response(
+    mut response: DiscoverV2CapabilitiesResponse,
+    truncated: bool,
+) -> io::Result<DiscoverV2CapabilitiesResponse> {
+    response.plugins.sort_unstable_by(|left, right| {
+        left.id
+            .cmp(&right.id)
+            .then_with(|| left.root.to_string().cmp(&right.root.to_string()))
+    });
+    // Keep standalone skill precedence.
+    response.warnings.truncate(MAX_WARNINGS);
+
+    enforce_response_size(&mut response, truncated)?;
+    Ok(response)
+}
+
+// Include JSON overhead and warnings in the final size check.
+fn enforce_response_size(
+    response: &mut DiscoverV2CapabilitiesResponse,
+    truncated: bool,
+) -> io::Result<()> {
+    if truncated
+        || serde_json::to_vec(&*response)
+            .map_err(io::Error::other)?
+            .len()
+            > MAX_RESPONSE_BYTES
+    {
+        response.warnings.truncate(MAX_WARNINGS - 1);
+        response
+            .warnings
+            .insert(0, "capability discovery size limit reached".to_string());
+        while serde_json::to_vec(&*response)
+            .map_err(io::Error::other)?
+            .len()
+            > MAX_RESPONSE_BYTES
+        {
+            // Drop extra warnings first; keep the limit warning.
+            if response.warnings.len() > 1 {
+                response.warnings.truncate(1);
+            } else if response.skills.pop().is_none() && response.plugins.pop().is_none() {
+                break;
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+#[path = "tests/capability_discoveries_tests.rs"]
+mod tests;

codex-rs/exec-server/src/discoverV2/capability_locations.rs

@@ -0,0 +1,213 @@
+//! Prewarms installed cache packages and fixed standalone skill locations before discovery.
+
+mod candidates;
+use candidates::CandidateCapabilityLocations;
+
+use std::collections::BTreeMap;
+use std::io;
+
+use crate::LocalFileSystem;
+use codex_config::loader::system_config_toml_file;
+use codex_file_system::ExecutorFileSystem;
+use codex_file_system::WalkEntryKind;
+use codex_file_system::WalkOptions;
+use codex_file_watcher::WatchPath;
+use codex_utils_path_uri::PathUri;
+use serde::Deserialize;
+use serde::Serialize;
+
+use codex_core_plugin_common::installed::PLUGINS_CACHE_DIR;
+use codex_core_plugin_common::installed::REMOTE_PLUGIN_INSTALL_METADATA_FILE;
+use codex_core_plugin_common::installed::active_plugin_version;
+use codex_core_plugin_common::installed::parse_remote_plugin_id;
+use codex_core_plugin_common::plugin_id::PluginId;
+use tokio::io::AsyncReadExt;
+
+/// Home paths are supplied by the executor, not read from the sandbox helper's sanitized env.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct CapabilityLocationRequest {
+    pub codex_home: PathUri,
+    pub user_home: Option<PathUri>,
+}
+
+const MAX_LOCATIONS: usize = 512;
+const MAX_REMOTE_INSTALL_METADATA_BYTES: u64 = 16 * 1024;
+const MAX_REMOTE_PLUGIN_ID_BYTES: usize = 1024;
+const AGENTS_SKILLS_SUFFIX: &str = ".agents/skills";
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct CapabilityLocation {
+    pub root: PathUri,
+    pub plugin: Option<InstalledPlugin>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct InstalledPlugin {
+    pub id: String,
+    pub version: String,
+    pub remote_plugin_id: Option<String>,
+}
+
+/// Selected locations; membership does not authorize capability reads.
+#[derive(Clone, Default)]
+pub(crate) struct CapabilityLocations {
+    pub locations: Vec<CapabilityLocation>,
+    pub watches: Vec<WatchPath>,
+    pub warnings: Vec<String>,
+}
+
+/// Resolves fixed home/cache roots on the executor without reading configuration or using a cwd.
+pub(crate) async fn prewarm_locations(
+    fs: &LocalFileSystem,
+    request: &CapabilityLocationRequest,
+) -> io::Result<CapabilityLocations> {
+    let codex_home = request.codex_home.to_abs_path()?;
+    let user_home = request
+        .user_home
+        .as_ref()
+        .map(PathUri::to_abs_path)
+        .transpose()?;
+    // Fixed standalone roots extend Flora's V1 root; project skill discovery is deferred.
+    let mut skill_roots = vec![codex_home.join("skills")];
+    if let Some(home) = user_home {
+        skill_roots.push(home.join(AGENTS_SKILLS_SUFFIX));
+    }
+    if let Some(system_root) = system_config_toml_file()?.parent() {
+        skill_roots.push(system_root.join("skills"));
+    }
+    CandidateCapabilityLocations::resolve(
+        fs,
+        Some(PathUri::from_abs_path(&codex_home.join(PLUGINS_CACHE_DIR))),
+        skill_roots
+            .into_iter()
+            .map(|root| PathUri::from_abs_path(&root))
+            .collect(),
+    )
+    .await
+}
+
+async fn discover_cached_plugins(
+    fs: &LocalFileSystem,
+    cache: &PathUri,
+) -> io::Result<CapabilityLocations> {
+    let mut locations = CapabilityLocations::default();
+    // Enumerate packages; the shared selector scans their version directories.
+    let walk = match fs
+        .walk(
+            cache,
+            WalkOptions {
+                max_depth: 1,
+                max_directories: 2_000,
+                max_entries: 20_000,
+                follow_directory_symlinks: true,
+                prune_hidden_directories: false,
+            },
+            /*sandbox*/ None,
+        )
+        .await
+    {
+        Ok(walk) => walk,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(locations),
+        Err(error) => return Err(error),
+    };
+    // Reported walk errors must not become a cached partial inventory.
+    if let Some(error) = walk.errors.into_iter().next() {
+        return Err(io::Error::other(format!(
+            "failed to scan plugin cache path {}: {}",
+            error.path, error.message
+        )));
+    }
+    if walk.truncated {
+        locations
+            .warnings
+            .push("plugin cache traversal limit reached".to_string());
+    }
+    let mut packages = BTreeMap::new();
+    for entry in walk.entries {
+        if entry.kind != WalkEntryKind::Directory {
+            continue;
+        }
+        let Some(relative) = entry.path.relative_path_from(cache) else {
+            continue;
+        };
+        let parts = relative.split(['/', '\\']).collect::<Vec<_>>();
+        let [marketplace, name] = parts.as_slice() else {
+            continue;
+        };
+        let Ok(id) = PluginId::new((*name).to_string(), (*marketplace).to_string()) else {
+            continue;
+        };
+        packages.insert(id.as_key(), entry.path.to_abs_path()?);
+    }
+    let installed = tokio::task::spawn_blocking(move || {
+        packages
+            .into_iter()
+            .filter_map(|(id, package_root)| {
+                let version = active_plugin_version(&package_root)?;
+                Some((id, package_root, version))
+            })
+            .collect::<Vec<_>>()
+    })
+    .await
+    .map_err(io::Error::other)?;
+    for (id, package_root, version) in installed {
+        // Avoid reading metadata for plugin locations the final assembly would discard.
+        if locations.locations.len() == MAX_LOCATIONS {
+            locations.warnings.push(format!(
+                "plugin cache reached its {MAX_LOCATIONS}-location limit"
+            ));
+            break;
+        }
+        let metadata_path = package_root.join(REMOTE_PLUGIN_INSTALL_METADATA_FILE);
+        let remote_plugin_id = match read_remote_plugin_id(metadata_path.as_path()).await {
+            Ok(remote_plugin_id) => remote_plugin_id,
+            Err(error) => {
+                locations.warnings.push(format!(
+                    "could not read remote installation metadata for {id}: {error}"
+                ));
+                None
+            }
+        };
+        locations.locations.push(CapabilityLocation {
+            root: PathUri::from_abs_path(&package_root.join(&version)),
+            plugin: Some(InstalledPlugin {
+                id,
+                version,
+                remote_plugin_id,
+            }),
+        });
+    }
+    Ok(locations)
+}
+
+async fn read_remote_plugin_id(path: &std::path::Path) -> io::Result<Option<String>> {
+    // Nonblocking open and handle validation reject FIFOs and other non-regular files.
+    let file = match crate::regular_file::open(path).await {
+        Ok(file) => file,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(error),
+    };
+    let mut contents = Vec::new();
+    file.take(MAX_REMOTE_INSTALL_METADATA_BYTES + 1)
+        .read_to_end(&mut contents)
+        .await?;
+    if contents.len() as u64 > MAX_REMOTE_INSTALL_METADATA_BYTES {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "remote installation metadata exceeds its byte limit",
+        ));
+    }
+    let contents = std::str::from_utf8(&contents).map_err(io::Error::other)?;
+    let id = parse_remote_plugin_id(contents).map_err(io::Error::other)?;
+    if id.len() > MAX_REMOTE_PLUGIN_ID_BYTES {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "remote plugin ID exceeds its byte limit",
+        ));
+    }
+    Ok(Some(id))
+}
+
+#[cfg(test)]
+#[path = "tests/capability_locations_tests.rs"]
+pub(super) mod tests;

codex-rs/exec-server/src/discoverV2/capability_locations/candidates.rs

@@ -0,0 +1,123 @@
+//! Retains installed-cache and standalone-skill candidates and assembles their locations.
+
+use super::CapabilityLocation;
+use super::CapabilityLocations;
+use super::MAX_LOCATIONS;
+use super::discover_cached_plugins;
+use crate::LocalFileSystem;
+use codex_file_system::ExecutorFileSystem;
+use codex_file_watcher::WatchPath;
+use codex_utils_path_uri::PathUri;
+use std::io;
+
+const MAX_LOCATION_WARNINGS: usize = 64;
+
+/// Existing candidates plus watch paths for existing and not-yet-created roots.
+pub(super) struct CandidateCapabilityLocations {
+    candidates: Vec<CandidateCapabilityLocation>,
+    watches: Vec<WatchPath>,
+}
+
+/// An installed cache or standalone skill root whose locations can be resolved.
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum CandidateCapabilityLocation {
+    Cache(PathUri),
+    Standalone(PathUri),
+}
+
+impl CandidateCapabilityLocations {
+    /// Resolves existing cache and skill roots into one bounded location inventory.
+    pub(super) async fn resolve(
+        fs: &LocalFileSystem,
+        cache_root: Option<PathUri>,
+        skill_roots: Vec<PathUri>,
+    ) -> io::Result<CapabilityLocations> {
+        let mut candidates = cache_root
+            .into_iter()
+            .map(CandidateCapabilityLocation::Cache)
+            .collect::<Vec<_>>();
+        let mut standalone = skill_roots;
+        let mut seen = std::collections::HashSet::new();
+        standalone.retain(|root| seen.insert(root.clone()));
+        candidates.extend(
+            standalone
+                .into_iter()
+                .map(CandidateCapabilityLocation::Standalone),
+        );
+
+        let mut watches = Vec::new();
+        let mut retained_candidates = Vec::new();
+        for candidate in candidates {
+            let root = match &candidate {
+                CandidateCapabilityLocation::Cache(root)
+                | CandidateCapabilityLocation::Standalone(root) => root,
+            };
+            let exists = match fs
+                .get_metadata(root, Default::default(), /*sandbox*/ None)
+                .await
+            {
+                Ok(metadata) => metadata.is_directory,
+                Err(error) if error.kind() == io::ErrorKind::NotFound => false,
+                Err(error) => return Err(error),
+            };
+            // Missing roots use the watcher's ancestor fallback until they are created.
+            watches.push(WatchPath {
+                path: root.to_abs_path()?.into_path_buf(),
+                recursive: true,
+            });
+            if exists {
+                retained_candidates.push(candidate);
+            }
+        }
+        let candidates = Self {
+            candidates: retained_candidates,
+            watches,
+        };
+        let mut resolved = Vec::new();
+        for candidate in &candidates.candidates {
+            resolved.push(candidates.resolve_candidates(fs, candidate).await?);
+        }
+        Ok(candidates.assemble(resolved))
+    }
+
+    async fn resolve_candidates(
+        &self,
+        fs: &LocalFileSystem,
+        candidate: &CandidateCapabilityLocation,
+    ) -> io::Result<CapabilityLocations> {
+        match candidate {
+            CandidateCapabilityLocation::Cache(root) => discover_cached_plugins(fs, root).await,
+            // Standalone roots need no manifest or version selection.
+            CandidateCapabilityLocation::Standalone(root) => Ok(CapabilityLocations {
+                locations: vec![CapabilityLocation {
+                    root: root.clone(),
+                    plugin: None,
+                }],
+                ..Default::default()
+            }),
+        }
+    }
+
+    /// Combines and bounds inventory; Core applies plugin enablement.
+    fn assemble(&self, resolved_candidates: Vec<CapabilityLocations>) -> CapabilityLocations {
+        let mut result = CapabilityLocations {
+            watches: self.watches.clone(),
+            ..Default::default()
+        };
+        // Combine installed plugin and standalone skill locations in discovery order.
+        for candidate in resolved_candidates {
+            result.locations.extend(candidate.locations);
+            result.warnings.extend(candidate.warnings);
+        }
+
+        if result.locations.len() > MAX_LOCATIONS {
+            result.warnings.truncate(MAX_LOCATION_WARNINGS - 1);
+            result.warnings.push(format!(
+                "capability discovery reached its {MAX_LOCATIONS}-location limit"
+            ));
+            result.locations.truncate(MAX_LOCATIONS);
+        }
+        result.warnings.truncate(MAX_LOCATION_WARNINGS);
+        result
+    }
+}

codex-rs/exec-server/src/discoverV2/capability_manager.rs

@@ -0,0 +1,188 @@
+//! Retains watched global locations and sandbox-scoped response snapshots.
+//! Replacement snapshots isolate in-flight loads; unavailable watchers force request-time reloads.
+
+use crate::DiscoverV2CapabilitiesRequest;
+use crate::DiscoverV2CapabilitiesResponse;
+use crate::FileSystemSandboxContext;
+use crate::LocalFileSystem;
+use crate::discover_v2::capability_discoveries::load_capability_discoveries;
+use crate::discover_v2::capability_locations::CapabilityLocationRequest;
+use crate::discover_v2::capability_locations::CapabilityLocations;
+use crate::discover_v2::capability_locations::prewarm_locations;
+use codex_file_watcher::WatchRegistration;
+use std::io;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use tokio::sync::Mutex;
+use tokio::sync::OnceCell;
+use tokio::sync::Semaphore;
+
+mod capability_watchers;
+use capability_watchers::CapabilityWatchers;
+
+const MAX_CACHED_SANDBOX_CONTEXTS: usize = 16;
+
+pub(crate) struct CapabilityManager {
+    file_system: LocalFileSystem,
+    prewarmed_locations: OnceCell<(CapabilityLocationRequest, CapabilityLocations)>,
+    current_state: Mutex<Option<Arc<LocationState>>>,
+    location_refresh_lock: Semaphore,
+    watchers: CapabilityWatchers,
+}
+
+struct LocationState {
+    dirty: AtomicBool,
+    locations: CapabilityLocations,
+    sandbox_discoveries: Mutex<Vec<Arc<CachedSandboxDiscovery>>>,
+    _registration: WatchRegistration,
+}
+
+struct CachedSandboxDiscovery {
+    sandbox: Option<FileSystemSandboxContext>,
+    response: OnceCell<Arc<DiscoverV2CapabilitiesResponse>>,
+}
+
+impl CapabilityManager {
+    pub(crate) fn new(file_system: LocalFileSystem) -> Arc<Self> {
+        Arc::new(Self {
+            file_system,
+            prewarmed_locations: OnceCell::new(),
+            current_state: Mutex::default(),
+            location_refresh_lock: Semaphore::new(/*permits*/ 1),
+            watchers: CapabilityWatchers::new(),
+        })
+    }
+
+    pub(crate) async fn prewarm_locations(
+        self: &Arc<Self>,
+        request: CapabilityLocationRequest,
+    ) -> io::Result<&(CapabilityLocationRequest, CapabilityLocations)> {
+        let cached = self
+            .prewarmed_locations
+            .get_or_try_init(|| async {
+                let locations = prewarm_locations(&self.file_system, &request).await?;
+                Ok::<_, io::Error>((request.clone(), locations))
+            })
+            .await?;
+        // Check the winning request even when another caller initialized the cell.
+        if cached.0 != request {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "capability home paths differ from the prewarmed configuration",
+            ));
+        }
+        self.initialize_watchers(&cached.1).await;
+        Ok(cached)
+    }
+
+    /// Reuses startup locations, retrying prewarming if the background scan failed.
+    pub(crate) async fn get_or_refresh_discovery(
+        self: &Arc<Self>,
+        location_request: CapabilityLocationRequest,
+        params: DiscoverV2CapabilitiesRequest,
+    ) -> io::Result<Arc<DiscoverV2CapabilitiesResponse>> {
+        let (location_request, _) = match self.prewarm_locations(location_request).await {
+            Ok(prewarmed) => prewarmed,
+            Err(error) => {
+                return Ok(Arc::new(DiscoverV2CapabilitiesResponse {
+                    warnings: vec![format!(
+                        "could not prewarm capability discovery: {:?}",
+                        error.kind()
+                    )],
+                    ..Default::default()
+                }));
+            }
+        };
+        let location_state = self.get_or_refresh_locations(location_request).await?;
+        let discovery = {
+            let mut discoveries = location_state.sandbox_discoveries.lock().await;
+            if let Some(discovery) = discoveries
+                .iter()
+                .find(|discovery| discovery.sandbox == params.sandbox)
+            {
+                Arc::clone(discovery)
+            } else {
+                let discovery = Arc::new(CachedSandboxDiscovery {
+                    sandbox: params.sandbox,
+                    response: OnceCell::new(),
+                });
+                if discoveries.len() == MAX_CACHED_SANDBOX_CONTEXTS {
+                    // Evict the oldest context (FIFO); in-flight requests retain their own Arc.
+                    discoveries.remove(0);
+                }
+                discoveries.push(Arc::clone(&discovery));
+                discovery
+            }
+        };
+        if self.watchers.has_subscription() {
+            return discovery
+                .response
+                .get_or_try_init(|| {
+                    load_capability_discoveries(
+                        &self.file_system,
+                        &location_state.locations,
+                        discovery.sandbox.as_ref(),
+                    )
+                })
+                .await
+                .map(Arc::clone);
+        }
+        load_capability_discoveries(
+            &self.file_system,
+            &location_state.locations,
+            discovery.sandbox.as_ref(),
+        )
+        .await
+    }
+
+    async fn get_or_refresh_locations(
+        &self,
+        request: &CapabilityLocationRequest,
+    ) -> io::Result<Arc<LocationState>> {
+        if self.watchers.has_subscription()
+            && let Some(location_state) = self.location_snapshot().await
+            && !location_state.dirty.load(Ordering::Relaxed)
+        {
+            return Ok(location_state);
+        }
+        let _refresh = self
+            .location_refresh_lock
+            .acquire()
+            .await
+            .map_err(io::Error::other)?;
+        if self.watchers.has_subscription()
+            && let Some(location_state) = self.location_snapshot().await
+            && !location_state.dirty.load(Ordering::Relaxed)
+        {
+            return Ok(location_state);
+        }
+        // Retry dirty locations; without a subscription, reload on every request.
+        let locations = match prewarm_locations(&self.file_system, request).await {
+            Ok(locations) => locations,
+            Err(error) => {
+                if let Some(location_state) = self.location_snapshot().await {
+                    tracing::warn!(error_kind = ?error.kind(), "capability refresh failed; retrying on the next request");
+                    return Ok(location_state);
+                }
+                return Err(error);
+            }
+        };
+        let next = Arc::new(LocationState {
+            dirty: AtomicBool::new(false),
+            _registration: self.watchers.register_paths(&locations),
+            locations,
+            sandbox_discoveries: Mutex::default(),
+        });
+        *self.current_state.lock().await = Some(Arc::clone(&next));
+        Ok(next)
+    }
+
+    async fn location_snapshot(&self) -> Option<Arc<LocationState>> {
+        self.current_state.lock().await.clone()
+    }
+}
+
+#[cfg(test)]
+#[path = "tests/capability_manager_tests.rs"]
+mod tests;

codex-rs/exec-server/src/discoverV2/capability_manager/capability_watchers.rs

@@ -0,0 +1,146 @@
+//! Refreshes global locations and invalidates sandbox responses through one watcher subscription.
+//! In-flight readers retain the previous snapshot while a replacement is prepared.
+
+use super::CapabilityManager;
+use super::LocationState;
+use crate::discover_v2::capability_locations::CapabilityLocations;
+use crate::discover_v2::capability_locations::prewarm_locations;
+use codex_file_watcher::FileWatcher;
+use codex_file_watcher::FileWatcherSubscriber;
+use codex_file_watcher::ThrottledWatchReceiver;
+use codex_file_watcher::WatchRegistration;
+use std::sync::Arc;
+use std::sync::OnceLock;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio_util::task::AbortOnDropHandle;
+
+#[cfg(not(test))]
+const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_secs(10);
+#[cfg(test)]
+const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_millis(50);
+
+pub(super) struct CapabilityWatchers {
+    file_watcher: Option<Arc<FileWatcher>>,
+    subscription: OnceLock<RootSubscription>,
+}
+
+struct RootSubscription {
+    subscriber: FileWatcherSubscriber,
+    _listener: AbortOnDropHandle<()>,
+}
+
+impl CapabilityWatchers {
+    pub(super) fn new() -> Self {
+        let file_watcher = match FileWatcher::new() {
+            Ok(watcher) => Some(Arc::new(watcher)),
+            Err(error) => {
+                tracing::warn!(%error, "capability watcher unavailable; using uncached discovery");
+                None
+            }
+        };
+        Self {
+            file_watcher,
+            subscription: OnceLock::new(),
+        }
+    }
+
+    pub(super) fn has_subscription(&self) -> bool {
+        self.subscription.get().is_some()
+    }
+
+    /// Registers fixed roots, including missing ones, without separate watches for symlink targets.
+    pub(super) fn register_paths(&self, locations: &CapabilityLocations) -> WatchRegistration {
+        let Some(subscription) = self.subscription.get() else {
+            return WatchRegistration::default();
+        };
+        subscription
+            .subscriber
+            .register_paths(locations.watches.clone())
+    }
+}
+
+impl CapabilityManager {
+    pub(super) async fn initialize_watchers(self: &Arc<Self>, prewarmed: &CapabilityLocations) {
+        let Some(watcher) = &self.watchers.file_watcher else {
+            return;
+        };
+        if self.watchers.subscription.get().is_some() {
+            return;
+        }
+        let Ok(_refresh) = self.location_refresh_lock.acquire().await else {
+            return;
+        };
+        if self.watchers.subscription.get().is_some() {
+            return;
+        }
+        let (subscriber, events) = watcher.add_subscriber();
+        let mut events = ThrottledWatchReceiver::new(events, WATCHER_THROTTLE_INTERVAL);
+        let registration = subscriber.register_paths(prewarmed.watches.clone());
+        let next = Arc::new(LocationState {
+            dirty: AtomicBool::new(false),
+            locations: prewarmed.clone(),
+            sandbox_discoveries: Mutex::default(),
+            _registration: registration,
+        });
+        let mut current_state = self.current_state.lock().await;
+        let weak = Arc::downgrade(self);
+        let listener = tokio::spawn(async move {
+            while events.recv().await.is_some() {
+                let Some(manager) = weak.upgrade() else { break };
+                manager.refresh_locations().await;
+            }
+        });
+        // No await between retaining the subscription and publishing its snapshot.
+        if self
+            .watchers
+            .subscription
+            .set(RootSubscription {
+                subscriber,
+                _listener: AbortOnDropHandle::new(listener),
+            })
+            .is_err()
+        {
+            return;
+        }
+        *current_state = Some(next);
+    }
+
+    pub(super) async fn refresh_locations(&self) {
+        let result: std::io::Result<()> = async {
+            let _refresh = self
+                .location_refresh_lock
+                .acquire()
+                .await
+                .map_err(std::io::Error::other)?;
+            let Some((request, _)) = self.prewarmed_locations.get() else {
+                return Ok(());
+            };
+            // Keep failed or cancelled refreshes retryable without discarding the last snapshot.
+            if let Some(location_state) = self.location_snapshot().await {
+                location_state.dirty.store(true, Ordering::Relaxed);
+            }
+            let locations = prewarm_locations(&self.file_system, request).await?;
+            let registration = self.watchers.register_paths(&locations);
+            let next = Arc::new(LocationState {
+                dirty: AtomicBool::new(false),
+                locations,
+                sandbox_discoveries: Mutex::default(),
+                _registration: registration,
+            });
+            // Requests populate the new cache lazily; in-flight loads retain their old snapshot.
+            *self.current_state.lock().await = Some(next);
+            Ok(())
+        }
+        .await;
+        if let Err(error) = result {
+            tracing::warn!(error_kind = ?error.kind(), "capability refresh failed; keeping previous complete snapshot");
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "../tests/capability_watchers_tests.rs"]
+pub(super) mod tests;

codex-rs/exec-server/src/discoverV2/mod.rs

@@ -0,0 +1,9 @@
+//! Executor-side V2 capability discovery.
+//! Connection and RPC routing stay outside this module; legacy V1 scanning remains separate.
+
+// Staged implementation; the RPC is registered in E04.
+#![allow(dead_code)]
+
+pub(crate) mod capability_discoveries;
+pub(crate) mod capability_locations;
+pub(crate) mod capability_manager;

codex-rs/exec-server/src/discoverV2/tests/capability_discoveries_tests.rs

@@ -0,0 +1,219 @@
+//! Covers metadata parity and inventory discovery independent of plugin enablement.
+
+use pretty_assertions::assert_eq;
+
+use super::MAX_CAPABILITIES;
+use super::MAX_RESPONSE_BYTES;
+use super::MAX_WARNINGS;
+use crate::CapabilityRootDiscoverRequest;
+use crate::DiscoverV2CapabilitiesRequest;
+use crate::ExecutorSkill;
+use crate::LocalFileSystem;
+use crate::capability_discovery::discover_root;
+use crate::discover_v2::capability_locations::tests::MANIFEST;
+use crate::discover_v2::capability_locations::tests::request;
+use crate::discover_v2::capability_locations::tests::write;
+use crate::discover_v2::capability_manager::CapabilityManager;
+use codex_utils_path_uri::PathUri;
+
+const SKILL: &str =
+    "---\nname: deploy\ndescription: Deploy the service.\n---\nPRIVATE_INSTRUCTION_BODY_SENTINEL\n";
+
+#[tokio::test]
+async fn discovery_preserves_inventory_and_v1_metadata() -> anyhow::Result<()> {
+    let temp = tempfile::tempdir()?;
+    let request = request(temp.path())?;
+    write(
+        temp.path(),
+        "codex/config.toml",
+        "[plugins.\"demo@test\"]\n[plugins.\"disabled@test\"]\nenabled = false\n",
+    )?;
+    // Inventory includes disabled and unconfigured plugins; Core applies enablement.
+    for name in ["disabled", "unconfigured"] {
+        write(
+            temp.path(),
+            &format!("codex/plugins/cache/test/{name}/local/.codex-plugin/plugin.json"),
+            MANIFEST,
+        )?;
+    }
+    let root = "codex/plugins/cache/test/demo/local";
+    write(
+        temp.path(),
+        &format!("{root}/.codex-plugin/plugin.json"),
+        r#"{"name":"demo","skills":"./skills"}"#,
+    )?;
+    let mcp_path = ".mcp.json";
+    write(
+        temp.path(),
+        &format!("{root}/{mcp_path}"),
+        r#"{"mcpServers":{}}"#,
+    )?;
+    for relative in [
+        "skills/deploy/SKILL.md",
+        "examples/deploy/SKILL.md",
+        ".codex-plugin/migrated-command-skills/deploy/SKILL.md",
+        "skills/a/b/c/d/e/f/deploy/SKILL.md",
+    ] {
+        write(temp.path(), &format!("{root}/{relative}"), SKILL)?;
+    }
+    write(
+        temp.path(),
+        "codex/skills/deploy  skill/SKILL.md",
+        &SKILL.replace("name: deploy\n", ""),
+    )?;
+    for nested in [format!("{root}/nested"), "codex/skills/nested".to_string()] {
+        write(
+            temp.path(),
+            &format!("{nested}/.codex-plugin/plugin.json"),
+            r#"{"name":"team"}"#,
+        )?;
+        write(
+            temp.path(),
+            &format!("{nested}/skills/deploy/SKILL.md"),
+            SKILL,
+        )?;
+    }
+    let file_system = LocalFileSystem::unsandboxed();
+    let plugin_root = request.codex_home.join("plugins/cache/test/demo/local")?;
+    let v1 = discover_root(
+        &file_system,
+        CapabilityRootDiscoverRequest {
+            id: "demo@test".to_string(),
+            path: plugin_root.clone(),
+            sandbox: None,
+        },
+    )
+    .await;
+    // V1 keeps undeclared/migrated skills and excludes the skill beyond its depth limit.
+    assert_eq!(v1.skills.len(), 4);
+    let manager = CapabilityManager::new(file_system);
+    manager.prewarm_locations(request.clone()).await?;
+    let response = manager
+        .get_or_refresh_discovery(
+            request.clone(),
+            DiscoverV2CapabilitiesRequest {
+                cwd: PathUri::from_host_native_path(temp.path().join("project"))?,
+                sandbox: None,
+            },
+        )
+        .await?;
+    assert_eq!(
+        response
+            .plugins
+            .iter()
+            .map(|plugin| plugin.id.as_str())
+            .collect::<Vec<_>>(),
+        vec!["demo@test", "disabled@test", "unconfigured@test"],
+    );
+    let plugin = response
+        .plugins
+        .first()
+        .ok_or_else(|| anyhow::anyhow!("missing demo plugin"))?;
+    assert_eq!(
+        plugin.files.manifest.path,
+        plugin_root.join(".codex-plugin/plugin.json")?,
+    );
+    assert_eq!(
+        plugin.files.mcp_config.as_ref().map(|file| &file.path),
+        Some(&plugin_root.join(mcp_path)?),
+    );
+    assert_eq!(plugin.root, plugin_root);
+    assert_eq!(plugin.version, "local");
+    assert_eq!(Some(&plugin.files), v1.plugin.as_ref());
+    assert_eq!(
+        plugin.skills,
+        v1.skills
+            .into_iter()
+            .map(|skill| {
+                let namespace = if skill
+                    .instructions
+                    .path
+                    .starts_with(&plugin.root.join("nested")?)
+                {
+                    "team"
+                } else {
+                    "demo"
+                };
+                Ok(expected_skill(skill.instructions.path, Some(namespace)))
+            })
+            .collect::<anyhow::Result<Vec<_>>>()?
+    );
+    assert_eq!(
+        response.skills,
+        vec![
+            ExecutorSkill {
+                name: "deploy skill".to_string(),
+                ..expected_skill(
+                    request.codex_home.join("skills/deploy  skill/SKILL.md")?,
+                    /*namespace*/ None,
+                )
+            },
+            expected_skill(
+                request
+                    .codex_home
+                    .join("skills/nested/skills/deploy/SKILL.md")?,
+                Some("team")
+            ),
+        ],
+    );
+    assert_eq!(response.warnings, v1.warnings);
+    let json = serde_json::to_string(&response)?;
+    assert!(
+        response.plugins.len() + plugin.skills.len() + response.skills.len() <= MAX_CAPABILITIES
+    );
+    assert!(json.len() <= MAX_RESPONSE_BYTES);
+    assert!(response.warnings.len() <= MAX_WARNINGS);
+    assert!(!json.contains("PRIVATE_INSTRUCTION_BODY_SENTINEL"));
+    Ok(())
+}
+
+fn expected_skill(path: PathUri, namespace: Option<&str>) -> ExecutorSkill {
+    ExecutorSkill {
+        name: "deploy".to_string(),
+        namespace: namespace.map(str::to_string),
+        description: "Deploy the service.".to_string(),
+        short_description: None,
+        path,
+        metadata: None,
+    }
+}
+
+#[tokio::test]
+async fn discovery_reports_location_failures_without_paths() -> anyhow::Result<()> {
+    use codex_protocol::models::PermissionProfile;
+    use codex_protocol::permissions::FileSystemSandboxPolicy;
+    use codex_protocol::permissions::NetworkSandboxPolicy;
+
+    let temp = tempfile::tempdir()?;
+    request(temp.path())?;
+    let locations = crate::discover_v2::capability_locations::CapabilityLocations {
+        warnings: vec!["could not scan plugin cache /private/plugins/cache".into()],
+        ..Default::default()
+    };
+    let sandbox = crate::FileSystemSandboxContext::from_permission_profile(
+        PermissionProfile::from_runtime_permissions(
+            &FileSystemSandboxPolicy::restricted(Vec::new()),
+            NetworkSandboxPolicy::Restricted,
+        ),
+        PathUri::from_host_native_path(temp.path().join("project"))?,
+    );
+    for sandbox in [None, Some(&sandbox)] {
+        let response = super::load_capability_discoveries(
+            &LocalFileSystem::unsandboxed(),
+            &locations,
+            sandbox,
+        )
+        .await?;
+        assert_eq!(
+            *response,
+            crate::DiscoverV2CapabilitiesResponse {
+                warnings: vec![
+                    "some capability locations could not be resolved; discovery may be incomplete"
+                        .into(),
+                ],
+                ..Default::default()
+            }
+        );
+    }
+    Ok(())
+}

codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs

@@ -0,0 +1,125 @@
+//! Exercises installed-package and fixed skill location prewarming on real filesystem fixtures.
+
+use pretty_assertions::assert_eq;
+use std::path::Path;
+
+use codex_utils_path_uri::PathUri;
+
+use super::CapabilityLocation;
+use super::CapabilityLocationRequest;
+use super::InstalledPlugin;
+use super::prewarm_locations;
+use crate::LocalFileSystem;
+
+pub(crate) const MANIFEST: &str = r#"{"name":"demo","version":"1.10.0"}"#;
+
+pub(crate) fn request(root: &Path) -> anyhow::Result<CapabilityLocationRequest> {
+    for path in ["codex", "home", "project"] {
+        std::fs::create_dir_all(root.join(path))?;
+    }
+    Ok(CapabilityLocationRequest {
+        codex_home: PathUri::from_host_native_path(root.join("codex"))?,
+        user_home: Some(PathUri::from_host_native_path(root.join("home"))?),
+    })
+}
+
+pub(crate) fn write(root: &Path, relative: &str, contents: &str) -> anyhow::Result<()> {
+    let path = root.join(relative);
+    let parent = path
+        .parent()
+        .ok_or_else(|| anyhow::anyhow!("test fixture path has no parent"))?;
+    std::fs::create_dir_all(parent)?;
+    std::fs::write(path, contents)?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn prewarming_resolves_global_locations_without_cwd_or_config() -> anyhow::Result<()> {
+    let temp = tempfile::tempdir()?;
+    let request = request(temp.path())?;
+    // Prewarming must not read configuration.
+    write(temp.path(), "codex/config.toml", "invalid TOML [")?;
+    for root in [
+        "codex/skills",
+        "home/.agents/skills",
+        "project/.agents/skills",
+    ] {
+        write(temp.path(), &format!("{root}/demo/SKILL.md"), "unparsed")?;
+    }
+    for version in ["1.9.0", "1.10.0"] {
+        write(
+            temp.path(),
+            &format!("codex/plugins/cache/team/demo/{version}/.codex-plugin/plugin.json"),
+            MANIFEST,
+        )?;
+    }
+    write(
+        temp.path(),
+        "codex/plugins/cache/team/demo/.codex-remote-plugin-install.json",
+        r#"{"schema_version":1,"remote_plugin_id":"plugins~demo"}"#,
+    )?;
+    let locations = prewarm_locations(&LocalFileSystem::unsandboxed(), &request).await?;
+    let mut actual = locations.locations;
+    actual.sort_unstable_by_key(|location| location.root.to_string());
+    let mut expected = vec![CapabilityLocation {
+        root: request.codex_home.join("plugins/cache/team/demo/1.10.0")?,
+        plugin: Some(InstalledPlugin {
+            id: "demo@team".to_string(),
+            version: "1.10.0".to_string(),
+            remote_plugin_id: Some("plugins~demo".to_string()),
+        }),
+    }];
+    for root in ["codex/skills", "home/.agents/skills"] {
+        expected.push(CapabilityLocation {
+            root: PathUri::from_host_native_path(temp.path().join(root))?,
+            plugin: None,
+        });
+    }
+    expected.sort_unstable_by_key(|location| location.root.to_string());
+    assert_eq!(actual, expected);
+    assert!(locations.warnings.is_empty());
+
+    let metadata = "codex/plugins/cache/team/demo/.codex-remote-plugin-install.json";
+    for contents in [
+        "invalid JSON".to_string(),
+        " ".repeat(super::MAX_REMOTE_INSTALL_METADATA_BYTES as usize + 1),
+    ] {
+        write(temp.path(), metadata, &contents)?;
+        let locations = prewarm_locations(&LocalFileSystem::unsandboxed(), &request).await?;
+        assert!(locations.warnings.iter().any(|warning| {
+            warning.contains("could not read remote installation metadata for demo@team")
+        }));
+        assert!(locations.locations.iter().any(|location| {
+            location
+                .plugin
+                .as_ref()
+                .is_some_and(|plugin| plugin.id == "demo@team" && plugin.remote_plugin_id.is_none())
+        }));
+    }
+    // Missing remote metadata is valid for locally installed plugins.
+    std::fs::remove_file(temp.path().join(metadata))?;
+    let locations = prewarm_locations(&LocalFileSystem::unsandboxed(), &request).await?;
+    assert!(locations.warnings.is_empty());
+    assert!(locations.locations.iter().any(|location| {
+        location
+            .plugin
+            .as_ref()
+            .is_some_and(|plugin| plugin.id == "demo@team" && plugin.remote_plugin_id.is_none())
+    }));
+
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+
+        let directory = temp.path().join("codex/plugins/cache/team");
+        let permissions = std::fs::metadata(&directory)?.permissions();
+        std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o000))?;
+        let result = prewarm_locations(&LocalFileSystem::unsandboxed(), &request).await;
+        std::fs::set_permissions(&directory, permissions)?;
+        assert!(
+            result.is_err(),
+            "cache traversal failure must not retain partial locations"
+        );
+    }
+    Ok(())
+}

codex-rs/exec-server/src/discoverV2/tests/capability_manager_tests.rs

@@ -0,0 +1,202 @@
+//! Covers bounded context caching, shared initialization, cancellation retry, and cached warnings.
+
+use super::*;
+use crate::discover_v2::capability_locations::tests::request;
+use crate::discover_v2::capability_locations::tests::write;
+use codex_protocol::protocol::SandboxPolicy;
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+
+async fn setup() -> anyhow::Result<(
+    tempfile::TempDir,
+    CapabilityLocationRequest,
+    Arc<CapabilityManager>,
+)> {
+    let directory = tempfile::tempdir()?;
+    write(
+        directory.path(),
+        "codex/skills/demo/SKILL.md",
+        "---\nname: demo\ndescription: original\n---\nbody never cached",
+    )?;
+    let request = request(&std::fs::canonicalize(directory.path())?)?;
+    let file_system = LocalFileSystem::unsandboxed();
+    let manager = CapabilityManager::new(file_system);
+    manager.prewarm_locations(request.clone()).await?;
+    capability_watchers::tests::stop_listener(&manager).await?;
+    Ok((directory, request, manager))
+}
+
+#[tokio::test]
+async fn discovery_responses_are_cached_by_sandbox_context() -> anyhow::Result<()> {
+    let (directory, request, manager) = setup().await?;
+    let cwd = PathUri::from_host_native_path(directory.path().join("project"))?;
+    let other_cwd = PathUri::from_host_native_path(directory.path().join("other"))?;
+    let params = DiscoverV2CapabilitiesRequest {
+        cwd: cwd.clone(),
+        sandbox: None,
+    };
+    let first = manager
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    assert_eq!(first.skills.len(), 1);
+    let snapshot = manager
+        .location_snapshot()
+        .await
+        .ok_or_else(|| anyhow::anyhow!("missing snapshot"))?;
+    let other = manager
+        .get_or_refresh_discovery(
+            request.clone(),
+            DiscoverV2CapabilitiesRequest {
+                cwd: other_cwd.clone(),
+                ..params.clone()
+            },
+        )
+        .await?;
+    assert!(Arc::ptr_eq(&first, &other));
+
+    // The full sandbox context, including its own cwd, still identifies a separate cache.
+    let mut previous = Arc::clone(&first);
+    for index in 0..MAX_CACHED_SANDBOX_CONTEXTS {
+        let sandbox_cwd = cwd.join(&format!("context-{index}"))?;
+        let sandbox = DiscoverV2CapabilitiesRequest {
+            sandbox: Some(FileSystemSandboxContext::from_legacy_sandbox_policy(
+                SandboxPolicy::DangerFullAccess,
+                sandbox_cwd,
+            )?),
+            ..params.clone()
+        };
+        let response = manager
+            .get_or_refresh_discovery(request.clone(), sandbox.clone())
+            .await?;
+        assert_eq!(response, previous);
+        assert!(!Arc::ptr_eq(&response, &previous));
+        assert!(Arc::ptr_eq(
+            &response,
+            &manager
+                .get_or_refresh_discovery(request.clone(), sandbox)
+                .await?
+        ));
+        previous = response;
+    }
+    assert_eq!(
+        snapshot.sandbox_discoveries.lock().await.len(),
+        MAX_CACHED_SANDBOX_CONTEXTS
+    );
+    // The oldest context is evicted, but its previously returned response stays valid.
+    let reloaded = manager.get_or_refresh_discovery(request, params).await?;
+    assert_eq!(reloaded, first);
+    assert!(!Arc::ptr_eq(&reloaded, &first));
+    assert_eq!(
+        snapshot.sandbox_discoveries.lock().await.len(),
+        MAX_CACHED_SANDBOX_CONTEXTS
+    );
+    Ok(())
+}
+
+#[test]
+fn cancelled_initialization_retries_and_warning_responses_are_cached() -> anyhow::Result<()> {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .max_blocking_threads(1)
+        .build()?
+        .block_on(async {
+            let (directory, request, manager) = setup().await?;
+            let params = DiscoverV2CapabilitiesRequest {
+                cwd: PathUri::from_host_native_path(directory.path().join("project"))?,
+                sandbox: None,
+            };
+            let snapshot = manager
+                .location_snapshot()
+                .await
+                .ok_or_else(|| anyhow::anyhow!("missing snapshot"))?;
+            // Hold the only blocking worker so discovery's filesystem read cannot finish early.
+            // Dropping the sender also releases the worker if an assertion unwinds.
+            let (release, wait) = std::sync::mpsc::channel::<()>();
+            let (started, ready) = tokio::sync::oneshot::channel();
+            let blocker = tokio::task::spawn_blocking(move || {
+                let _ = started.send(());
+                let _ = wait.recv();
+            });
+            ready.await?;
+            let mut first =
+                Box::pin(manager.get_or_refresh_discovery(request.clone(), params.clone()));
+            assert!(futures::poll!(&mut first).is_pending());
+            let discovery = snapshot
+                .sandbox_discoveries
+                .lock()
+                .await
+                .first()
+                .cloned()
+                .ok_or_else(|| anyhow::anyhow!("missing sandbox discovery"))?;
+            let mut second =
+                Box::pin(manager.get_or_refresh_discovery(request.clone(), params.clone()));
+            assert!(futures::poll!(&mut second).is_pending());
+            assert!(discovery.response.get().is_none());
+            assert!(snapshot.sandbox_discoveries.try_lock().is_ok());
+            // Cancel the initializer; the waiting request must be able to initialize instead.
+            drop(first);
+            drop(release);
+            blocker.await?;
+            let original = second.await?;
+            assert_eq!(original.skills.len(), 1);
+            assert!(Arc::ptr_eq(
+                &original,
+                &manager
+                    .get_or_refresh_discovery(request.clone(), params.clone())
+                    .await?,
+            ));
+
+            write(
+                directory.path(),
+                "codex/skills/demo/SKILL.md",
+                "invalid skill metadata",
+            )?;
+            // Invalidation drops entries instead of overwriting their initialized responses.
+            snapshot.sandbox_discoveries.lock().await.clear();
+            let (partial, concurrent) = tokio::try_join!(
+                manager.get_or_refresh_discovery(request.clone(), params.clone()),
+                manager.get_or_refresh_discovery(request.clone(), params.clone()),
+            )?;
+            assert!(Arc::ptr_eq(&partial, &concurrent));
+            assert!(partial.skills.is_empty());
+            assert!(!partial.warnings.is_empty());
+            assert!(Arc::ptr_eq(
+                &partial,
+                &manager
+                    .get_or_refresh_discovery(request.clone(), params.clone())
+                    .await?
+            ));
+
+            write(
+                directory.path(),
+                "codex/skills/demo/SKILL.md",
+                "---\nname: demo\ndescription: recovered\n---\n",
+            )?;
+            assert!(Arc::ptr_eq(
+                &partial,
+                &manager
+                    .get_or_refresh_discovery(request.clone(), params.clone())
+                    .await?
+            ));
+            snapshot.sandbox_discoveries.lock().await.clear();
+            let recovered = manager
+                .get_or_refresh_discovery(request.clone(), params.clone())
+                .await?;
+            assert!(recovered.warnings.is_empty());
+            assert_eq!(
+                recovered
+                    .skills
+                    .iter()
+                    .map(|skill| skill.description.as_str())
+                    .collect::<Vec<_>>(),
+                vec!["recovered"]
+            );
+            assert!(Arc::ptr_eq(
+                &recovered,
+                &manager
+                    .get_or_refresh_discovery(request.clone(), params)
+                    .await?
+            ));
+            Ok(())
+        })
+}

codex-rs/exec-server/src/discoverV2/tests/capability_watchers_tests.rs

@@ -0,0 +1,287 @@
+//! Watch events invalidate discovery; no subscription keeps requests uncached.
+
+use super::*;
+use crate::DiscoverV2CapabilitiesRequest;
+use crate::LocalFileSystem;
+use crate::discover_v2::capability_locations::tests::request;
+use crate::discover_v2::capability_locations::tests::write;
+use anyhow::Context;
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+use std::time::Duration;
+
+// Keep native events from racing deterministic cache assertions.
+pub(in crate::discover_v2::capability_manager) async fn stop_listener(
+    manager: &CapabilityManager,
+) -> anyhow::Result<()> {
+    manager
+        .watchers
+        .subscription
+        .get()
+        .context("watcher unavailable")?
+        ._listener
+        .abort();
+    drop(manager.location_refresh_lock.acquire().await?);
+    Ok(())
+}
+
+#[tokio::test]
+async fn watch_events_invalidate_discovery_and_no_subscription_disables_caching()
+-> anyhow::Result<()> {
+    let directory = tempfile::tempdir()?;
+    let root = std::fs::canonicalize(directory.path())?;
+    write(
+        &root,
+        "codex/skills/demo/SKILL.md",
+        "---\nname: demo\ndescription: original\n---\n",
+    )?;
+    let request = request(&root)?;
+    let demo_skill = request.codex_home.join("skills/demo/SKILL.md")?;
+    let manager = CapabilityManager::new(LocalFileSystem::unsandboxed());
+    let params = DiscoverV2CapabilitiesRequest {
+        cwd: PathUri::from_host_native_path(&root)?,
+        sandbox: None,
+    };
+    let original = manager
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    assert!(
+        original
+            .skills
+            .iter()
+            .any(|skill| skill.path == demo_skill && skill.description == "original")
+    );
+    let initial = manager
+        .location_snapshot()
+        .await
+        .context("missing initial snapshot")?;
+
+    // Wait for native delivery before testing one-shot creation of missing roots.
+    tokio::time::timeout(Duration::from_secs(10), async {
+        let mut interval = tokio::time::interval(Duration::from_millis(50));
+        loop {
+            interval.tick().await;
+            write(
+                &root,
+                "codex/skills/demo/SKILL.md",
+                "---\nname: demo\ndescription: ready\n---\n",
+            )?;
+            let snapshot = manager
+                .location_snapshot()
+                .await
+                .context("missing snapshot")?;
+            if !Arc::ptr_eq(&initial, &snapshot) {
+                assert!(snapshot.sandbox_discoveries.lock().await.is_empty());
+                break anyhow::Ok(());
+            }
+        }
+    })
+    .await
+    .context("native watcher did not deliver its first content event")??;
+
+    // Neither the plugin cache nor the user skills directory existed at startup.
+    write(
+        &root,
+        "codex/plugins/cache/acme/gmail/1.0.0/.codex-plugin/plugin.json",
+        r#"{"name":"gmail"}"#,
+    )?;
+    write(
+        &root,
+        "home/.agents/skills/added/SKILL.md",
+        "---\nname: added\ndescription: new global root\n---\n",
+    )?;
+    let home_skills = PathUri::from_host_native_path(root.join("home/.agents/skills"))?;
+    tokio::time::timeout(Duration::from_secs(10), async {
+        let mut interval = tokio::time::interval(Duration::from_millis(50));
+        loop {
+            interval.tick().await;
+            // Inspect only locations: a request-time rescan must not hide a missed event.
+            let snapshot = manager
+                .location_snapshot()
+                .await
+                .context("missing snapshot")?;
+            if snapshot
+                .locations
+                .locations
+                .iter()
+                .any(|location| location.root == home_skills)
+                && snapshot.locations.locations.iter().any(|location| {
+                    location
+                        .plugin
+                        .as_ref()
+                        .is_some_and(|plugin| plugin.id == "gmail@acme")
+                })
+            {
+                assert!(snapshot.sandbox_discoveries.lock().await.is_empty());
+                break anyhow::Ok(());
+            }
+        }
+    })
+    .await
+    .context("native watcher missed creation of a global root")??;
+    // Drive subsequent refreshes explicitly so native delivery cannot race cache assertions.
+    stop_listener(&manager).await?;
+    let refreshed = manager
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    let added_skill = home_skills.join("added/SKILL.md")?;
+    assert!(
+        refreshed
+            .skills
+            .iter()
+            .any(|skill| skill.path == added_skill)
+    );
+    assert_eq!(refreshed.plugins.len(), 1);
+    assert!(
+        refreshed
+            .skills
+            .iter()
+            .any(|skill| skill.path == demo_skill && skill.description == "ready")
+    );
+    let sandbox_params = DiscoverV2CapabilitiesRequest {
+        sandbox: Some(crate::FileSystemSandboxContext::from_legacy_sandbox_policy(
+            codex_protocol::protocol::SandboxPolicy::DangerFullAccess,
+            params.cwd.clone(),
+        )?),
+        ..params.clone()
+    };
+    manager
+        .get_or_refresh_discovery(request.clone(), sandbox_params)
+        .await?;
+    let retained = manager
+        .location_snapshot()
+        .await
+        .context("missing populated snapshot")?;
+    assert_eq!(retained.sandbox_discoveries.lock().await.len(), 2);
+    write(
+        &root,
+        "codex/plugins/cache/acme/gmail/2.0.0/.codex-plugin/plugin.json",
+        r#"{"name":"gmail"}"#,
+    )?;
+    write(
+        &root,
+        "codex/skills/demo/SKILL.md",
+        "---\nname: demo\ndescription: refreshed\n---\n",
+    )?;
+    manager.refresh_locations().await;
+    let next = manager
+        .location_snapshot()
+        .await
+        .context("missing refreshed snapshot")?;
+    assert_eq!(
+        next.locations
+            .locations
+            .iter()
+            .filter_map(|location| location.plugin.as_ref())
+            .find(|plugin| plugin.id == "gmail@acme")
+            .map(|plugin| plugin.version.as_str()),
+        Some("2.0.0")
+    );
+    assert!(next.sandbox_discoveries.lock().await.is_empty());
+    assert_eq!(retained.sandbox_discoveries.lock().await.len(), 2);
+    assert_eq!(
+        retained
+            .locations
+            .locations
+            .iter()
+            .filter_map(|location| location.plugin.as_ref())
+            .find(|plugin| plugin.id == "gmail@acme")
+            .map(|plugin| plugin.version.as_str()),
+        Some("1.0.0")
+    );
+    let after = manager
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    assert!(
+        after
+            .skills
+            .iter()
+            .any(|skill| skill.path == demo_skill && skill.description == "refreshed")
+    );
+    // Only the requested context is rebuilt; subsequent requests reuse it.
+    assert_eq!(next.sandbox_discoveries.lock().await.len(), 1);
+    assert!(Arc::ptr_eq(
+        &after,
+        &manager
+            .get_or_refresh_discovery(request.clone(), params.clone())
+            .await?
+    ));
+
+    #[cfg(unix)]
+    {
+        // A failed event refresh retains the response but retries without another event.
+        let cache = root.join("codex/plugins/cache");
+        let saved_cache = root.join("codex/plugins/saved-cache");
+        std::fs::rename(&cache, &saved_cache)?;
+        std::os::unix::fs::symlink(&cache, &cache)?;
+        manager.refresh_locations().await;
+        let fallback = manager
+            .get_or_refresh_discovery(request.clone(), params.clone())
+            .await?;
+        assert!(Arc::ptr_eq(&after, &fallback));
+        std::fs::remove_file(&cache)?;
+        std::fs::rename(&saved_cache, &cache)?;
+        write(
+            &root,
+            "codex/skills/recovered/SKILL.md",
+            "---\nname: recovered\ndescription: recovered after failed refresh\n---\n",
+        )?;
+        let recovered = manager
+            .get_or_refresh_discovery(request.clone(), params.clone())
+            .await?;
+        assert!(
+            recovered
+                .skills
+                .iter()
+                .any(|skill| skill.name == "recovered")
+        );
+        assert!(!Arc::ptr_eq(&fallback, &recovered));
+        assert!(Arc::ptr_eq(
+            &recovered,
+            &manager
+                .get_or_refresh_discovery(request.clone(), params.clone())
+                .await?
+        ));
+    }
+
+    // Without a subscription, requests reload both membership and content.
+    let mut uncached = CapabilityManager::new(LocalFileSystem::unsandboxed());
+    Arc::get_mut(&mut uncached)
+        .context("new manager unexpectedly shared")?
+        .watchers
+        .file_watcher = None;
+    let before = uncached
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    assert!(uncached.watchers.subscription.get().is_none());
+    write(
+        &root,
+        "codex/skills/demo/SKILL.md",
+        "---\nname: demo\ndescription: uncached edit\n---\n",
+    )?;
+    write(
+        &root,
+        "codex/plugins/cache/acme/calendar/1.0.0/.codex-plugin/plugin.json",
+        r#"{"name":"calendar"}"#,
+    )?;
+    let after = uncached
+        .get_or_refresh_discovery(request.clone(), params.clone())
+        .await?;
+    assert_eq!(after.plugins.len(), before.plugins.len() + 1);
+    assert!(
+        after
+            .plugins
+            .iter()
+            .any(|plugin| plugin.id == "calendar@acme")
+    );
+    assert!(
+        after
+            .skills
+            .iter()
+            .any(|skill| skill.path == demo_skill && skill.description == "uncached edit")
+    );
+    let repeated = uncached.get_or_refresh_discovery(request, params).await?;
+    assert_eq!(after, repeated);
+    assert!(!Arc::ptr_eq(&after, &repeated));
+    Ok(())
+}

codex-rs/exec-server/src/fs_helper.rs

@@ -9,11 +9,15 @@ use crate::CapabilityRootsDiscoverParams;
 use crate::CapabilityRootsDiscoverResponse;
 use crate::CopyOptions;
 use crate::CreateDirectoryOptions;
+use crate::DiscoverV2CapabilitiesResponse;
 use crate::ExecutorFileSystem;
 use crate::GetMetadataOptions;
+use crate::LocalFileSystem;
 use crate::ReadFileOptions;
 use crate::RemoveOptions;
 use crate::WriteFileOptions;
+use crate::discover_v2::capability_discoveries::load_capability_discovery_batch;
+use crate::discover_v2::capability_locations::CapabilityLocation;
 use crate::local_file_system::DirectFileSystem;
 use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD;
 use crate::protocol::FS_CANONICALIZE_METHOD;
@@ -54,6 +58,11 @@ pub const CODEX_FS_HELPER_ARG1: &str = "--codex-run-as-fs-helper";
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "operation", content = "params")]
 pub(crate) enum FsHelperRequest {
+    #[serde(rename = "capabilities/loadDiscoveries")]
+    LoadCapabilityDiscoveries {
+        locations: Vec<CapabilityLocation>,
+        warnings: Vec<String>,
+    },
     #[serde(rename = "capabilityRoots/discoverV1")]
     DiscoverCapabilityRoots(CapabilityRootsDiscoverParams),
     #[serde(rename = "fs/open")]
@@ -99,6 +108,8 @@ pub(crate) struct FsHelperOpenResponse {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "operation", content = "response")]
 pub(crate) enum FsHelperPayload {
+    #[serde(rename = "capabilities/loadDiscoveries")]
+    LoadCapabilityDiscoveries(DiscoverV2CapabilitiesResponse),
     #[serde(rename = "capabilityRoots/discoverV1")]
     DiscoverCapabilityRoots(CapabilityRootsDiscoverResponse),
     #[serde(rename = "fs/open")]
@@ -126,6 +137,7 @@ pub(crate) enum FsHelperPayload {
 impl FsHelperPayload {
     fn operation(&self) -> &'static str {
         match self {
+            Self::LoadCapabilityDiscoveries(_) => "capabilities/loadDiscoveries",
             Self::DiscoverCapabilityRoots(_) => CAPABILITY_ROOTS_DISCOVER_METHOD,
             Self::Open(_) => FS_OPEN_METHOD,
             Self::ReadFile(_) => FS_READ_FILE_METHOD,
@@ -140,6 +152,18 @@ impl FsHelperPayload {
         }
     }
 
+    pub(crate) fn expect_capability_discoveries(
+        self,
+    ) -> Result<DiscoverV2CapabilitiesResponse, JSONRPCErrorError> {
+        match self {
+            Self::LoadCapabilityDiscoveries(discovery) => Ok(discovery),
+            other => Err(unexpected_response(
+                "capabilities/loadDiscoveries",
+                other.operation(),
+            )),
+        }
+    }
+
     pub(crate) fn expect_capability_roots_discover(
         self,
     ) -> Result<CapabilityRootsDiscoverResponse, JSONRPCErrorError> {
@@ -243,6 +267,16 @@ pub(crate) async fn run_direct_request(
 ) -> Result<FsHelperPayload, JSONRPCErrorError> {
     let file_system = DirectFileSystem;
     match request {
+        FsHelperRequest::LoadCapabilityDiscoveries {
+            locations,
+            warnings,
+        } => {
+            let file_system = LocalFileSystem::unsandboxed();
+            let response = load_capability_discovery_batch(&file_system, &locations, warnings)
+                .await
+                .map_err(map_fs_error)?;
+            Ok(FsHelperPayload::LoadCapabilityDiscoveries(response))
+        }
         FsHelperRequest::DiscoverCapabilityRoots(params) => {
             let response = crate::discover_capability_roots(&file_system, params)
                 .await

codex-rs/exec-server/src/lib.rs

@@ -7,6 +7,8 @@ mod client_error;
 mod client_telemetry;
 mod client_transport;
 mod connection;
+#[path = "discoverV2/mod.rs"]
+mod discover_v2;
 mod environment;
 mod environment_bootstrap;
 mod environment_config;
@@ -137,12 +139,15 @@ pub use process::ExecProcessEventReceiver;
 pub use process::ExecProcessFuture;
 pub use process::StartedExecProcess;
 pub use protocol::ByteChunk;
+pub use protocol::CAPABILITIES_DISCOVER_V2_METHOD;
 pub use protocol::CAPABILITY_ROOTS_DISCOVER_METHOD;
 pub use protocol::CapabilityRootDiscoverRequest;
 pub use protocol::CapabilityRootDiscovery;
 pub use protocol::CapabilityRootsDiscoverParams;
 pub use protocol::CapabilityRootsDiscoverResponse;
 pub use protocol::CapabilityTextFile;
+pub use protocol::DiscoverV2CapabilitiesRequest;
+pub use protocol::DiscoverV2CapabilitiesResponse;
 pub use protocol::DiscoveredPluginFiles;
 pub use protocol::DiscoveredSkillFiles;
 pub use protocol::EnvironmentCapabilities;
@@ -164,6 +169,8 @@ pub use protocol::ExecResponse;
 pub use protocol::ExecServerNetworkPolicyDecision;
 pub use protocol::ExecServerNetworkPolicyRequest;
 pub use protocol::ExecServerNetworkProtocol;
+pub use protocol::ExecutorPlugin;
+pub use protocol::ExecutorSkill;
 pub use protocol::FsCanonicalizeParams;
 pub use protocol::FsCanonicalizeResponse;
 pub use protocol::FsCloseParams;

codex-rs/exec-server/src/sandboxed_file_system.rs

@@ -9,6 +9,7 @@ use crate::CapabilityRootsDiscoverParams;
 use crate::CapabilityRootsDiscoverResponse;
 use crate::CopyOptions;
 use crate::CreateDirectoryOptions;
+use crate::DiscoverV2CapabilitiesResponse;
 use crate::ExecServerRuntimePaths;
 use crate::ExecutorFileSystem;
 use crate::ExecutorFileSystemFuture;
@@ -24,6 +25,7 @@ use crate::RemoveOptions;
 use crate::WalkOptions;
 use crate::WalkOutcome;
 use crate::WriteFileOptions;
+use crate::discover_v2::capability_locations::CapabilityLocation;
 use crate::fs_helper::FsHelperPayload;
 use crate::fs_helper::FsHelperRequest;
 use crate::fs_sandbox::FileSystemSandboxRunner;
@@ -43,6 +45,25 @@ pub struct SandboxedFileSystem {
 }
 
 impl SandboxedFileSystem {
+    pub(crate) async fn load_sandboxed_capability_discoveries(
+        &self,
+        locations: Vec<CapabilityLocation>,
+        warnings: Vec<String>,
+        sandbox: &FileSystemSandboxContext,
+    ) -> FileSystemResult<DiscoverV2CapabilitiesResponse> {
+        require_platform_sandbox(Some(sandbox))?;
+        self.run_sandboxed(
+            sandbox,
+            FsHelperRequest::LoadCapabilityDiscoveries {
+                locations,
+                warnings,
+            },
+        )
+        .await?
+        .expect_capability_discoveries()
+        .map_err(map_sandbox_error)
+    }
+
     #[tracing::instrument(
         name = "capability_roots.discover_v1",
         skip_all,

codex-rs/exec-server/src/server/processor.rs

@@ -1,16 +1,21 @@
+use std::io;
 use std::sync::Arc;
 use std::time::Instant;
 
 use codex_build_info::BuildInfo;
 use codex_exec_server_protocol::JSONRPCMessage;
 use tokio::sync::mpsc;
+use tokio_util::task::AbortOnDropHandle;
 use tracing::debug;
 use tracing::warn;
 
 use crate::ExecServerRuntimePaths;
+use crate::LocalFileSystem;
 use crate::connection::CHANNEL_CAPACITY;
 use crate::connection::JsonRpcConnection;
 use crate::connection::JsonRpcConnectionEvent;
+use crate::discover_v2::capability_locations::CapabilityLocationRequest;
+use crate::discover_v2::capability_manager::CapabilityManager;
 use crate::rpc::RpcCallError;
 use crate::rpc::RpcNotificationSender;
 use crate::rpc::RpcServerOutboundMessage;
@@ -26,9 +31,12 @@ use crate::telemetry::ConnectionTransport;
 use crate::telemetry::ExecServerTelemetry;
 use crate::telemetry::ExecutorRegistration;
 use codex_http_client::HttpClientFactory;
+use codex_utils_path_uri::PathUri;
 
 #[derive(Clone)]
 pub(crate) struct ConnectionProcessor {
+    capability_manager: Arc<CapabilityManager>,
+    capability_prewarm: Arc<AbortOnDropHandle<()>>,
     session_registry: Arc<SessionRegistry>,
     runtime_paths: ExecServerRuntimePaths,
     telemetry: ExecServerTelemetry,
@@ -39,13 +47,17 @@ pub(crate) struct ConnectionProcessor {
 impl ConnectionProcessor {
     #[cfg(test)]
     pub(crate) fn new(runtime_paths: ExecServerRuntimePaths) -> Self {
-        Self::new_with_telemetry(
+        Self::new_with_location_request(
             runtime_paths,
             ExecServerTelemetry::default(),
             codex_http_client::HttpClientFactory::new(
                 codex_http_client::OutboundProxyPolicy::ReqwestDefault,
             ),
             RequestDispatchMode::Inline,
+            // Connection-only tests must not scan the real user home.
+            Err(io::Error::other(
+                "capability home paths not supplied by this test",
+            )),
         )
     }
 
@@ -54,11 +66,49 @@ impl ConnectionProcessor {
         telemetry: ExecServerTelemetry,
         http_client_factory: HttpClientFactory,
         request_dispatch_mode: RequestDispatchMode,
+    ) -> Self {
+        let request =
+            codex_utils_home_dir::find_codex_home().map(|codex_home| CapabilityLocationRequest {
+                codex_home: PathUri::from_abs_path(&codex_home),
+                user_home: dirs::home_dir()
+                    .and_then(|home| PathUri::from_host_native_path(home).ok()),
+            });
+        Self::new_with_location_request(
+            runtime_paths,
+            telemetry,
+            http_client_factory,
+            request_dispatch_mode,
+            request,
+        )
+    }
+
+    fn new_with_location_request(
+        runtime_paths: ExecServerRuntimePaths,
+        telemetry: ExecServerTelemetry,
+        http_client_factory: HttpClientFactory,
+        request_dispatch_mode: RequestDispatchMode,
+        request: io::Result<CapabilityLocationRequest>,
     ) -> Self {
         // Library callers may bypass CLI startup. Capture the version before serving clients.
         let _ = BuildInfo::get();
+        let capability_manager =
+            CapabilityManager::new(LocalFileSystem::with_runtime_paths(runtime_paths.clone()));
+        let manager = Arc::clone(&capability_manager);
+        // Start before any client connects; all sessions share these inert global locations.
+        let prewarm = tokio::spawn(async move {
+            // Home resolution failures are nonfatal, just like scan failures.
+            let result = match request {
+                Ok(request) => manager.prewarm_locations(request).await.map(|_| ()),
+                Err(error) => Err(error),
+            };
+            if let Err(error) = result {
+                tracing::warn!(error_kind = ?error.kind(), "capability location prewarming unavailable");
+            }
+        });
         Self {
             session_registry: SessionRegistry::new(telemetry.clone()),
+            capability_manager,
+            capability_prewarm: Arc::new(AbortOnDropHandle::new(prewarm)),
             runtime_paths,
             telemetry,
             http_client_factory,
@@ -95,6 +145,7 @@ impl ConnectionProcessor {
     }
 
     pub(crate) async fn shutdown(&self) {
+        self.capability_prewarm.abort();
         self.session_registry.shutdown().await;
     }
 }
@@ -106,6 +157,8 @@ async fn run_connection(
     executor_registration: Option<Arc<ExecutorRegistration>>,
 ) {
     let ConnectionProcessor {
+        capability_manager: _capability_manager,
+        capability_prewarm: _capability_prewarm,
         session_registry,
         runtime_paths,
         telemetry,
@@ -317,6 +370,69 @@ mod tests {
     use crate::rpc_server_requests::RpcServerRequestSender;
     use crate::server::session_registry::SessionRegistry;
 
+    #[tokio::test]
+    async fn startup_prewarms_without_a_client_connection() -> anyhow::Result<()> {
+        let paths = ExecServerRuntimePaths::new(
+            std::env::current_exe()?,
+            /*codex_linux_sandbox_exe*/ None,
+        )?;
+        let directory = tempfile::tempdir()?;
+        let request = crate::discover_v2::capability_locations::CapabilityLocationRequest {
+            codex_home: PathUri::from_host_native_path(directory.path().join("codex"))?,
+            user_home: Some(PathUri::from_host_native_path(
+                directory.path().join("home"),
+            )?),
+        };
+        let skill_root = directory.path().join("codex/skills");
+        std::fs::create_dir_all(&skill_root)?;
+        let mut processor = super::ConnectionProcessor::new_with_location_request(
+            paths,
+            crate::ExecServerTelemetry::default(),
+            codex_http_client::HttpClientFactory::new(
+                codex_http_client::OutboundProxyPolicy::ReqwestDefault,
+            ),
+            crate::server::RequestDispatchMode::Inline,
+            Ok(request.clone()),
+        );
+        Arc::get_mut(&mut processor.capability_prewarm)
+            .ok_or_else(|| anyhow::anyhow!("startup task unexpectedly shared"))?
+            .await?;
+        let cloned_processor = processor.clone();
+        assert!(Arc::ptr_eq(
+            &processor.capability_manager,
+            &cloned_processor.capability_manager,
+        ));
+        let manager = &processor.capability_manager;
+        // Removing the fixture proves startup retained it before any later call.
+        std::fs::remove_dir(&skill_root)?;
+        let (first, second) = tokio::try_join!(
+            manager.prewarm_locations(request.clone()),
+            manager.prewarm_locations(request.clone()),
+        )?;
+        assert!(std::ptr::eq(first, second));
+        assert_eq!(first.0, request);
+        let expected_root = request.codex_home.join("skills")?;
+        assert!(
+            first
+                .1
+                .locations
+                .iter()
+                .any(|location| location.root == expected_root)
+        );
+        let mut different_request = request.clone();
+        different_request.codex_home = request.codex_home.join("different")?;
+        let Err(error) = manager.prewarm_locations(different_request).await else {
+            anyhow::bail!("different home paths must not reuse cached locations");
+        };
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
+        assert!(std::ptr::eq(
+            first,
+            manager.prewarm_locations(request).await?,
+        ));
+        processor.shutdown().await;
+        Ok(())
+    }
+
     #[tokio::test]
     async fn connection_accepts_pipelined_scalar_requests() {
         let registry = SessionRegistry::new(crate::ExecServerTelemetry::default());

codex-rs/exec-server/tests/capability_discovery.rs

@@ -74,6 +74,10 @@ async fn discovers_a_complete_capability_bundle_in_one_request() -> anyhow::Resu
         &root.path().join("config/apps.json"),
         r#"{"apps":{"demo":{"connector_id":"connector-demo"}}}"#,
     )?;
+    write_file(
+        &root.path().join(".app.json"),
+        r#"{"apps":{"fallback":{"id":"connector-fallback"}}}"#,
+    )?;
     write_file(
         &root.path().join("skills/deploy/SKILL.md"),
         "---\nname: deploy\ndescription: Deploy the service.\n---\n\nDeploy instructions.\n",
@@ -156,6 +160,29 @@ async fn discovers_a_complete_capability_bundle_in_one_request() -> anyhow::Resu
         ]
     );
 
+    // Without an apps declaration, discovery returns the default .app.json.
+    let mut manifest: serde_json::Value = serde_json::from_str(&plugin.manifest.contents)?;
+    manifest
+        .as_object_mut()
+        .ok_or_else(|| anyhow::anyhow!("manifest must be an object"))?
+        .remove("apps");
+    write_file(
+        &root.path().join(".codex-plugin/plugin.json"),
+        &serde_json::to_string(&manifest)?,
+    )?;
+    let fallback = discover_root(&mut server, "demo@1", root_uri.clone()).await?;
+    assert_eq!(fallback.error, None);
+    assert_eq!(fallback.warnings, Vec::<String>::new());
+    let apps = fallback
+        .plugin
+        .and_then(|plugin| plugin.apps_config)
+        .ok_or_else(|| anyhow::anyhow!("missing fallback apps config"))?;
+    assert_eq!(apps.path, root_uri.join(".app.json")?);
+    assert_eq!(
+        apps.contents,
+        r#"{"apps":{"fallback":{"id":"connector-fallback"}}}"#,
+    );
+
     server.shutdown().await?;
     Ok(())
 }
@@ -186,6 +213,7 @@ async fn discovers_cursor_plugin_without_reading_default_mcp_for_inline_servers(
         root_uri.join(".cursor-plugin/plugin.json")?
     );
     assert_eq!(plugin.mcp_config, None);
+    assert_eq!(plugin.apps_config, None);
     assert_eq!(
         discovery
             .namespace_manifests
@@ -908,6 +936,113 @@ async fn executor_legacy_filesystem_cwd_keeps_absolute_read_allow_and_deny_rules
     Ok(())
 }
 
+// Exercise one helper request with multiple roots, including the response budget before IPC.
+#[tokio::test]
+async fn v2_helper_batches_roots_and_bounds_its_response() -> anyhow::Result<()> {
+    use tokio::io::AsyncWriteExt;
+
+    let workspace = tempfile::tempdir()?;
+    let mut locations = Vec::new();
+    // Five individually valid configs exceed the shared 4 MiB response budget.
+    let config = serde_json::to_string(&serde_json::json!({"padding": "x".repeat(900 * 1024)}))?;
+    for index in 0..5 {
+        let root = workspace.path().join(format!("plugin-{index}"));
+        write_file(
+            &root.join(".codex-plugin/plugin.json"),
+            r#"{"name":"demo"}"#,
+        )?;
+        write_file(&root.join(".mcp.json"), &config)?;
+        write_file(
+            &root.join("skills/deploy/SKILL.md"),
+            "---\nname: deploy\ndescription: Deploy.\n---\n",
+        )?;
+        locations.push(serde_json::json!({
+            "root": PathUri::from_host_native_path(root)?,
+            "plugin": {
+                "id": format!("plugin-{index}@test"),
+                "version": "local",
+                "remote_plugin_id": null,
+            },
+        }));
+    }
+    let small_root = workspace.path().join("plugin-small");
+    write_file(
+        &small_root.join(".codex-plugin/plugin.json"),
+        r#"{"name":"small"}"#,
+    )?;
+    write_file(
+        &small_root.join("skills/deploy/SKILL.md"),
+        "---\nname: deploy\ndescription: Deploy.\n---\n",
+    )?;
+    locations.push(serde_json::json!({
+        "root": PathUri::from_host_native_path(small_root)?,
+        "plugin": {
+            "id": "plugin-small@test",
+            "version": "local",
+            "remote_plugin_id": null,
+        },
+    }));
+    let request = serde_json::json!({
+        "operation": "capabilities/loadDiscoveries",
+        "params": {"locations": locations, "warnings": []},
+    });
+    let (helper_binary, _) = common::current_test_binary_helper_paths()?;
+    let mut child = tokio::process::Command::new(helper_binary)
+        .arg(codex_exec_server::CODEX_FS_HELPER_ARG1)
+        .stdin(std::process::Stdio::piped())
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .kill_on_drop(true)
+        .spawn()?;
+    let mut stdin = child
+        .stdin
+        .take()
+        .ok_or_else(|| anyhow::anyhow!("helper stdin missing"))?;
+    stdin.write_all(format!("{request}\n").as_bytes()).await?;
+    drop(stdin);
+    let output = tokio::time::timeout(
+        std::time::Duration::from_secs(/*secs*/ 30),
+        child.wait_with_output(),
+    )
+    .await??;
+    assert!(
+        output.status.success(),
+        "helper failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let wire: serde_json::Value = serde_json::from_slice(&output.stdout)?;
+    assert_eq!(wire["status"], "ok", "helper response: {wire}");
+    assert_eq!(wire["payload"]["operation"], "capabilities/loadDiscoveries");
+    let response: codex_exec_server_protocol::DiscoverV2CapabilitiesResponse =
+        serde_json::from_value(wire["payload"]["response"].clone())?;
+    assert_eq!(
+        response
+            .plugins
+            .iter()
+            .map(|plugin| plugin.id.as_str())
+            .collect::<Vec<_>>(),
+        vec![
+            "plugin-0@test",
+            "plugin-1@test",
+            "plugin-2@test",
+            "plugin-3@test",
+            "plugin-small@test",
+        ],
+    );
+    assert!(
+        response
+            .plugins
+            .iter()
+            .all(|plugin| plugin.skills.len() == 1)
+    );
+    assert_eq!(
+        response.warnings,
+        vec!["capability discovery size limit reached"]
+    );
+    assert!(serde_json::to_vec(&response)?.len() <= 4 * 1024 * 1024);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn sandboxed_discovery_batches_roots_without_combining_different_permissions()
 -> anyhow::Result<()> {

codex-rs/exec-server/tests/common/exec_server.rs

@@ -96,11 +96,18 @@ where
 }
 
 impl ExecServerHarness {
-    pub(crate) async fn start(mut command: Command) -> anyhow::Result<Self> {
+    pub(crate) async fn start(command: Command) -> anyhow::Result<Self> {
+        Self::start_with_stderr(command, Stdio::inherit()).await
+    }
+
+    pub(crate) async fn start_with_stderr(
+        mut command: Command,
+        stderr: Stdio,
+    ) -> anyhow::Result<Self> {
         let codex_home = TempDir::new()?;
         command.stdin(Stdio::null());
         command.stdout(Stdio::piped());
-        command.stderr(Stdio::inherit());
+        command.stderr(stderr);
         command.kill_on_drop(true);
         if !command
             .as_std()

codex-rs/exec-server/tests/common/mod.rs

@@ -165,6 +165,17 @@ fn maybe_run_exec_server_from_test_binary(guard: Option<&TestBinaryDispatchGuard
     if command != "exec-server" {
         return;
     }
+    // Enable requested child diagnostics so integration tests can observe background failures.
+    if env::var_os("RUST_LOG").is_some()
+        && let Err(error) = tracing_subscriber::fmt()
+            .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+            .with_writer(std::io::stderr)
+            .with_ansi(false)
+            .try_init()
+    {
+        eprintln!("failed to initialize executor test logging: {error}");
+        std::process::exit(1);
+    }
     // Initialize in the executor child, just as the real CLI does at startup.
     codex_build_info::BuildInfo::initialize(TEST_BUILD_COMMIT);
 

codex-rs/exec-server/tests/initialize.rs

@@ -101,6 +101,86 @@ async fn exec_server_accepts_initialize(version: Option<&str>) -> anyhow::Result
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn discovery_support_does_not_require_client_opt_in() -> anyhow::Result<()> {
+    let codex_home = TempDir::new()?;
+    #[cfg(not(windows))]
+    let discovery_home = {
+        // Fail the scan before startup, without racing the background task.
+        std::fs::write(codex_home.path().join("plugins"), "not a directory")?;
+        codex_home.path().to_path_buf()
+    };
+    // Windows treats the file-parent fixture as NotFound, which discovery skips.
+    // A missing explicit home instead makes home resolution fail deterministically.
+    #[cfg(windows)]
+    let discovery_home = codex_home.path().join("missing");
+    let user_home = codex_home.path().join("home");
+    std::fs::create_dir_all(user_home.join(".agents/skills"))?;
+    let stderr_path = codex_home.path().join("executor.log");
+    let stderr = std::fs::File::create(&stderr_path)?;
+    let helper_paths = common::exec_server::test_codex_helper_paths()?;
+    let mut command = Command::new(helper_paths.codex_exe);
+    command.args(["exec-server", "--listen", "ws://127.0.0.1:0"]);
+    command.env("CODEX_HOME", discovery_home);
+    command.env("HOME", &user_home);
+    command.env("USERPROFILE", &user_home);
+    command.env("RUST_LOG", "codex_exec_server=warn");
+    let mut server = ExecServerHarness::start_with_stderr(command, stderr.into()).await?;
+    let observed_failure = tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            let logs = tokio::fs::read_to_string(&stderr_path).await?;
+            if logs.contains("capability location prewarming unavailable") {
+                return Ok::<_, std::io::Error>(());
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await;
+    anyhow::ensure!(
+        observed_failure.is_ok(),
+        "prewarm failure was not observed; executor stderr:\n{}",
+        std::fs::read_to_string(&stderr_path)?
+    );
+    observed_failure??;
+    let initialize_id = server
+        .send_request(
+            "initialize",
+            serde_json::to_value(InitializeParams {
+                client_name: "exec-server-test".to_string(),
+                resume_session_id: None,
+            })?,
+        )
+        .await?;
+
+    let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = server.next_event().await?
+    else {
+        anyhow::bail!("expected initialize response without V2 opt-in");
+    };
+    assert_eq!(id, initialize_id);
+    let response: InitializeResponse = serde_json::from_value(result)?;
+    assert!(
+        response
+            .environment_info
+            .context("initialize metadata missing")?
+            .capabilities
+            .capability_discovery_v2
+    );
+    server
+        .send_notification("initialized", serde_json::json!({}))
+        .await?;
+
+    let environment_info_id = server
+        .send_request("environment/info", serde_json::json!({}))
+        .await?;
+    let JSONRPCMessage::Response(JSONRPCResponse { id, .. }) = server.next_event().await? else {
+        panic!("expected environment info response after degraded discovery startup");
+    };
+    assert_eq!(id, environment_info_id);
+
+    server.shutdown().await?;
+    Ok(())
+}
+
 /// Requests retain their wire-order initialization errors even when later handshake messages are pipelined.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn exec_server_rejects_pipelined_requests_before_initialized() -> anyhow::Result<()> {

codex-rs/plugin/Cargo.toml

@@ -14,6 +14,7 @@ workspace = true
 
 [dependencies]
 codex-config = { workspace = true }
+codex-core-plugin-common = { workspace = true }
 codex-protocol = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-path-uri = { workspace = true }

codex-rs/plugin/src/lib.rs

@@ -7,7 +7,7 @@ pub use codex_utils_plugins::mention_syntax;
 mod bundled_hooks;
 mod load_outcome;
 pub mod manifest;
-mod plugin_id;
+use codex_core_plugin_common::plugin_id;
 mod provider;
 
 pub use bundled_hooks::is_allowlisted_bundled_cleanup_hook;