diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index d23a71ba17..2fd33cc85e 100644 --- a/codex-rs/Cargo.lock +++ b/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", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index d2fff95cda..4b10699320 100644 --- a/codex-rs/Cargo.toml +++ b/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" } diff --git a/codex-rs/core-plugin-common/BUILD.bazel b/codex-rs/core-plugin-common/BUILD.bazel new file mode 100644 index 0000000000..769b65aba1 --- /dev/null +++ b/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", +) diff --git a/codex-rs/core-plugin-common/Cargo.toml b/codex-rs/core-plugin-common/Cargo.toml new file mode 100644 index 0000000000..654faa8af1 --- /dev/null +++ b/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 } diff --git a/codex-rs/core-plugin-common/src/installed.rs b/codex-rs/core-plugin-common/src/installed.rs new file mode 100644 index 0000000000..5a2d3a0f61 --- /dev/null +++ b/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 { + 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::>(); + 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 { + 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()) +} diff --git a/codex-rs/core-plugin-common/src/lib.rs b/codex-rs/core-plugin-common/src/lib.rs new file mode 100644 index 0000000000..4f5c0bdea9 --- /dev/null +++ b/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; diff --git a/codex-rs/plugin/src/plugin_id.rs b/codex-rs/core-plugin-common/src/plugin_id.rs similarity index 100% rename from codex-rs/plugin/src/plugin_id.rs rename to codex-rs/core-plugin-common/src/plugin_id.rs diff --git a/codex-rs/plugin/src/plugin_id_tests.rs b/codex-rs/core-plugin-common/src/plugin_id_tests.rs similarity index 100% rename from codex-rs/plugin/src/plugin_id_tests.rs rename to codex-rs/core-plugin-common/src/plugin_id_tests.rs diff --git a/codex-rs/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index f5e5160707..af58b8e9e8 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/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 } diff --git a/codex-rs/core-plugins/src/store.rs b/codex-rs/core-plugins/src/store.rs index 4f37bbef9c..4a52d58a81 100644 --- a/codex-rs/core-plugins/src/store.rs +++ b/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 { - 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::>(); - 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 { @@ -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> { diff --git a/codex-rs/exec-server-protocol/src/capabilities.rs b/codex-rs/exec-server-protocol/src/capabilities.rs new file mode 100644 index 0000000000..2583b9f25f --- /dev/null +++ b/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, +} + +/// 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, + /// Skills outside installed packages. + pub skills: Vec, + #[serde(default)] + pub warnings: Vec, +} + +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExecutorPlugin { + /// Environment-local ID: `@`. + pub id: String, + pub remote_plugin_id: Option, + pub version: String, + pub root: PathUri, + pub files: DiscoveredPluginFiles, + pub skills: Vec, +} + +/// 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, + pub description: String, + #[serde(default)] + pub short_description: Option, + /// Path to SKILL.md, not its containing directory. + pub path: PathUri, + pub metadata: Option, +} + +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; diff --git a/codex-rs/exec-server-protocol/src/capabilities_tests.rs b/codex-rs/exec-server-protocol/src/capabilities_tests.rs new file mode 100644 index 0000000000..d2cdab71b1 --- /dev/null +++ b/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> { + let request = DiscoverV2CapabilitiesRequest { + cwd: PathUri::parse("file:///workspace")?, + sandbox: None, + }; + let wire = serde_json::to_value(&request)?; + assert_eq!( + serde_json::from_value::(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::(wire)?, + response + ); + Ok(()) +} diff --git a/codex-rs/exec-server-protocol/src/lib.rs b/codex-rs/exec-server-protocol/src/lib.rs index bdfaa3d145..e9eb3aee31 100644 --- a/codex-rs/exec-server-protocol/src/lib.rs +++ b/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; diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index fc9060f564..8f7385d296 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/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::(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, diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index 7795c4fd8d..c390da762f 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/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 } diff --git a/codex-rs/exec-server/src/capability_discovery.rs b/codex-rs/exec-server/src/capability_discovery.rs index a32d84f87a..2b071884c9 100644 --- a/codex-rs/exec-server/src/capability_discovery.rs +++ b/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, diff --git a/codex-rs/exec-server/src/discoverV2/capability_discoveries.rs b/codex-rs/exec-server/src/discoverV2/capability_discoveries.rs new file mode 100644 index 0000000000..bda6f20e67 --- /dev/null +++ b/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), + Skills(Vec), +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct CapabilityDiscovery { + content: DiscoveredSource, + warnings: Vec, +} + +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> { + 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, +) -> io::Result { + 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 { + // 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::(&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::>().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 { + 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; diff --git a/codex-rs/exec-server/src/discoverV2/capability_locations.rs b/codex-rs/exec-server/src/discoverV2/capability_locations.rs new file mode 100644 index 0000000000..04f6a373e0 --- /dev/null +++ b/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, +} + +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, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct InstalledPlugin { + pub id: String, + pub version: String, + pub remote_plugin_id: Option, +} + +/// Selected locations; membership does not authorize capability reads. +#[derive(Clone, Default)] +pub(crate) struct CapabilityLocations { + pub locations: Vec, + pub watches: Vec, + pub warnings: Vec, +} + +/// 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 { + 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 { + 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::>(); + 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::>() + }) + .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> { + // 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; diff --git a/codex-rs/exec-server/src/discoverV2/capability_locations/candidates.rs b/codex-rs/exec-server/src/discoverV2/capability_locations/candidates.rs new file mode 100644 index 0000000000..332baeb3a2 --- /dev/null +++ b/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, + watches: Vec, +} + +/// 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, + skill_roots: Vec, + ) -> io::Result { + let mut candidates = cache_root + .into_iter() + .map(CandidateCapabilityLocation::Cache) + .collect::>(); + 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 { + 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 { + 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 + } +} diff --git a/codex-rs/exec-server/src/discoverV2/capability_manager.rs b/codex-rs/exec-server/src/discoverV2/capability_manager.rs new file mode 100644 index 0000000000..2f3e37de57 --- /dev/null +++ b/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>>, + location_refresh_lock: Semaphore, + watchers: CapabilityWatchers, +} + +struct LocationState { + dirty: AtomicBool, + locations: CapabilityLocations, + sandbox_discoveries: Mutex>>, + _registration: WatchRegistration, +} + +struct CachedSandboxDiscovery { + sandbox: Option, + response: OnceCell>, +} + +impl CapabilityManager { + pub(crate) fn new(file_system: LocalFileSystem) -> Arc { + 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, + 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, + location_request: CapabilityLocationRequest, + params: DiscoverV2CapabilitiesRequest, + ) -> io::Result> { + 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> { + 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> { + self.current_state.lock().await.clone() + } +} + +#[cfg(test)] +#[path = "tests/capability_manager_tests.rs"] +mod tests; diff --git a/codex-rs/exec-server/src/discoverV2/capability_manager/capability_watchers.rs b/codex-rs/exec-server/src/discoverV2/capability_manager/capability_watchers.rs new file mode 100644 index 0000000000..be92ca91a7 --- /dev/null +++ b/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>, + subscription: OnceLock, +} + +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, 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; diff --git a/codex-rs/exec-server/src/discoverV2/mod.rs b/codex-rs/exec-server/src/discoverV2/mod.rs new file mode 100644 index 0000000000..faab1775e6 --- /dev/null +++ b/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; diff --git a/codex-rs/exec-server/src/discoverV2/tests/capability_discoveries_tests.rs b/codex-rs/exec-server/src/discoverV2/tests/capability_discoveries_tests.rs new file mode 100644 index 0000000000..23f2a267ae --- /dev/null +++ b/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!["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::>>()? + ); + 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(()) +} diff --git a/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs b/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs new file mode 100644 index 0000000000..5f6ce6b8fa --- /dev/null +++ b/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 { + 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(()) +} diff --git a/codex-rs/exec-server/src/discoverV2/tests/capability_manager_tests.rs b/codex-rs/exec-server/src/discoverV2/tests/capability_manager_tests.rs new file mode 100644 index 0000000000..ca300006e1 --- /dev/null +++ b/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, +)> { + 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!["recovered"] + ); + assert!(Arc::ptr_eq( + &recovered, + &manager + .get_or_refresh_discovery(request.clone(), params) + .await? + )); + Ok(()) + }) +} diff --git a/codex-rs/exec-server/src/discoverV2/tests/capability_watchers_tests.rs b/codex-rs/exec-server/src/discoverV2/tests/capability_watchers_tests.rs new file mode 100644 index 0000000000..4f3484bf6a --- /dev/null +++ b/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(()) +} diff --git a/codex-rs/exec-server/src/fs_helper.rs b/codex-rs/exec-server/src/fs_helper.rs index ad6eacf93c..31ab576d73 100644 --- a/codex-rs/exec-server/src/fs_helper.rs +++ b/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, + warnings: Vec, + }, #[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 { + match self { + Self::LoadCapabilityDiscoveries(discovery) => Ok(discovery), + other => Err(unexpected_response( + "capabilities/loadDiscoveries", + other.operation(), + )), + } + } + pub(crate) fn expect_capability_roots_discover( self, ) -> Result { @@ -243,6 +267,16 @@ pub(crate) async fn run_direct_request( ) -> Result { 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 diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index ddee602bb4..be0b6f31fe 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/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; diff --git a/codex-rs/exec-server/src/sandboxed_file_system.rs b/codex-rs/exec-server/src/sandboxed_file_system.rs index 0d3e90fa05..92b857676f 100644 --- a/codex-rs/exec-server/src/sandboxed_file_system.rs +++ b/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, + warnings: Vec, + sandbox: &FileSystemSandboxContext, + ) -> FileSystemResult { + 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, diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 1fd2a1f1d6..289551b389 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/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, + capability_prewarm: Arc>, session_registry: Arc, 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, ) -> 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>, ) { 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()); diff --git a/codex-rs/exec-server/tests/capability_discovery.rs b/codex-rs/exec-server/tests/capability_discovery.rs index e811bfbbe5..9044555e0d 100644 --- a/codex-rs/exec-server/tests/capability_discovery.rs +++ b/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::::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![ + "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<()> { diff --git a/codex-rs/exec-server/tests/common/exec_server.rs b/codex-rs/exec-server/tests/common/exec_server.rs index d0fc3892fc..51c99f1ffa 100644 --- a/codex-rs/exec-server/tests/common/exec_server.rs +++ b/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 { + pub(crate) async fn start(command: Command) -> anyhow::Result { + Self::start_with_stderr(command, Stdio::inherit()).await + } + + pub(crate) async fn start_with_stderr( + mut command: Command, + stderr: Stdio, + ) -> anyhow::Result { 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() diff --git a/codex-rs/exec-server/tests/common/mod.rs b/codex-rs/exec-server/tests/common/mod.rs index 427c11f47d..569f310337 100644 --- a/codex-rs/exec-server/tests/common/mod.rs +++ b/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); diff --git a/codex-rs/exec-server/tests/initialize.rs b/codex-rs/exec-server/tests/initialize.rs index 8300f62164..435ebcc8a0 100644 --- a/codex-rs/exec-server/tests/initialize.rs +++ b/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<()> { diff --git a/codex-rs/plugin/Cargo.toml b/codex-rs/plugin/Cargo.toml index 89e2bc2fc9..462fb5e189 100644 --- a/codex-rs/plugin/Cargo.toml +++ b/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 } diff --git a/codex-rs/plugin/src/lib.rs b/codex-rs/plugin/src/lib.rs index ed81bb5e92..d623c76466 100644 --- a/codex-rs/plugin/src/lib.rs +++ b/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;