From 08e30a2e4e38187f38f59aa4778c9213b56ff00e Mon Sep 17 00:00:00 2001 From: Bryan Ashley Date: Fri, 17 Jul 2026 01:18:22 +0000 Subject: [PATCH] Add batched executor capability discovery (#33852) ## Why Selected capability roots can contribute plugins, MCP servers, connectors, and skills. Discovering each contribution separately requires repeated access to the executor filesystem. ## What changed - Add the `capabilityRoots/discoverV1` exec-server RPC to scan selected roots and materialize recognized plugin manifests, configuration files, skill instructions, and skill metadata in one bounded request. - Add the opt-in `executor_capability_discovery` feature, with a thread-scoped cache and per-step snapshot shared by MCP and skill discovery. - Parse MCP, connector, and skill contributions from the materialized snapshot, including serving cached skill instructions without another filesystem read. ## Testing - Cover discovery limits, manifest precedence, root-local failures, cache reuse, plugin contributions, and parity with the existing environment skill loader. GitOrigin-RevId: f98fd2321cafb58c596db02da1f83c09d8eb375d --- codex-rs/Cargo.lock | 3 + .../suite/v2/selected_capability_stack.rs | 2 +- codex-rs/core-plugins/src/manifest.rs | 4 +- codex-rs/core-skills/src/loader.rs | 3 + .../core-skills/src/loader/environment.rs | 139 ++++++ .../core-skills/tests/environment_loader.rs | 141 ++++++ codex-rs/core/config.schema.json | 6 + codex-rs/core/src/mcp.rs | 3 + codex-rs/core/src/session/mcp.rs | 50 +- codex-rs/core/src/session/mod.rs | 10 + codex-rs/core/src/session/session.rs | 1 + codex-rs/core/src/session/step_context.rs | 5 + codex-rs/core/src/session/tests.rs | 8 +- codex-rs/core/src/session/world_state.rs | 3 + codex-rs/core/src/thread_manager_tests.rs | 2 + codex-rs/core/src/tools/spec_plan_tests.rs | 1 + codex-rs/exec-server-protocol/src/protocol.rs | 123 +++++ .../exec-server/src/capability_discovery.rs | 462 ++++++++++++++++++ .../src/capability_discovery_cache.rs | 197 ++++++++ codex-rs/exec-server/src/client.rs | 10 + codex-rs/exec-server/src/environment.rs | 15 + codex-rs/exec-server/src/lib.rs | 14 + .../src/server/file_system_handler.rs | 11 + codex-rs/exec-server/src/server/handler.rs | 10 + codex-rs/exec-server/src/server/registry.rs | 8 + .../exec-server/tests/capability_discovery.rs | 229 +++++++++ codex-rs/ext/extension-api/Cargo.toml | 1 + .../ext/extension-api/src/contributors/mcp.rs | 11 + .../src/contributors/world_state.rs | 3 + codex-rs/ext/mcp/Cargo.toml | 1 + codex-rs/ext/mcp/src/executor_plugin.rs | 89 +++- .../ext/mcp/src/executor_plugin/discovery.rs | 154 ++++++ codex-rs/ext/mcp/src/lib_tests.rs | 1 + codex-rs/ext/mcp/tests/executor_plugin_mcp.rs | 50 +- codex-rs/ext/skills/src/catalog.rs | 25 + codex-rs/ext/skills/src/extension.rs | 3 + codex-rs/ext/skills/src/provider.rs | 3 + codex-rs/ext/skills/src/provider/executor.rs | 69 ++- codex-rs/ext/skills/src/state.rs | 42 ++ codex-rs/ext/skills/src/tools/mod.rs | 1 + .../tests/executor_file_system_authority.rs | 54 ++ codex-rs/ext/skills/tests/skills_extension.rs | 5 + codex-rs/features/src/lib.rs | 8 + codex-rs/features/src/tests.rs | 13 + codex-rs/utils/plugins/Cargo.toml | 1 + codex-rs/utils/plugins/src/lib.rs | 2 +- .../utils/plugins/src/plugin_namespace.rs | 8 +- 47 files changed, 1957 insertions(+), 47 deletions(-) create mode 100644 codex-rs/exec-server/src/capability_discovery.rs create mode 100644 codex-rs/exec-server/src/capability_discovery_cache.rs create mode 100644 codex-rs/exec-server/tests/capability_discovery.rs create mode 100644 codex-rs/ext/mcp/src/executor_plugin/discovery.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 25d76433675a..2e045fb2b848 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2985,6 +2985,7 @@ version = "0.0.0" dependencies = [ "codex-config", "codex-context-fragments", + "codex-exec-server-protocol", "codex-protocol", "codex-tools", "codex-utils-absolute-path", @@ -3388,6 +3389,7 @@ name = "codex-mcp-extension" version = "0.0.0" dependencies = [ "codex-config", + "codex-connectors", "codex-connectors-extension", "codex-core", "codex-core-plugins", @@ -4349,6 +4351,7 @@ name = "codex-utils-plugins" version = "0.0.0" dependencies = [ "codex-exec-server", + "codex-exec-server-protocol", "codex-utils-absolute-path", "codex-utils-path-uri", "serde", diff --git a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs index 20ec96481db1..c104313a1755 100644 --- a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs +++ b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs @@ -462,7 +462,7 @@ fn selected_capability_fixture( std::fs::write( config_path, format!( - "{config}\n[features]\napps = true\ndeferred_executor = true\n\n[skills]\ninclude_instructions = true\n" + "{config}\n[features]\napps = true\ndeferred_executor = true\nexecutor_capability_discovery = true\n\n[skills]\ninclude_instructions = true\n" ), )?; write_chatgpt_auth( diff --git a/codex-rs/core-plugins/src/manifest.rs b/codex-rs/core-plugins/src/manifest.rs index ff6d2c902a1a..4cc50328e9a3 100644 --- a/codex-rs/core-plugins/src/manifest.rs +++ b/codex-rs/core-plugins/src/manifest.rs @@ -19,7 +19,7 @@ pub type PluginManifestMcpServers = codex_plugin::manifest::PluginManifestMcpServers; pub type PluginManifestPaths = codex_plugin::manifest::PluginManifestPaths; -pub(crate) type UriPluginManifest = codex_plugin::manifest::PluginManifest; +pub type UriPluginManifest = codex_plugin::manifest::PluginManifest; #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] @@ -178,7 +178,7 @@ pub(crate) fn parse_plugin_manifest( .try_map_resources(|path| path.to_abs_path().map_err(serde_json::Error::io)) } -pub(crate) fn parse_plugin_manifest_uri( +pub fn parse_plugin_manifest_uri( plugin_root: &PathUri, manifest_path: &PathUri, contents: &str, diff --git a/codex-rs/core-skills/src/loader.rs b/codex-rs/core-skills/src/loader.rs index d03e4fc12d6e..c95a688cf66b 100644 --- a/codex-rs/core-skills/src/loader.rs +++ b/codex-rs/core-skills/src/loader.rs @@ -4,6 +4,9 @@ mod namespace; pub use environment::EnvironmentSkillLoadOutcome; pub use environment::EnvironmentSkillMetadata; +pub use environment::EnvironmentSkillSnapshot; +pub use environment::EnvironmentSkillSnapshotOutcome; +pub use environment::load_environment_skills_from_discovery; pub use environment::load_environment_skills_from_root; use crate::model::SkillDependencies; diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index 052f29265133..5bef5481f77b 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -1,5 +1,7 @@ +use std::collections::HashMap; use std::io; +use codex_exec_server::CapabilityRootDiscovery; use codex_exec_server::ExecutorFileSystem; use codex_protocol::protocol::Product; use codex_utils_path_uri::PathUri; @@ -45,6 +47,19 @@ pub struct EnvironmentSkillMetadata { pub policy: Option, } +/// Parsed executor skill plus the instructions already materialized by capability discovery. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct EnvironmentSkillSnapshot { + pub metadata: EnvironmentSkillMetadata, + pub instructions: String, +} + +#[derive(Debug, Default)] +pub struct EnvironmentSkillSnapshotOutcome { + pub skills: Vec, + pub warnings: Vec, +} + impl EnvironmentSkillMetadata { pub fn allows_implicit_invocation(&self) -> bool { self.policy @@ -212,6 +227,130 @@ pub async fn load_environment_skills_from_root( outcome } +/// Parses an executor-produced manifest bundle without issuing additional filesystem requests. +pub fn load_environment_skills_from_discovery( + discovery: &CapabilityRootDiscovery, + restriction_product: Option, +) -> EnvironmentSkillSnapshotOutcome { + let mut outcome = EnvironmentSkillSnapshotOutcome { + warnings: discovery.warnings.clone(), + ..Default::default() + }; + if let Some(error) = &discovery.error { + outcome.warnings.push(error.clone()); + return outcome; + } + + let mut plugin_namespaces = HashMap::new(); + for (plugin_root, name) in discovery.namespace_manifests.iter().filter_map(|manifest| { + #[derive(serde::Deserialize)] + struct ManifestName { + #[serde(default)] + name: String, + } + + let plugin_root = manifest.path.parent()?.parent()?; + let parsed = serde_json::from_str::(&manifest.contents).ok()?; + let name = if parsed.name.trim().is_empty() { + plugin_root.basename()? + } else { + parsed.name + }; + Some((plugin_root, name)) + }) { + // Exec-server orders manifests by the same precedence as local discovery. Preserve the + // first manifest if an older or alternate server returns duplicates for one plugin root. + plugin_namespaces.entry(plugin_root).or_insert(name); + } + + for skill in &discovery.skills { + let plugin_namespace = + nearest_plugin_namespace(&skill.instructions.path, &plugin_namespaces); + let ParsedSkillFrontmatter { + name: base_name, + description, + short_description, + } = match parse_skill_frontmatter_metadata_inner(&skill.instructions.contents, || { + default_skill_name(&skill.instructions.path) + }) { + Ok(frontmatter) => frontmatter, + Err(error) => { + outcome.warnings.push(format!( + "Failed to load environment skill at {}: {error}", + skill.instructions.path + )); + continue; + } + }; + let name = plugin_namespace + .map(|namespace| format!("{namespace}:{base_name}")) + .unwrap_or(base_name); + if let Err(error) = validate_len(&name, MAX_QUALIFIED_NAME_LEN, "qualified name") { + outcome.warnings.push(format!( + "Failed to load environment skill at {}: {error}", + skill.instructions.path + )); + continue; + } + let (dependencies, policy) = skill + .metadata + .as_ref() + .and_then(|metadata| { + serde_yaml::from_str::(&metadata.contents) + .map_err(|error| { + tracing::warn!( + path = %metadata.path, + "ignoring invalid discovered skill metadata: {error}" + ); + }) + .ok() + }) + .map(|metadata| { + ( + resolve_dependencies(metadata.dependencies), + resolve_policy(metadata.policy), + ) + }) + .unwrap_or((None, None)); + let metadata = EnvironmentSkillMetadata { + path_to_skills_md: skill.instructions.path.clone(), + name, + description, + short_description, + dependencies, + policy, + }; + if metadata.matches_product_restriction(restriction_product) { + outcome.skills.push(EnvironmentSkillSnapshot { + metadata, + instructions: skill.instructions.contents.clone(), + }); + } + } + outcome.skills.sort_by(|left, right| { + left.metadata.name.cmp(&right.metadata.name).then_with(|| { + left.metadata + .path_to_skills_md + .to_string() + .cmp(&right.metadata.path_to_skills_md.to_string()) + }) + }); + outcome +} + +fn nearest_plugin_namespace<'a>( + skill_path: &PathUri, + plugin_namespaces: &'a HashMap, +) -> Option<&'a str> { + let mut ancestor = skill_path.parent(); + while let Some(path) = ancestor { + if let Some(namespace) = plugin_namespaces.get(&path) { + return Some(namespace); + } + ancestor = path.parent(); + } + None +} async fn read_skill_contents( file_system: &dyn ExecutorFileSystem, skill_path: &PathUri, diff --git a/codex-rs/core-skills/tests/environment_loader.rs b/codex-rs/core-skills/tests/environment_loader.rs index aef4fb0af0be..2482f7ce2985 100644 --- a/codex-rs/core-skills/tests/environment_loader.rs +++ b/codex-rs/core-skills/tests/environment_loader.rs @@ -6,7 +6,10 @@ use std::sync::atomic::Ordering; use std::time::Duration; use codex_core_skills::loader::EnvironmentSkillMetadata; +use codex_core_skills::loader::load_environment_skills_from_discovery; use codex_core_skills::loader::load_environment_skills_from_root; +use codex_exec_server::CapabilityRootDiscoverRequest; +use codex_exec_server::CapabilityRootsDiscoverParams; use codex_exec_server::CopyOptions; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::ExecutorFileSystem; @@ -19,6 +22,7 @@ use codex_exec_server::ReadDirectoryEntry; use codex_exec_server::RemoveOptions; use codex_exec_server::WalkOptions; use codex_exec_server::WalkOutcome; +use codex_exec_server::discover_capability_roots; use codex_utils_path_uri::PathUri; use pretty_assertions::assert_eq; use tempfile::tempdir; @@ -540,3 +544,140 @@ async fn host_loading_reuses_walk_inventory_for_symlinked_skill_pack() { 1 ); } + +#[tokio::test] +async fn executor_bundle_parser_matches_the_existing_environment_loader() { + let root = tempdir().expect("tempdir"); + let plugin_manifest = root.path().join(".codex-plugin/plugin.json"); + let nested_manifest = root.path().join("nested/.claude-plugin/plugin.json"); + let deploy_skill = root.path().join("skills/deploy/SKILL.md"); + let deploy_metadata = root.path().join("skills/deploy/agents/openai.yaml"); + let audit_skill = root.path().join("nested/skills/audit/SKILL.md"); + for (path, contents) in [ + (&plugin_manifest, r#"{"name":"demo"}"#), + (&nested_manifest, r#"{"name":"nested"}"#), + ( + &deploy_skill, + "---\nname: deploy\ndescription: Deploy the service.\n---\n\nDeploy.\n", + ), + ( + &deploy_metadata, + "policy:\n allow_implicit_invocation: false\n", + ), + ( + &audit_skill, + "---\nname: audit\ndescription: Audit the service.\n---\n\nAudit.\n", + ), + ] { + fs::create_dir_all(path.parent().expect("test file parent")).expect("test directory"); + fs::write(path, contents).expect("test file"); + } + + let root_uri = PathUri::from_host_native_path(root.path()).expect("root URI"); + let existing = load_environment_skills_from_root( + LOCAL_FS.as_ref(), + &root_uri, + /*restriction_product*/ None, + ) + .await; + let response = discover_capability_roots( + LOCAL_FS.as_ref(), + CapabilityRootsDiscoverParams { + roots: vec![CapabilityRootDiscoverRequest { + id: "demo@1".to_string(), + path: root_uri, + }], + }, + ) + .await + .expect("capability discovery"); + let bundled = load_environment_skills_from_discovery( + response.roots.first().expect("discovered root"), + /*restriction_product*/ None, + ); + + assert_eq!(bundled.warnings, existing.warnings); + assert_eq!( + bundled + .skills + .iter() + .map(|skill| skill.metadata.clone()) + .collect::>(), + existing.skills + ); + assert_eq!( + bundled + .skills + .iter() + .map(|skill| skill.instructions.as_str()) + .collect::>(), + vec![ + "---\nname: deploy\ndescription: Deploy the service.\n---\n\nDeploy.\n", + "---\nname: audit\ndescription: Audit the service.\n---\n\nAudit.\n", + ] + ); +} + +#[tokio::test] +async fn executor_bundle_preserves_parent_namespace_and_manifest_precedence() { + let plugin = tempdir().expect("tempdir"); + for (relative_path, name) in [ + (".codex-plugin/plugin.json", "codex-name"), + (".claude-plugin/plugin.json", "claude-name"), + (".cursor-plugin/plugin.json", "cursor-name"), + ] { + let manifest = plugin.path().join(relative_path); + fs::create_dir_all(manifest.parent().expect("manifest parent")) + .expect("manifest directory"); + fs::write(&manifest, format!(r#"{{"name":"{name}"}}"#)).expect("manifest"); + } + let skills_root = plugin.path().join("skills"); + let skill_path = skills_root.join("search/SKILL.md"); + fs::create_dir_all(skill_path.parent().expect("skill parent")).expect("skill directory"); + fs::write( + &skill_path, + "---\nname: search\ndescription: Search the project.\n---\n\nSearch.\n", + ) + .expect("skill"); + + let root_uri = PathUri::from_host_native_path(&skills_root).expect("skills root URI"); + let existing = load_environment_skills_from_root( + LOCAL_FS.as_ref(), + &root_uri, + /*restriction_product*/ None, + ) + .await; + let response = discover_capability_roots( + LOCAL_FS.as_ref(), + CapabilityRootsDiscoverParams { + roots: vec![CapabilityRootDiscoverRequest { + id: "skills-only".to_string(), + path: root_uri, + }], + }, + ) + .await + .expect("capability discovery"); + let discovery = response.roots.first().expect("discovered root"); + let bundled = + load_environment_skills_from_discovery(discovery, /*restriction_product*/ None); + + assert_eq!(discovery.plugin, None); + assert_eq!(discovery.namespace_manifests.len(), 1); + assert!( + discovery.namespace_manifests[0] + .path + .to_string() + .ends_with("/.codex-plugin/plugin.json") + ); + assert_eq!(bundled.warnings, existing.warnings); + assert_eq!( + bundled + .skills + .iter() + .map(|skill| skill.metadata.clone()) + .collect::>(), + existing.skills + ); + assert_eq!(bundled.skills[0].metadata.name, "codex-name:search"); +} diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 0666e065c6c8..2fa453666ccb 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -518,6 +518,9 @@ "exec_permission_approvals": { "type": "boolean" }, + "executor_capability_discovery": { + "type": "boolean" + }, "experimental_use_unified_exec_tool": { "type": "boolean" }, @@ -4912,6 +4915,9 @@ "exec_permission_approvals": { "type": "boolean" }, + "executor_capability_discovery": { + "type": "boolean" + }, "experimental_use_unified_exec_tool": { "type": "boolean" }, diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index b33111373a5e..d5e6fb1efcf7 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -7,6 +7,7 @@ use codex_connectors::ConnectorRuntimeManager; use codex_connectors::ConnectorSnapshot; use codex_connectors::PluginConnectorSource; use codex_core_plugins::PluginsManager; +use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; @@ -109,6 +110,7 @@ impl McpManager { thread_store: &ExtensionData, originator: &str, ready_selected_capability_roots: &[SelectedCapabilityRoot], + executor_capability_discovery: Option<&ExecutorCapabilityDiscoverySnapshot>, ) -> McpRuntimeProjection { self.runtime_config_with_context( McpServerContributionContext::for_step( @@ -117,6 +119,7 @@ impl McpManager { thread_store, originator, ready_selected_capability_roots, + executor_capability_discovery, ), Some(originator), ) diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index b984867a8fec..ee7b7d17c662 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -1,5 +1,7 @@ use super::*; use crate::mcp::McpRuntimeProjection; +use codex_exec_server::ExecutorCapabilityDiscoveryCache; +use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_exec_server::MAX_SELECTED_CAPABILITY_ROOTS; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::ElicitationReviewRequest; @@ -87,6 +89,9 @@ impl Session { .await; let ready_selected_capability_roots = Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery = self + .executor_capability_discovery_for_step(config, &ready_selected_capability_roots) + .await; self.services .mcp_manager .runtime_config_for_step( @@ -95,6 +100,7 @@ impl Session { &self.services.thread_extension_data, &originator, &ready_selected_capability_roots, + executor_capability_discovery.as_deref(), ) .await .config @@ -117,6 +123,7 @@ impl Session { turn_context: &TurnContext, environments: &TurnEnvironmentSnapshot, selected_capability_roots: &[ResolvedSelectedCapabilityRoot], + executor_capability_discovery: Option<&ExecutorCapabilityDiscoverySnapshot>, ) -> Arc { let ready_selected_capability_roots = Self::ready_selected_capability_roots(selected_capability_roots); @@ -141,6 +148,7 @@ impl Session { &self.services.thread_extension_data, &turn_context.originator, &ready_selected_capability_roots, + executor_capability_discovery, ) .await; let mcp_config = &mcp_projection.config; @@ -190,6 +198,32 @@ impl Session { .await } + #[tracing::instrument( + name = "capability_roots.snapshot_for_step", + skip_all, + fields(root_count = ready_selected_capability_roots.len()) + )] + pub(crate) async fn executor_capability_discovery_for_step( + &self, + config: &Config, + ready_selected_capability_roots: &[SelectedCapabilityRoot], + ) -> Option> { + if !config + .features + .enabled(Feature::ExecutorCapabilityDiscovery) + { + return None; + } + let environment_manager = self.services.turn_environments.environment_manager(); + let cache = self + .services + .thread_extension_data + .get_or_init(|| ExecutorCapabilityDiscoveryCache::new(environment_manager)); + Some(Arc::new( + cache.snapshot(ready_selected_capability_roots).await, + )) + } + pub(crate) async fn resolve_selected_capability_roots_for_step( &self, environments: &TurnEnvironmentSnapshot, @@ -511,6 +545,12 @@ impl Session { let current_runtime = self.services.latest_mcp_runtime(); let ready_selected_capability_roots = current_runtime.ready_selected_capability_roots().to_vec(); + let executor_capability_discovery = self + .executor_capability_discovery_for_step( + &refresh_config, + &ready_selected_capability_roots, + ) + .await; let mut mcp_projection = self .services .mcp_manager @@ -520,6 +560,7 @@ impl Session { &self.services.thread_extension_data, &turn_context.originator, &ready_selected_capability_roots, + executor_capability_discovery.as_deref(), ) .await; mcp_projection.config.mcp_server_catalog = mcp_projection @@ -578,6 +619,12 @@ impl Session { let current_runtime = self.services.latest_mcp_runtime(); let ready_selected_capability_roots = current_runtime.ready_selected_capability_roots().to_vec(); + let executor_capability_discovery = self + .executor_capability_discovery_for_step( + refresh_config, + &ready_selected_capability_roots, + ) + .await; let mcp_projection = self .services .mcp_manager @@ -587,6 +634,7 @@ impl Session { &self.services.thread_extension_data, &turn_context.originator, &ready_selected_capability_roots, + executor_capability_discovery.as_deref(), ) .await; self.refresh_mcp_servers_inner( @@ -613,7 +661,7 @@ impl Session { available } - fn ready_selected_capability_roots( + pub(crate) fn ready_selected_capability_roots( selected_capability_roots: &[ResolvedSelectedCapabilityRoot], ) -> Vec { selected_capability_roots diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index c7ead4c21e39..b85a3af54fdb 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2859,17 +2859,27 @@ impl Session { let selected_capability_roots = self .resolve_selected_capability_roots_for_step(&environments) .await; + let ready_selected_capability_roots = + Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery = self + .executor_capability_discovery_for_step( + &turn_context.config, + &ready_selected_capability_roots, + ) + .await; let mcp = self .mcp_runtime_for_step( turn_context.as_ref(), &environments, &selected_capability_roots, + executor_capability_discovery.as_deref(), ) .await; Arc::new(StepContext::new( turn_context, environments, selected_capability_roots, + executor_capability_discovery, mcp, loaded_agents_md, )) diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 02293ffe09fd..0759789ab2b7 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -717,6 +717,7 @@ impl Session { thread_extension_data_for_mcp, &mcp_originator, /*ready_selected_capability_roots*/ &[], + /*executor_capability_discovery*/ None, ) .await; let mcp_config = &mcp_projection.config; diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index 1d7316d91443..78895950a444 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/codex-rs/core/src/session/step_context.rs @@ -4,6 +4,7 @@ use crate::agents_md::LoadedAgentsMd; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::session::McpRuntimeSnapshot; use crate::session::turn_context::TurnContext; +use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::ToolInfo; use tokio::sync::OnceCell; @@ -15,6 +16,8 @@ pub(crate) struct StepContext { pub(crate) environments: TurnEnvironmentSnapshot, /// Capability roots bound to ready environments in this exact step. pub(crate) selected_capability_roots: Vec, + /// Executor-materialized capability files shared by MCP and skills in this exact step. + pub(crate) executor_capability_discovery: Option>, /// The exact MCP config and manager used to advertise and execute tools for this step. pub(crate) mcp: Arc, /// The fixed MCP tool list used for this exact sampling request. @@ -28,6 +31,7 @@ impl StepContext { turn: Arc, environments: TurnEnvironmentSnapshot, selected_capability_roots: Vec, + executor_capability_discovery: Option>, mcp: Arc, loaded_agents_md: Option>, ) -> Self { @@ -35,6 +39,7 @@ impl StepContext { turn, environments, selected_capability_roots, + executor_capability_discovery, mcp, mcp_tool_snapshot: OnceCell::new(), loaded_agents_md, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 808ad1f47d8c..bc180b46294f 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -200,6 +200,7 @@ impl StepContext { Arc::clone(&turn), environments, Vec::new(), + /*executor_capability_discovery*/ None, crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config), /*loaded_agents_md*/ None, )) @@ -7813,7 +7814,12 @@ async fn deferred_environment_roots_refresh_plugin_availability() { ); let new_runtime = session - .mcp_runtime_for_step(&turn_context, &environments, &resolved_roots) + .mcp_runtime_for_step( + &turn_context, + &environments, + &resolved_roots, + /*executor_capability_discovery*/ None, + ) .await; assert!(!old_runtime.plugins_available()); diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index e4632544f499..65071deca3a6 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -77,6 +77,9 @@ impl Session { turn_id: turn_context.sub_id.as_str(), environments: &environments, ready_selected_capability_roots: &ready_selected_capability_roots, + executor_capability_discovery: step_context + .executor_capability_discovery + .as_deref(), session_store: &self.services.session_extension_data, thread_store: &self.services.thread_extension_data, turn_store: turn_context.extension_data.as_ref(), diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index ce3589357cd1..2c3bfd80eac5 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -687,6 +687,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() &first_session.services.thread_extension_data, &first_originator, /*ready_selected_capability_roots*/ &[], + /*executor_capability_discovery*/ None, ) .await; let second_session = &second_thread.thread.session; @@ -700,6 +701,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() &second_session.services.thread_extension_data, &second_originator, /*ready_selected_capability_roots*/ &[], + /*executor_capability_discovery*/ None, ) .await; diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index acc08cb3a491..e8b7ba732b55 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -685,6 +685,7 @@ async fn environment_tools_follow_the_step_context() { Arc::clone(&turn), environments, Vec::new(), + /*executor_capability_discovery*/ None, crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config), /*loaded_agents_md*/ None, )); diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index 6eb2b92848cb..86c90c8108d7 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -1,10 +1,12 @@ use std::collections::HashMap; +use std::sync::Arc; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use codex_file_system::FileSystemSandboxContext; pub use codex_file_system::WalkOptions; pub use codex_file_system::WalkOutcome; use codex_network_proxy::ManagedNetworkSandboxContext; +use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::ShellEnvironmentPolicyInherit; use codex_shell_command::shell_detect::DetectedShell; use codex_utils_path_uri::PathUri; @@ -37,6 +39,14 @@ pub const FS_READ_DIRECTORY_METHOD: &str = "fs/readDirectory"; pub const FS_WALK_METHOD: &str = "fs/walk"; pub const FS_REMOVE_METHOD: &str = "fs/remove"; pub const FS_COPY_METHOD: &str = "fs/copy"; +/// Discovers capability manifests below selected roots using executor-local filesystem access. +pub const CAPABILITY_ROOTS_DISCOVER_METHOD: &str = "capabilityRoots/discoverV1"; +/// Ordered plugin manifest paths recognized beneath a plugin root. +pub const DISCOVERABLE_PLUGIN_MANIFEST_PATHS: &[&str] = &[ + ".codex-plugin/plugin.json", + ".claude-plugin/plugin.json", + ".cursor-plugin/plugin.json", +]; /// JSON-RPC request method for executor-side HTTP requests. pub const HTTP_REQUEST_METHOD: &str = "http/request"; /// JSON-RPC notification method for streamed executor HTTP response bodies. @@ -431,6 +441,119 @@ pub struct FsCopyParams { #[serde(rename_all = "camelCase")] pub struct FsCopyResponse {} +/// Roots to inspect for plugin and skill capability manifests. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapabilityRootsDiscoverParams { + pub roots: Vec, +} + +/// One caller-selected capability root. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapabilityRootDiscoverRequest { + /// Opaque caller identity returned unchanged in the response. + pub id: String, + /// Absolute root URI interpreted using the exec-server host's path rules. + pub path: PathUri, +} + +/// Executor-local discovery results in request order. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapabilityRootsDiscoverResponse { + pub roots: Vec, +} + +/// Recognized UTF-8 capability file materialized by the exec-server. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapabilityTextFile { + pub path: PathUri, + pub contents: String, +} + +/// Plugin files declared directly by a selected root. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DiscoveredPluginFiles { + pub manifest: CapabilityTextFile, + /// File-backed MCP declarations, including the conventional `.mcp.json` fallback. + #[serde(default)] + pub mcp_config: Option, + /// File-backed connector declarations. + #[serde(default)] + pub apps_config: Option, +} + +/// A skill instructions file and its optional sibling metadata. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DiscoveredSkillFiles { + pub instructions: CapabilityTextFile, + #[serde(default)] + pub metadata: Option, +} + +/// Manifest bundle for one selected root. +/// +/// Discovery failures are root-local so one broken package does not discard valid siblings. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapabilityRootDiscovery { + pub id: String, + pub path: PathUri, + #[serde(default)] + pub plugin: Option, + #[serde(default)] + pub skills: Vec, + /// Plugin manifests found while scanning the root, used to namespace nested skills. + #[serde(default)] + pub namespace_manifests: Vec, + #[serde(default)] + pub warnings: Vec, + #[serde(default)] + pub error: Option, +} + +/// Immutable results for the selected capability roots visible in one model step. +#[derive(Clone, Debug)] +pub struct ExecutorCapabilityDiscoverySnapshot { + roots: Arc<[ExecutorCapabilityDiscoverySnapshotEntry]>, +} + +#[derive(Clone, Debug)] +pub struct ExecutorCapabilityDiscoverySnapshotEntry { + pub selected_root: SelectedCapabilityRoot, + pub result: Result, String>, +} + +impl ExecutorCapabilityDiscoverySnapshot { + pub fn new( + selected_roots: &[SelectedCapabilityRoot], + discoveries: Vec, String>>, + ) -> Self { + debug_assert_eq!(selected_roots.len(), discoveries.len()); + Self { + roots: selected_roots + .iter() + .cloned() + .zip(discoveries) + .map( + |(selected_root, result)| ExecutorCapabilityDiscoverySnapshotEntry { + selected_root, + result, + }, + ) + .collect(), + } + } + + pub fn roots(&self) -> &[ExecutorCapabilityDiscoverySnapshotEntry] { + &self.roots + } +} + /// HTTP header represented in the executor protocol. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] diff --git a/codex-rs/exec-server/src/capability_discovery.rs b/codex-rs/exec-server/src/capability_discovery.rs new file mode 100644 index 000000000000..5de1e252ecbd --- /dev/null +++ b/codex-rs/exec-server/src/capability_discovery.rs @@ -0,0 +1,462 @@ +use std::collections::HashSet; +use std::io; + +use codex_exec_server_protocol::CapabilityRootDiscoverRequest; +use codex_exec_server_protocol::CapabilityRootDiscovery; +use codex_exec_server_protocol::CapabilityRootsDiscoverParams; +use codex_exec_server_protocol::CapabilityRootsDiscoverResponse; +use codex_exec_server_protocol::CapabilityTextFile; +use codex_exec_server_protocol::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; +use codex_exec_server_protocol::DiscoveredPluginFiles; +use codex_exec_server_protocol::DiscoveredSkillFiles; +use codex_file_system::ExecutorFileSystem; +use codex_file_system::WalkEntryKind; +use codex_file_system::WalkOptions; +use codex_utils_path_uri::PathUri; +use futures::StreamExt; +use serde::Deserialize; +use serde_json::Value; + +const MAX_ROOTS_PER_REQUEST: usize = 128; +const MAX_SCAN_DEPTH: usize = 6; +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; +const SKILL_FILE_NAME: &str = "SKILL.md"; +const SKILL_METADATA_PATH: &str = "agents/openai.yaml"; +const DEFAULT_MCP_CONFIG_PATH: &str = ".mcp.json"; + +#[derive(Debug, thiserror::Error)] +pub enum CapabilityDiscoveryError { + #[error("capability root discovery accepts at most {MAX_ROOTS_PER_REQUEST} roots")] + TooManyRoots, +} + +/// Discovers and materializes capability manifests using one executor-local filesystem. +/// +/// Product parsing and policy intentionally remain with the caller. This operation owns the +/// filesystem-expensive portion: bounded traversal, recognized-file selection, and reads. +#[tracing::instrument( + name = "capability_roots.discover_v1", + skip_all, + fields(root_count = params.roots.len()) +)] +pub async fn discover_capability_roots( + file_system: &dyn ExecutorFileSystem, + params: CapabilityRootsDiscoverParams, +) -> Result { + if params.roots.len() > MAX_ROOTS_PER_REQUEST { + return Err(CapabilityDiscoveryError::TooManyRoots); + } + + let roots = futures::stream::iter(params.roots) + .map(|root| discover_root(file_system, root)) + .buffered(MAX_CONCURRENT_ROOTS) + .collect() + .await; + Ok(CapabilityRootsDiscoverResponse { roots }) +} + +async fn discover_root( + file_system: &dyn ExecutorFileSystem, + request: CapabilityRootDiscoverRequest, +) -> CapabilityRootDiscovery { + let CapabilityRootDiscoverRequest { id, path } = request; + let mut discovery = CapabilityRootDiscovery { + id, + path: path.clone(), + plugin: None, + skills: Vec::new(), + namespace_manifests: Vec::new(), + warnings: Vec::new(), + error: None, + }; + + match file_system.get_metadata(&path, /*sandbox*/ None).await { + Ok(metadata) if metadata.is_directory => {} + Ok(_) => { + discovery.error = Some(format!("capability root {path} is not a directory")); + return discovery; + } + Err(error) => { + discovery.error = Some(format!("failed to inspect capability root {path}: {error}")); + return discovery; + } + } + + let walk = match file_system + .walk( + &path, + WalkOptions { + max_depth: MAX_SCAN_DEPTH, + max_directories: MAX_DIRECTORIES_PER_ROOT, + max_entries: MAX_ENTRIES_PER_ROOT, + follow_directory_symlinks: true, + prune_hidden_directories: false, + }, + /*sandbox*/ None, + ) + .await + { + Ok(walk) => walk, + Err(error) => { + discovery.error = Some(format!("failed to scan capability root {path}: {error}")); + return discovery; + } + }; + discovery + .warnings + .extend(walk.errors.into_iter().map(|error| { + format!( + "failed to scan capability path {}: {}", + error.path, error.message + ) + })); + if walk.truncated { + discovery.warnings.push(format!( + "capability scan reached its traversal limit (root: {path})" + )); + } + + let mut skill_paths = Vec::new(); + let mut namespace_manifest_paths = Vec::new(); + for entry in walk.entries { + if entry.kind != WalkEntryKind::File { + continue; + } + if entry.path.basename().as_deref() == Some(SKILL_FILE_NAME) { + skill_paths.push(entry.path.clone()); + } + if is_plugin_manifest_path(&entry.path) { + namespace_manifest_paths.push(entry.path); + } + } + skill_paths.sort_unstable_by_key(PathUri::to_string); + namespace_manifest_paths.sort_unstable_by(|left, right| { + let left_root = plugin_root_for_manifest(left).map(|path| path.to_string()); + let right_root = plugin_root_for_manifest(right).map(|path| path.to_string()); + left_root + .cmp(&right_root) + .then_with(|| plugin_manifest_priority(left).cmp(&plugin_manifest_priority(right))) + }); + + let mut budget = BundleBudget::default(); + let root_manifest = + read_first_plugin_manifest(file_system, &path, &mut budget, &mut discovery.warnings).await; + + let inherited_manifest = match root_manifest.as_ref() { + Some(manifest) => Some(manifest.clone()), + None => { + read_nearest_ancestor_manifest(file_system, &path, &mut budget, &mut discovery.warnings) + .await + } + }; + let mut seen_namespace_roots = HashSet::new(); + if let Some(manifest) = inherited_manifest { + if let Some(plugin_root) = plugin_root_for_manifest(&manifest.path) { + seen_namespace_roots.insert(plugin_root); + } + discovery.namespace_manifests.push(manifest); + } + for manifest_path in namespace_manifest_paths { + let Some(plugin_root) = plugin_root_for_manifest(&manifest_path) else { + continue; + }; + if !seen_namespace_roots.insert(plugin_root) { + continue; + } + if let Some(manifest) = read_optional_text_file( + file_system, + manifest_path, + &mut budget, + &mut discovery.warnings, + ) + .await + { + discovery.namespace_manifests.push(manifest); + } + } + + if let Some(manifest) = root_manifest { + let declarations = plugin_declaration_paths(&path, &manifest, &mut discovery.warnings); + let mcp_path = if declarations.mcp_inline { + None + } else { + declarations + .mcp_config + .or_else(|| path.join(DEFAULT_MCP_CONFIG_PATH).ok()) + }; + let mcp_config = match mcp_path { + Some(path) => { + read_optional_text_file(file_system, path, &mut budget, &mut discovery.warnings) + .await + } + None => None, + }; + let apps_config = match declarations.apps_config { + Some(path) => { + read_optional_text_file(file_system, path, &mut budget, &mut discovery.warnings) + .await + } + None => None, + }; + discovery.plugin = Some(DiscoveredPluginFiles { + manifest, + mcp_config, + apps_config, + }); + } + + for skill_path in skill_paths { + let Some(instructions) = read_optional_text_file( + file_system, + skill_path.clone(), + &mut budget, + &mut discovery.warnings, + ) + .await + else { + continue; + }; + let metadata = match skill_path + .parent() + .and_then(|skill_dir| skill_dir.join(SKILL_METADATA_PATH).ok()) + { + Some(metadata_path) => { + read_optional_text_file( + file_system, + metadata_path, + &mut budget, + &mut discovery.warnings, + ) + .await + } + None => None, + }; + discovery.skills.push(DiscoveredSkillFiles { + instructions, + metadata, + }); + } + + discovery +} + +async fn read_first_plugin_manifest( + file_system: &dyn ExecutorFileSystem, + root: &PathUri, + budget: &mut BundleBudget, + warnings: &mut Vec, +) -> Option { + for relative_path in DISCOVERABLE_PLUGIN_MANIFEST_PATHS { + let Ok(path) = root.join(relative_path) else { + continue; + }; + if let Some(manifest) = read_optional_text_file(file_system, path, budget, warnings).await { + return Some(manifest); + } + } + None +} + +async fn read_nearest_ancestor_manifest( + file_system: &dyn ExecutorFileSystem, + root: &PathUri, + budget: &mut BundleBudget, + warnings: &mut Vec, +) -> Option { + let mut ancestor = root.parent(); + while let Some(path) = ancestor { + if let Some(manifest) = + read_first_plugin_manifest(file_system, &path, budget, warnings).await + { + return Some(manifest); + } + ancestor = path.parent(); + } + None +} + +async fn read_optional_text_file( + file_system: &dyn ExecutorFileSystem, + path: PathUri, + budget: &mut BundleBudget, + warnings: &mut Vec, +) -> Option { + let metadata = match file_system.get_metadata(&path, /*sandbox*/ None).await { + Ok(metadata) if metadata.is_file => metadata, + Ok(_) => return None, + Err(error) if error.kind() == io::ErrorKind::NotFound => return None, + Err(error) => { + warnings.push(format!("failed to inspect capability file {path}: {error}")); + return None; + } + }; + let Ok(size) = usize::try_from(metadata.size) else { + warnings.push(format!("capability file {path} is too large")); + return None; + }; + if size > MAX_FILE_BYTES { + warnings.push(format!( + "capability file {path} exceeds the {MAX_FILE_BYTES}-byte limit" + )); + return None; + } + if !budget.can_add(size) { + warnings.push(format!( + "capability root bundle exceeds the {MAX_BUNDLE_BYTES_PER_ROOT}-byte limit" + )); + return None; + } + let mut stream = match file_system.read_file_stream(&path, /*sandbox*/ None).await { + Ok(stream) => stream, + Err(error) => { + warnings.push(format!("failed to read capability file {path}: {error}")); + return None; + } + }; + let mut contents = Vec::with_capacity(size); + while let Some(chunk) = stream.next().await { + let chunk = match chunk { + Ok(chunk) => chunk, + Err(error) => { + warnings.push(format!("failed to read capability file {path}: {error}")); + return None; + } + }; + let Some(new_len) = contents.len().checked_add(chunk.len()) else { + warnings.push(format!("capability file {path} exceeded its read limit")); + return None; + }; + if new_len > MAX_FILE_BYTES || !budget.can_add(new_len) { + warnings.push(format!("capability file {path} exceeded its read limit")); + return None; + } + contents.extend_from_slice(&chunk); + } + let contents = match String::from_utf8(contents) { + Ok(contents) => contents, + Err(error) => { + warnings.push(format!("capability file {path} is not UTF-8: {error}")); + return None; + } + }; + budget.add(contents.len()); + Some(CapabilityTextFile { path, contents }) +} + +fn is_plugin_manifest_path(path: &PathUri) -> bool { + plugin_manifest_priority(path).is_some() +} + +fn plugin_manifest_priority(path: &PathUri) -> Option { + if path.basename().as_deref() != Some("plugin.json") { + return None; + } + let manifest_directory = path.parent()?.basename()?; + DISCOVERABLE_PLUGIN_MANIFEST_PATHS + .iter() + .position(|relative_path| { + relative_path.strip_suffix("/plugin.json") == Some(manifest_directory.as_str()) + }) +} + +fn plugin_root_for_manifest(path: &PathUri) -> Option { + path.parent()?.parent() +} + +#[derive(Default)] +struct BundleBudget { + bytes: usize, +} + +impl BundleBudget { + fn can_add(&self, bytes: usize) -> bool { + self.bytes + .checked_add(bytes) + .is_some_and(|total| total <= MAX_BUNDLE_BYTES_PER_ROOT) + } + + fn add(&mut self, bytes: usize) { + self.bytes += bytes; + } +} + +#[derive(Default)] +struct PluginDeclarationPaths { + mcp_config: Option, + mcp_inline: bool, + apps_config: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RawPluginDeclarations { + #[serde(default)] + mcp_servers: Option, + #[serde(default)] + apps: Option, +} + +fn plugin_declaration_paths( + root: &PathUri, + manifest: &CapabilityTextFile, + warnings: &mut Vec, +) -> PluginDeclarationPaths { + let declarations = match serde_json::from_str::(&manifest.contents) { + Ok(declarations) => declarations, + Err(_) => return PluginDeclarationPaths::default(), + }; + PluginDeclarationPaths { + mcp_config: declarations.mcp_servers.as_ref().and_then(|value| { + declared_file_path(root, "mcpServers", value, &manifest.path, warnings) + }), + mcp_inline: declarations + .mcp_servers + .as_ref() + .is_some_and(Value::is_object), + apps_config: declarations + .apps + .as_ref() + .and_then(|value| declared_file_path(root, "apps", value, &manifest.path, warnings)), + } +} + +fn declared_file_path( + root: &PathUri, + field: &str, + value: &Value, + manifest_path: &PathUri, + warnings: &mut Vec, +) -> Option { + let Value::String(path) = value else { + return None; + }; + let Some(relative_path) = path.strip_prefix("./") else { + warnings.push(format!( + "ignoring {field} in {manifest_path}: path must start with `./`" + )); + return None; + }; + if relative_path.is_empty() + || relative_path + .split(['/', '\\']) + .any(|component| component == "..") + { + warnings.push(format!( + "ignoring {field} in {manifest_path}: path must remain below the capability root" + )); + return None; + } + match root.join(relative_path) { + Ok(path) if path.starts_with(root) => Some(path), + Ok(_) | Err(_) => { + warnings.push(format!( + "ignoring {field} in {manifest_path}: path must remain below the capability root" + )); + None + } + } +} diff --git a/codex-rs/exec-server/src/capability_discovery_cache.rs b/codex-rs/exec-server/src/capability_discovery_cache.rs new file mode 100644 index 000000000000..dddf87a160ae --- /dev/null +++ b/codex-rs/exec-server/src/capability_discovery_cache.rs @@ -0,0 +1,197 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use codex_protocol::capabilities::CapabilityRootLocation; +use codex_protocol::capabilities::SelectedCapabilityRoot; +use tokio::sync::Mutex; + +use crate::CapabilityRootDiscoverRequest; +use crate::CapabilityRootDiscovery; +use crate::CapabilityRootsDiscoverParams; +use crate::EnvironmentManager; +use crate::ExecutorCapabilityDiscoverySnapshot; + +/// Thread-scoped cache shared by capability consumers using the high-level executor API. +/// +/// A single miss batches every requested root by environment. The cache deliberately has no +/// invalidation: selected roots are already treated as stable for the lifetime of a thread by the +/// existing plugin and skill providers. +pub struct ExecutorCapabilityDiscoveryCache { + environment_manager: Arc, + entries: Mutex>, +} + +impl std::fmt::Debug for ExecutorCapabilityDiscoveryCache { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ExecutorCapabilityDiscoveryCache") + .finish_non_exhaustive() + } +} + +struct CachedRoot { + selected_root: SelectedCapabilityRoot, + // Both successes and failures are memoized for the thread. Retrying a transient failure for + // the same stable selected root requires explicit invalidation or a new thread. + result: Result, String>, +} + +impl ExecutorCapabilityDiscoveryCache { + pub fn new(environment_manager: Arc) -> Self { + Self { + environment_manager, + entries: Mutex::new(Vec::new()), + } + } + + /// Returns discoveries in the same order as `selected_roots`. + #[tracing::instrument( + name = "capability_roots.discovery_cache.resolve", + skip_all, + fields(root_count = selected_roots.len()) + )] + pub async fn discover( + &self, + selected_roots: &[SelectedCapabilityRoot], + ) -> Vec, String>> { + let missing = { + let entries = self.entries.lock().await; + selected_roots + .iter() + .filter(|selected_root| { + !entries + .iter() + .any(|cached| cached.selected_root == **selected_root) + }) + .cloned() + .collect::>() + }; + let discovered = self.discover_missing(missing).await; + let mut entries = self.entries.lock().await; + for discovered_root in discovered { + if !entries + .iter() + .any(|cached| cached.selected_root == discovered_root.selected_root) + { + entries.push(discovered_root); + } + } + selected_roots + .iter() + .map(|selected_root| { + match entries + .iter() + .find(|cached| cached.selected_root == *selected_root) + { + Some(cached) => cached.result.clone(), + None => Err(format!( + "selected capability root `{}` was not discovered", + selected_root.id + )), + } + }) + .collect() + } + + /// Resolves the selected roots once and freezes their results for one model step. + pub async fn snapshot( + &self, + selected_roots: &[SelectedCapabilityRoot], + ) -> ExecutorCapabilityDiscoverySnapshot { + ExecutorCapabilityDiscoverySnapshot::new( + selected_roots, + self.discover(selected_roots).await, + ) + } + + async fn discover_missing(&self, missing: Vec) -> Vec { + let mut grouped = BTreeMap::>::new(); + for selected_root in missing { + let CapabilityRootLocation::Environment { environment_id, .. } = + &selected_root.location; + grouped + .entry(environment_id.clone()) + .or_default() + .push(selected_root); + } + + let discoveries = futures::future::join_all(grouped.into_iter().map( + |(environment_id, selected_roots)| async move { + let Some(environment) = self.environment_manager.get_environment(&environment_id) + else { + let error = format!("environment `{environment_id}` is unavailable"); + return selected_roots + .into_iter() + .map(|selected_root| CachedRoot { + selected_root, + result: Err(error.clone()), + }) + .collect::>(); + }; + let params = CapabilityRootsDiscoverParams { + roots: selected_roots + .iter() + .map(|selected_root| { + let CapabilityRootLocation::Environment { path, .. } = + &selected_root.location; + CapabilityRootDiscoverRequest { + id: selected_root.id.clone(), + path: path.clone(), + } + }) + .collect(), + }; + let response = match environment.discover_capability_roots(params).await { + Ok(response) => response, + Err(error) => { + let error = error.to_string(); + return selected_roots + .into_iter() + .map(|selected_root| CachedRoot { + selected_root, + result: Err(error.clone()), + }) + .collect(); + } + }; + if response.roots.len() != selected_roots.len() { + let error = format!( + "exec-server returned {} capability roots for {} requests", + response.roots.len(), + selected_roots.len() + ); + return selected_roots + .into_iter() + .map(|selected_root| CachedRoot { + selected_root, + result: Err(error.clone()), + }) + .collect(); + } + selected_roots + .into_iter() + .zip(response.roots) + .map(|(selected_root, discovery)| { + let CapabilityRootLocation::Environment { path, .. } = + &selected_root.location; + let result = if discovery.id == selected_root.id && discovery.path == *path + { + Ok(Arc::new(discovery)) + } else { + Err(format!( + "exec-server returned mismatched capability root `{}` at {}", + discovery.id, discovery.path + )) + }; + CachedRoot { + selected_root, + result, + } + }) + .collect() + }, + )) + .await; + discoveries.into_iter().flatten().collect() + } +} diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 736fc9ca5a2a..77e116f6ce4f 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -39,6 +39,9 @@ use crate::environment::EnvironmentConnectionState; use crate::process::ExecProcessEvent; use crate::process::ExecProcessEventLog; use crate::process::ExecProcessEventReceiver; +use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD; +use crate::protocol::CapabilityRootsDiscoverParams; +use crate::protocol::CapabilityRootsDiscoverResponse; use crate::protocol::ENVIRONMENT_INFO_METHOD; use crate::protocol::ENVIRONMENT_STATUS_METHOD; use crate::protocol::EXEC_CLOSED_METHOD; @@ -695,6 +698,13 @@ impl ExecServerClient { ) } + pub async fn discover_capability_roots( + &self, + params: CapabilityRootsDiscoverParams, + ) -> Result { + self.call(CAPABILITY_ROOTS_DISCOVER_METHOD, ¶ms).await + } + pub async fn read(&self, params: ReadParams) -> Result { self.call(EXEC_READ_METHOD, ¶ms).await } diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 58cf5193cf87..1192cfe70d13 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -9,6 +9,8 @@ use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; use futures::FutureExt; +use crate::CapabilityRootsDiscoverParams; +use crate::CapabilityRootsDiscoverResponse; use crate::ExecServerError; use crate::ExecServerRuntimePaths; use crate::ExecutorFileSystem; @@ -713,6 +715,19 @@ impl Environment { } } + /// Discovers plugin and skill manifests through the environment's high-level discovery API. + pub async fn discover_capability_roots( + &self, + params: CapabilityRootsDiscoverParams, + ) -> Result { + match &self.remote_client { + Some(client) => client.get().await?.discover_capability_roots(params).await, + None => crate::discover_capability_roots(self.filesystem.as_ref(), params) + .await + .map_err(|error| ExecServerError::Protocol(error.to_string())), + } + } + /// Starts connecting a remote environment without waiting for it. /// Requires an active Tokio runtime when background startup is supported. pub fn start_connecting(&self) { diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index c192eebd1e0f..4d315b9ba6fa 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -1,3 +1,5 @@ +mod capability_discovery; +mod capability_discovery_cache; mod client; mod client_api; mod client_transport; @@ -33,6 +35,9 @@ mod websocket_pong_watchdog; use codex_exec_server_protocol as protocol; +pub use capability_discovery::CapabilityDiscoveryError; +pub use capability_discovery::discover_capability_roots; +pub use capability_discovery_cache::ExecutorCapabilityDiscoveryCache; pub use client::ExecServerClient; pub use client::ExecServerError; pub use client::http_client::HttpResponseBodyStream; @@ -43,6 +48,7 @@ pub use client_api::NoiseRendezvousConnectArgs; pub use client_api::NoiseRendezvousConnectBundle; pub use client_api::NoiseRendezvousConnectProvider; pub use client_api::RemoteExecServerConnectArgs; +pub use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; pub use codex_exec_server_protocol::ProcessId; pub use codex_file_system::CopyOptions; pub use codex_file_system::CreateDirectoryOptions; @@ -98,6 +104,14 @@ pub use process::ExecProcessEventReceiver; pub use process::ExecProcessFuture; pub use process::StartedExecProcess; pub use protocol::ByteChunk; +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::DiscoveredPluginFiles; +pub use protocol::DiscoveredSkillFiles; pub use protocol::EnvironmentInfo; pub use protocol::EnvironmentStatus; pub use protocol::EnvironmentStatusKind; diff --git a/codex-rs/exec-server/src/server/file_system_handler.rs b/codex-rs/exec-server/src/server/file_system_handler.rs index 21da65053a2c..7ead65016518 100644 --- a/codex-rs/exec-server/src/server/file_system_handler.rs +++ b/codex-rs/exec-server/src/server/file_system_handler.rs @@ -4,6 +4,8 @@ use base64::Engine as _; use base64::engine::general_purpose::STANDARD; use codex_exec_server_protocol::JSONRPCErrorError; +use crate::CapabilityRootsDiscoverParams; +use crate::CapabilityRootsDiscoverResponse; use crate::CopyOptions; use crate::CreateDirectoryOptions; use crate::ExecServerRuntimePaths; @@ -65,6 +67,15 @@ impl FileSystemHandler { self.file_reads.close_all().await; } + pub(crate) async fn discover_capability_roots( + &self, + params: CapabilityRootsDiscoverParams, + ) -> Result { + crate::discover_capability_roots(&self.file_system, params) + .await + .map_err(|error| invalid_request(error.to_string())) + } + pub(crate) async fn open( &self, params: FsOpenParams, diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index e6754e950d34..fd49f4912cc1 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -14,6 +14,8 @@ use tokio_util::task::TaskTracker; use crate::ExecServerRuntimePaths; use crate::client::http_client::PendingReqwestHttpBodyStream; use crate::client::http_client::ReqwestHttpRequestRunner; +use crate::protocol::CapabilityRootsDiscoverParams; +use crate::protocol::CapabilityRootsDiscoverResponse; use crate::protocol::EnvironmentInfo; use crate::protocol::EnvironmentStatus; use crate::protocol::EnvironmentStatusKind; @@ -258,6 +260,14 @@ impl ExecServerHandler { self.file_system.read_file(params).await } + pub(crate) async fn discover_capability_roots( + &self, + params: CapabilityRootsDiscoverParams, + ) -> Result { + self.require_initialized_for("capability discovery")?; + self.file_system.discover_capability_roots(params).await + } + pub(crate) async fn fs_open( &self, params: FsOpenParams, diff --git a/codex-rs/exec-server/src/server/registry.rs b/codex-rs/exec-server/src/server/registry.rs index 33bff4847970..e47f9a37c2d0 100644 --- a/codex-rs/exec-server/src/server/registry.rs +++ b/codex-rs/exec-server/src/server/registry.rs @@ -1,5 +1,7 @@ use std::sync::Arc; +use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD; +use crate::protocol::CapabilityRootsDiscoverParams; use crate::protocol::ENVIRONMENT_INFO_METHOD; use crate::protocol::ENVIRONMENT_STATUS_METHOD; use crate::protocol::EXEC_METHOD; @@ -76,6 +78,12 @@ pub(crate) fn build_router() -> RpcRouter { ENVIRONMENT_STATUS_METHOD, |handler: Arc, _params: ()| async move { handler.environment_status() }, ); + router.request( + CAPABILITY_ROOTS_DISCOVER_METHOD, + |handler: Arc, params: CapabilityRootsDiscoverParams| async move { + handler.discover_capability_roots(params).await + }, + ); router.request( EXEC_READ_METHOD, |handler: Arc, params: ReadParams| async move { diff --git a/codex-rs/exec-server/tests/capability_discovery.rs b/codex-rs/exec-server/tests/capability_discovery.rs new file mode 100644 index 000000000000..988da27b27ea --- /dev/null +++ b/codex-rs/exec-server/tests/capability_discovery.rs @@ -0,0 +1,229 @@ +mod common; + +use codex_exec_server::CAPABILITY_ROOTS_DISCOVER_METHOD; +use codex_exec_server::CapabilityRootDiscovery; +use codex_exec_server::CapabilityRootsDiscoverParams; +use codex_exec_server::CapabilityRootsDiscoverResponse; +use codex_exec_server::InitializeParams; +use codex_exec_server::InitializeResponse; +use codex_exec_server_protocol::CapabilityRootDiscoverRequest; +use codex_exec_server_protocol::JSONRPCMessage; +use codex_exec_server_protocol::JSONRPCResponse; +use codex_utils_path_uri::PathUri; +use common::exec_server::exec_server; +use pretty_assertions::assert_eq; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovers_a_complete_capability_bundle_in_one_request() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + write_file( + &root.path().join(".codex-plugin/plugin.json"), + r#"{ + "name": "demo", + "interface": {"displayName": "Demo Plugin"}, + "mcpServers": "./config/mcp.json", + "apps": "./config/apps.json" +}"#, + )?; + write_file( + &root.path().join(".claude-plugin/plugin.json"), + r#"{"name":"lower-priority-claude"}"#, + )?; + write_file( + &root.path().join(".cursor-plugin/plugin.json"), + r#"{"name":"lower-priority-cursor"}"#, + )?; + write_file( + &root.path().join("config/mcp.json"), + r#"{"mcpServers":{"demo":{"command":"demo-server"}}}"#, + )?; + write_file( + &root.path().join("config/apps.json"), + r#"{"apps":{"demo":{"connector_id":"connector-demo"}}}"#, + )?; + write_file( + &root.path().join("skills/deploy/SKILL.md"), + "---\nname: deploy\ndescription: Deploy the service.\n---\n\nDeploy instructions.\n", + )?; + write_file( + &root.path().join("skills/deploy/agents/openai.yaml"), + "policy:\n allow_implicit_invocation: false\n", + )?; + write_file( + &root.path().join("nested/.claude-plugin/plugin.json"), + r#"{"name":"nested"}"#, + )?; + write_file( + &root.path().join("nested/skills/audit/SKILL.md"), + "---\nname: audit\ndescription: Audit the service.\n---\n", + )?; + write_file( + &root.path().join("nested-cursor/.cursor-plugin/plugin.json"), + r#"{"name":"cursor-nested"}"#, + )?; + write_file( + &root.path().join("nested-cursor/skills/review/SKILL.md"), + "---\nname: review\ndescription: Review the service.\n---\n", + )?; + + let mut server = exec_server().await?; + initialize(&mut server).await?; + let root_uri = PathUri::from_host_native_path(root.path())?; + let discovery = discover_root(&mut server, "demo@1", root_uri.clone()).await?; + + assert_eq!(discovery.id, "demo@1"); + assert_eq!(discovery.path, root_uri); + assert_eq!(discovery.error, None); + assert_eq!(discovery.warnings, Vec::::new()); + let plugin = discovery.plugin.as_ref().expect("root plugin"); + assert_eq!( + plugin.manifest.path, + root_uri.join(".codex-plugin/plugin.json")? + ); + assert!(plugin.manifest.contents.contains("Demo Plugin")); + assert_eq!( + plugin.mcp_config.as_ref().map(|file| &file.path), + Some(&root_uri.join("config/mcp.json")?) + ); + assert_eq!( + plugin.apps_config.as_ref().map(|file| &file.path), + Some(&root_uri.join("config/apps.json")?) + ); + assert_eq!( + discovery + .namespace_manifests + .iter() + .map(|file| file.path.clone()) + .collect::>(), + vec![ + root_uri.join(".codex-plugin/plugin.json")?, + root_uri.join("nested/.claude-plugin/plugin.json")?, + root_uri.join("nested-cursor/.cursor-plugin/plugin.json")?, + ] + ); + assert_eq!( + discovery + .skills + .iter() + .map(|skill| ( + skill.instructions.path.clone(), + skill + .metadata + .as_ref() + .map(|metadata| metadata.path.clone()), + )) + .collect::>(), + vec![ + (root_uri.join("nested-cursor/skills/review/SKILL.md")?, None,), + (root_uri.join("nested/skills/audit/SKILL.md")?, None,), + ( + root_uri.join("skills/deploy/SKILL.md")?, + Some(root_uri.join("skills/deploy/agents/openai.yaml")?), + ), + ] + ); + + server.shutdown().await?; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovers_cursor_plugin_without_reading_default_mcp_for_inline_servers() +-> anyhow::Result<()> { + let root = tempfile::tempdir()?; + write_file( + &root.path().join(".cursor-plugin/plugin.json"), + r#"{"name":"cursor-demo","mcpServers":{"inline":{"command":"inline"}}}"#, + )?; + write_file( + &root.path().join(".mcp.json"), + r#"{"mcpServers":{"should-not-load":{"command":"wrong"}}}"#, + )?; + + let mut server = exec_server().await?; + initialize(&mut server).await?; + let root_uri = PathUri::from_host_native_path(root.path())?; + let discovery = discover_root(&mut server, "cursor@1", root_uri.clone()).await?; + + assert_eq!(discovery.error, None); + assert_eq!(discovery.warnings, Vec::::new()); + let plugin = discovery.plugin.expect("cursor plugin"); + assert_eq!( + plugin.manifest.path, + root_uri.join(".cursor-plugin/plugin.json")? + ); + assert_eq!(plugin.mcp_config, None); + assert_eq!( + discovery + .namespace_manifests + .iter() + .map(|manifest| manifest.path.clone()) + .collect::>(), + vec![root_uri.join(".cursor-plugin/plugin.json")?] + ); + + server.shutdown().await?; + Ok(()) +} + +async fn discover_root( + server: &mut common::exec_server::ExecServerHarness, + id: &str, + path: PathUri, +) -> anyhow::Result { + let request_id = server + .send_request( + CAPABILITY_ROOTS_DISCOVER_METHOD, + serde_json::to_value(CapabilityRootsDiscoverParams { + roots: vec![CapabilityRootDiscoverRequest { + id: id.to_string(), + path, + }], + })?, + ) + .await?; + let response = server.next_event().await?; + let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else { + anyhow::bail!("expected discovery response, received {response:?}"); + }; + assert_eq!(id, request_id); + let response: CapabilityRootsDiscoverResponse = serde_json::from_value(result)?; + let [discovery] = response.roots.as_slice() else { + anyhow::bail!("expected exactly one discovered root"); + }; + Ok(discovery.clone()) +} + +async fn initialize(server: &mut common::exec_server::ExecServerHarness) -> anyhow::Result<()> { + let initialize_id = server + .send_request( + "initialize", + serde_json::to_value(InitializeParams { + client_name: "capability-discovery-test".to_string(), + resume_session_id: None, + })?, + ) + .await?; + let response = server + .wait_for_event(|event| { + matches!(event, JSONRPCMessage::Response(response) if response.id == initialize_id) + }) + .await?; + let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = response else { + unreachable!("wait predicate only accepts a response"); + }; + let _: InitializeResponse = serde_json::from_value(result)?; + server + .send_notification("initialized", serde_json::json!({})) + .await?; + Ok(()) +} + +fn write_file(path: &std::path::Path, contents: &str) -> anyhow::Result<()> { + let parent = path + .parent() + .ok_or_else(|| anyhow::anyhow!("test file should have a parent"))?; + std::fs::create_dir_all(parent)?; + std::fs::write(path, contents)?; + Ok(()) +} diff --git a/codex-rs/ext/extension-api/Cargo.toml b/codex-rs/ext/extension-api/Cargo.toml index ce441576ec4e..f465bef36fde 100644 --- a/codex-rs/ext/extension-api/Cargo.toml +++ b/codex-rs/ext/extension-api/Cargo.toml @@ -16,6 +16,7 @@ workspace = true [dependencies] codex-config = { workspace = true } codex-context-fragments = { workspace = true } +codex-exec-server-protocol = { workspace = true } codex-protocol = { workspace = true } codex-tools = { workspace = true } codex-utils-absolute-path = { workspace = true } diff --git a/codex-rs/ext/extension-api/src/contributors/mcp.rs b/codex-rs/ext/extension-api/src/contributors/mcp.rs index 30c67918d36e..66a716d2c2db 100644 --- a/codex-rs/ext/extension-api/src/contributors/mcp.rs +++ b/codex-rs/ext/extension-api/src/contributors/mcp.rs @@ -1,4 +1,5 @@ use codex_config::McpServerConfig; +use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; use codex_protocol::capabilities::SelectedCapabilityRoot; use crate::ExtensionData; @@ -20,6 +21,8 @@ pub struct McpServerContributionContext<'a, C> { originator: Option<&'a str>, /// Selected roots resolved against ready environments for this exact step. ready_selected_capability_roots: Option<&'a [SelectedCapabilityRoot]>, + /// Executor-materialized capability files shared by all consumers in this exact step. + executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>, } impl Clone for McpServerContributionContext<'_, C> { @@ -39,6 +42,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { thread_init: None, originator: None, ready_selected_capability_roots: None, + executor_capability_discovery: None, } } @@ -49,6 +53,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { thread_store: &'a ExtensionData, originator: &'a str, ready_selected_capability_roots: &'a [SelectedCapabilityRoot], + executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>, ) -> Self { Self { config, @@ -56,6 +61,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { thread_init: Some(thread_init), originator: Some(originator), ready_selected_capability_roots: Some(ready_selected_capability_roots), + executor_capability_discovery, } } @@ -83,6 +89,11 @@ impl<'a, C> McpServerContributionContext<'a, C> { pub fn ready_selected_capability_roots(&self) -> Option<&'a [SelectedCapabilityRoot]> { self.ready_selected_capability_roots } + + /// Returns the executor-materialized capability files for this model step, when enabled. + pub fn executor_capability_discovery(&self) -> Option<&'a ExecutorCapabilityDiscoverySnapshot> { + self.executor_capability_discovery + } } /// One extension-owned overlay for the runtime MCP server configuration. diff --git a/codex-rs/ext/extension-api/src/contributors/world_state.rs b/codex-rs/ext/extension-api/src/contributors/world_state.rs index 3bafebffa2f4..75fae6d4a83d 100644 --- a/codex-rs/ext/extension-api/src/contributors/world_state.rs +++ b/codex-rs/ext/extension-api/src/contributors/world_state.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::protocol::TurnEnvironmentSelection; @@ -14,6 +15,8 @@ pub struct WorldStateContributionInput<'a> { pub environments: &'a [TurnEnvironmentSelection], /// Selected roots whose stable environments are ready in this sampling step. pub ready_selected_capability_roots: &'a [SelectedCapabilityRoot], + /// Executor-materialized capability files shared by all consumers in this exact step. + pub executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>, pub session_store: &'a ExtensionData, pub thread_store: &'a ExtensionData, pub turn_store: &'a ExtensionData, diff --git a/codex-rs/ext/mcp/Cargo.toml b/codex-rs/ext/mcp/Cargo.toml index d918bd5bd9a5..e639081358ca 100644 --- a/codex-rs/ext/mcp/Cargo.toml +++ b/codex-rs/ext/mcp/Cargo.toml @@ -16,6 +16,7 @@ workspace = true codex-core = { workspace = true } codex-core-plugins = { workspace = true } codex-config = { workspace = true } +codex-connectors = { workspace = true } codex-connectors-extension = { workspace = true } codex-exec-server = { workspace = true } codex-extension-api = { workspace = true } diff --git a/codex-rs/ext/mcp/src/executor_plugin.rs b/codex-rs/ext/mcp/src/executor_plugin.rs index 2d63ab551fa7..f38971e3945f 100644 --- a/codex-rs/ext/mcp/src/executor_plugin.rs +++ b/codex-rs/ext/mcp/src/executor_plugin.rs @@ -13,6 +13,7 @@ use std::sync::Mutex; use self::provider::ExecutorPluginMcpProvider; +mod discovery; mod provider; /// Frozen MCP and connector declarations for one selected package. @@ -153,37 +154,75 @@ impl McpServerContributor for SelectedExecutorPluginMcpContributor { let Some(selected_roots) = context.ready_selected_capability_roots() else { return Vec::new(); }; - let state = thread_store.get_or_init(SelectedExecutorPluginMcpState::default); let mut contributions = Vec::new(); - for (selection_order, selected_root) in selected_roots.iter().enumerate() { - let Some(plugin) = self.metadata_for_root(&state, selected_root).await else { - continue; - }; - let mut servers = plugin.servers.iter().cloned().collect::>(); - context - .config() - .apply_plugin_mcp_server_requirements(&plugin.plugin_id, &mut servers); - let mut servers = servers.into_iter().collect::>(); - servers.sort_unstable_by(|left, right| left.0.cmp(&right.0)); - contributions.extend(servers.into_iter().map(|(name, config)| { - McpServerContribution::SelectedPlugin { - name, - plugin_id: plugin.plugin_id.clone(), - plugin_display_name: plugin.plugin_display_name.clone(), + if let Some(snapshot) = context.executor_capability_discovery() { + for (selection_order, root) in snapshot.roots().iter().enumerate() { + let discovery = match &root.result { + Ok(discovery) => discovery.as_ref(), + Err(error) => { + tracing::warn!( + selected_root = root.selected_root.id, + error, + "exec-server capability discovery request failed" + ); + continue; + } + }; + let Some(plugin) = + discovery::metadata_from_discovery(&root.selected_root, discovery) + else { + continue; + }; + contributions.extend(project_metadata( + context.config(), selection_order, - config: Box::new(config), - } - })); - // Keep the package visible even when it contributes only skills. - contributions.push(McpServerContribution::SelectedPluginPackage { - plugin_id: plugin.plugin_id, - plugin_display_name: plugin.plugin_display_name, - connector_ids: plugin.connector_ids, - }); + plugin, + )); + } + } else { + let state = thread_store.get_or_init(SelectedExecutorPluginMcpState::default); + for (selection_order, selected_root) in selected_roots.iter().enumerate() { + let Some(plugin) = self.metadata_for_root(&state, selected_root).await else { + continue; + }; + contributions.extend(project_metadata( + context.config(), + selection_order, + plugin, + )); + } } contributions }) } } + +fn project_metadata( + config: &Config, + selection_order: usize, + plugin: SelectedPluginMetadata, +) -> Vec { + let mut servers = plugin.servers.iter().cloned().collect::>(); + config.apply_plugin_mcp_server_requirements(&plugin.plugin_id, &mut servers); + let mut servers = servers.into_iter().collect::>(); + servers.sort_unstable_by(|left, right| left.0.cmp(&right.0)); + let mut contributions = servers + .into_iter() + .map(|(name, config)| McpServerContribution::SelectedPlugin { + name, + plugin_id: plugin.plugin_id.clone(), + plugin_display_name: plugin.plugin_display_name.clone(), + selection_order, + config: Box::new(config), + }) + .collect::>(); + // Keep the package visible even when it contributes only skills. + contributions.push(McpServerContribution::SelectedPluginPackage { + plugin_id: plugin.plugin_id, + plugin_display_name: plugin.plugin_display_name, + connector_ids: plugin.connector_ids, + }); + contributions +} diff --git a/codex-rs/ext/mcp/src/executor_plugin/discovery.rs b/codex-rs/ext/mcp/src/executor_plugin/discovery.rs new file mode 100644 index 000000000000..583698ff670a --- /dev/null +++ b/codex-rs/ext/mcp/src/executor_plugin/discovery.rs @@ -0,0 +1,154 @@ +use codex_connectors::parse_plugin_app_config; +use codex_core_plugins::manifest::parse_plugin_manifest_uri; +use codex_exec_server::CapabilityRootDiscovery; +use codex_mcp::parse_executor_plugin_mcp_config; +use codex_plugin::manifest::PluginManifestMcpServers; +use codex_protocol::capabilities::CapabilityRootLocation; +use codex_protocol::capabilities::SelectedCapabilityRoot; + +use super::SelectedPluginMetadata; + +pub(super) fn metadata_from_discovery( + selected_root: &SelectedCapabilityRoot, + discovery: &CapabilityRootDiscovery, +) -> Option { + for warning in &discovery.warnings { + tracing::warn!( + selected_root = selected_root.id, + warning, + "exec-server capability discovery warning" + ); + } + if let Some(error) = &discovery.error { + tracing::warn!( + selected_root = selected_root.id, + error, + "exec-server capability discovery failed" + ); + return None; + } + let plugin_files = discovery.plugin.as_ref()?; + let manifest = match parse_plugin_manifest_uri( + &discovery.path, + &plugin_files.manifest.path, + &plugin_files.manifest.contents, + ) { + Ok(manifest) => manifest, + Err(error) => { + tracing::warn!( + selected_root = selected_root.id, + path = %plugin_files.manifest.path, + %error, + "failed to parse exec-server-discovered plugin manifest" + ); + return None; + } + }; + let CapabilityRootLocation::Environment { environment_id, .. } = &selected_root.location; + let servers = match manifest.paths.mcp_servers.as_ref() { + Some(PluginManifestMcpServers::Object(contents)) => { + parse_mcp_servers(selected_root, &discovery.path, contents, environment_id) + } + Some(PluginManifestMcpServers::Path(path)) => plugin_files + .mcp_config + .as_ref() + .filter(|file| file.path == *path) + .map(|file| { + parse_mcp_servers( + selected_root, + &discovery.path, + &file.contents, + environment_id, + ) + }) + .unwrap_or_else(|| { + tracing::warn!( + selected_root = selected_root.id, + path = %path, + "exec-server capability bundle omitted declared MCP config" + ); + Vec::new() + }), + None => plugin_files + .mcp_config + .as_ref() + .map(|file| { + parse_mcp_servers( + selected_root, + &discovery.path, + &file.contents, + environment_id, + ) + }) + .unwrap_or_default(), + }; + let connector_ids = manifest + .paths + .apps + .as_ref() + .and_then(|path| { + plugin_files + .apps_config + .as_ref() + .filter(|file| file.path == *path) + .or_else(|| { + tracing::warn!( + selected_root = selected_root.id, + path = %path, + "exec-server capability bundle omitted declared connector config" + ); + None + }) + }) + .and_then(|file| match parse_plugin_app_config(&file.contents) { + Ok(declarations) => Some(declarations), + Err(error) => { + tracing::warn!( + selected_root = selected_root.id, + path = %file.path, + %error, + "failed to parse exec-server-discovered connector config" + ); + None + } + }) + .unwrap_or_default() + .into_iter() + .map(|declaration| declaration.connector_id.0) + .collect(); + + Some(SelectedPluginMetadata { + plugin_id: selected_root.id.clone(), + plugin_display_name: manifest.display_name().to_string(), + servers, + connector_ids, + }) +} + +fn parse_mcp_servers( + selected_root: &SelectedCapabilityRoot, + plugin_root: &codex_utils_path_uri::PathUri, + contents: &str, + environment_id: &str, +) -> Vec<(String, codex_config::McpServerConfig)> { + let parsed = match parse_executor_plugin_mcp_config(plugin_root, contents, environment_id) { + Ok(parsed) => parsed, + Err(error) => { + tracing::warn!( + selected_root = selected_root.id, + %error, + "failed to parse exec-server-discovered MCP config" + ); + return Vec::new(); + } + }; + for error in parsed.errors { + tracing::warn!( + selected_root = selected_root.id, + server = error.name, + error = error.message, + "ignoring invalid exec-server-discovered MCP server" + ); + } + parsed.servers.into_iter().collect() +} diff --git a/codex-rs/ext/mcp/src/lib_tests.rs b/codex-rs/ext/mcp/src/lib_tests.rs index 220aec1e6b14..7bdd3a594040 100644 --- a/codex-rs/ext/mcp/src/lib_tests.rs +++ b/codex-rs/ext/mcp/src/lib_tests.rs @@ -28,6 +28,7 @@ async fn hosted_plugin_runtime_forwards_thread_originator() -> Result<(), Box TestResult { + let codex_home = tempfile::tempdir()?; + let plugin_root = tempfile::tempdir()?; + std::fs::create_dir_all(plugin_root.path().join(".codex-plugin"))?; + std::fs::write( + plugin_root.path().join(".codex-plugin/plugin.json"), + r#"{"name":"demo","interface":{"displayName":"Demo"},"mcpServers":"./servers.json"}"#, + )?; + std::fs::write( + plugin_root.path().join("servers.json"), + r#"{"mcpServers":{"first":{"command":"first"},"second":{"command":"second"}}}"#, + )?; + let mut config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(codex_home.path().to_path_buf())) + .build() + .await?; + let existing = selected_plugin_contributions(&config, plugin_root.path()).await?; + config + .features + .enable(Feature::ExecutorCapabilityDiscovery) + .expect("test config should allow feature update"); + let high_level = selected_plugin_contributions(&config, plugin_root.path()).await?; + + assert_eq!(high_level, existing); + Ok(()) +} + async fn selected_plugin_contributions( config: &Config, plugin_root: &std::path::Path, @@ -181,10 +212,8 @@ async fn raw_selected_plugin_contributions( plugin_root: &std::path::Path, ) -> Result, Box> { let mut builder = ExtensionRegistryBuilder::new(); - codex_mcp_extension::install_executor_plugins( - &mut builder, - Arc::new(EnvironmentManager::default_for_tests()), - ); + let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); + codex_mcp_extension::install_executor_plugins(&mut builder, Arc::clone(&environment_manager)); let registry = builder.build(); let thread_init = ExtensionDataInit::new(); let selected_capability_roots = vec![SelectedCapabilityRoot { @@ -195,6 +224,18 @@ async fn raw_selected_plugin_contributions( }, }]; let thread_store = ExtensionData::new_with_init("test-thread", thread_init.clone()); + let executor_capability_discovery = if config + .features + .enabled(Feature::ExecutorCapabilityDiscovery) + { + Some( + ExecutorCapabilityDiscoveryCache::new(environment_manager) + .snapshot(&selected_capability_roots) + .await, + ) + } else { + None + }; Ok(registry.mcp_server_contributors()[0] .contribute(McpServerContributionContext::for_step( @@ -203,6 +244,7 @@ async fn raw_selected_plugin_contributions( &thread_store, "test_originator", &selected_capability_roots, + executor_capability_discovery.as_ref(), )) .await) } diff --git a/codex-rs/ext/skills/src/catalog.rs b/codex-rs/ext/skills/src/catalog.rs index 59977a47cac2..d6e30ed0e65e 100644 --- a/codex-rs/ext/skills/src/catalog.rs +++ b/codex-rs/ext/skills/src/catalog.rs @@ -1,5 +1,6 @@ use codex_core_skills::model::SkillDependencies; use codex_utils_path_uri::PathUri; +use std::sync::Arc; /// Source authority that owns a skill package and must be used to read it. #[derive(Clone, Debug, PartialEq, Eq, Hash)] @@ -83,6 +84,23 @@ impl SkillResourceId { environment_path: Some(EnvironmentSkillResource { environment_id: environment_id.into(), path, + contents: None, + }), + } + } + + pub fn environment_with_contents( + id: impl Into, + environment_id: impl Into, + path: PathUri, + contents: String, + ) -> Self { + Self { + id: id.into(), + environment_path: Some(EnvironmentSkillResource { + environment_id: environment_id.into(), + path, + contents: Some(contents.into()), }), } } @@ -96,12 +114,19 @@ impl SkillResourceId { .as_ref() .map(|resource| (resource.environment_id.as_str(), &resource.path)) } + + pub(crate) fn environment_contents(&self) -> Option<&str> { + self.environment_path + .as_ref() + .and_then(|resource| resource.contents.as_deref()) + } } #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct EnvironmentSkillResource { environment_id: String, path: PathUri, + contents: Option>, } /// Metadata shown in the always-visible skills catalog. diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index 2b0ca8307a85..115d1662e107 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -132,6 +132,7 @@ where include_bundled_skills: config.bundled_skills_enabled, include_orchestrator_skills: thread_state.orchestrator_skills_enabled(), mcp_resources: session_store.get::(), + executor_capability_discovery: None, }, &thread_state, ) @@ -169,6 +170,7 @@ where include_bundled_skills: config.bundled_skills_enabled, include_orchestrator_skills: false, mcp_resources: input.session_store.get::(), + executor_capability_discovery: input.executor_capability_discovery.cloned(), }, ) .await; @@ -285,6 +287,7 @@ where include_bundled_skills: config.bundled_skills_enabled, include_orchestrator_skills: thread_state.orchestrator_skills_enabled(), mcp_resources: session_store.get::(), + executor_capability_discovery: None, }; let host_query = query.clone(); let mut catalog = turn_store diff --git a/codex-rs/ext/skills/src/provider.rs b/codex-rs/ext/skills/src/provider.rs index 405592e93478..b4e174a33871 100644 --- a/codex-rs/ext/skills/src/provider.rs +++ b/codex-rs/ext/skills/src/provider.rs @@ -7,6 +7,7 @@ mod host; mod orchestrator; use codex_core_skills::HostSkillsSnapshot; +use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_mcp::McpResourceClient; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -31,6 +32,8 @@ pub struct SkillListQuery { pub include_bundled_skills: bool, pub include_orchestrator_skills: bool, pub mcp_resources: Option>, + /// Present only when the opt-in high-level executor discovery path is selected. + pub executor_capability_discovery: Option, } #[derive(Clone, Debug)] diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 9938035e9228..97ee7e9a4ed8 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use codex_core_skills::loader::EnvironmentSkillMetadata; +use codex_core_skills::loader::load_environment_skills_from_discovery; use codex_core_skills::loader::load_environment_skills_from_root; use codex_exec_server::EnvironmentManager; use codex_protocol::capabilities::CapabilityRootLocation; @@ -44,6 +45,9 @@ impl ExecutorSkillProvider { impl SkillProvider for ExecutorSkillProvider { fn list(&self, query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { Box::pin(async move { + if let Some(discovery) = query.executor_capability_discovery { + return Ok(self.list_from_discovery(&discovery)); + } let mut catalog = SkillCatalog::default(); for selected_root in query.executor_roots { let selected_root_id = selected_root.id; @@ -74,6 +78,7 @@ impl SkillProvider for ExecutorSkillProvider { authority.clone(), &selected_root_id, &environment_id, + /*instructions*/ None, )); } } @@ -95,6 +100,12 @@ impl SkillProvider for ExecutorSkillProvider { "executor skill resource does not match its package", )); } + if let Some(contents) = request.resource.environment_contents() { + return Ok(SkillReadResult { + resource: request.resource.clone(), + contents: contents.to_string(), + }); + } let Some((environment_id, resource_path)) = request.resource.environment_path() else { return Err(SkillProviderError::new( "executor skill resource is not bound to an environment", @@ -128,11 +139,50 @@ impl SkillProvider for ExecutorSkillProvider { } } +impl ExecutorSkillProvider { + fn list_from_discovery( + &self, + snapshot: &codex_exec_server::ExecutorCapabilityDiscoverySnapshot, + ) -> SkillCatalog { + let mut catalog = SkillCatalog::default(); + for root in snapshot.roots() { + let selected_root_id = &root.selected_root.id; + let CapabilityRootLocation::Environment { environment_id, .. } = + &root.selected_root.location; + let discovery = match &root.result { + Ok(discovery) => discovery.as_ref(), + Err(error) => { + catalog.warnings.push(format!( + "Selected capability root `{selected_root_id}` discovery failed: {error}" + )); + continue; + } + }; + let outcome = + load_environment_skills_from_discovery(discovery, self.restriction_product); + catalog.warnings.extend(outcome.warnings); + let authority = + SkillAuthority::new(SkillSourceKind::Executor, selected_root_id.clone()); + for skill in outcome.skills { + catalog.push_entry(catalog_entry_from_skill( + &skill.metadata, + authority.clone(), + selected_root_id, + environment_id, + Some(skill.instructions), + )); + } + } + catalog + } +} + fn catalog_entry_from_skill( skill: &EnvironmentSkillMetadata, authority: SkillAuthority, selected_root_id: &str, environment_id: &str, + instructions: Option, ) -> SkillCatalogEntry { let skill_path = skill.path_to_skills_md.inferred_native_path_string(); let normalized_path = match skill.path_to_skills_md.infer_path_convention() { @@ -143,16 +193,25 @@ fn catalog_entry_from_skill( "skill://{selected_root_id}/{}", normalized_path.trim_start_matches('/') ); + let main_prompt = match instructions { + Some(contents) => SkillResourceId::environment_with_contents( + display_path.clone(), + environment_id, + skill.path_to_skills_md.clone(), + contents, + ), + None => SkillResourceId::environment( + display_path.clone(), + environment_id, + skill.path_to_skills_md.clone(), + ), + }; let entry = SkillCatalogEntry::new( SkillPackageId(display_path.clone()), authority, skill.name.clone(), skill.description.clone(), - SkillResourceId::environment( - display_path.clone(), - environment_id, - skill.path_to_skills_md.clone(), - ), + main_prompt, ) .with_short_description(skill.short_description.clone()) .with_display_path(display_path) diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index db6fae949495..9f91f9ea2a65 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -30,6 +30,7 @@ pub(crate) struct SkillsThreadState { config: Mutex, orchestrator_skills_available: bool, executor_cache: Mutex>, + executor_discovery_cache: Mutex>, orchestrator_cache: Mutex>>, shadow_selection_turn: Mutex>, } @@ -40,6 +41,7 @@ impl SkillsThreadState { config: Mutex::new(config), orchestrator_skills_available, executor_cache: Mutex::new(Vec::new()), + executor_discovery_cache: Mutex::new(None), orchestrator_cache: Mutex::new(None), shadow_selection_turn: Mutex::new(None), } @@ -107,6 +109,11 @@ impl SkillsThreadState { providers: &SkillProviders, mut query: SkillListQuery, ) -> SkillCatalog { + if query.executor_capability_discovery.is_some() { + return self + .executor_discovery_catalog_snapshot(providers, query) + .await; + } let roots = std::mem::take(&mut query.executor_roots); let mut catalog = SkillCatalog::default(); for root in roots { @@ -119,6 +126,36 @@ impl SkillsThreadState { catalog } + async fn executor_discovery_catalog_snapshot( + &self, + providers: &SkillProviders, + query: SkillListQuery, + ) -> SkillCatalog { + if let Some(cached) = self + .executor_discovery_cache + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .filter(|cached| cached.roots == query.executor_roots) + { + return cached.catalog.clone(); + } + let roots = query.executor_roots.clone(); + let discovered = providers.list_executor_for_turn(query).await; + let mut cache = self + .executor_discovery_cache + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(cached) = cache.as_ref().filter(|cached| cached.roots == roots) { + return cached.catalog.clone(); + } + *cache = Some(CachedExecutorDiscoveryCatalog { + roots, + catalog: discovered.clone(), + }); + discovered + } + pub(crate) async fn orchestrator_catalog_snapshot( &self, mcp_resources: Option<&McpResourceClient>, @@ -236,6 +273,11 @@ struct CachedExecutorCatalog { catalog: SkillCatalog, } +struct CachedExecutorDiscoveryCatalog { + roots: Vec, + catalog: SkillCatalog, +} + struct OrchestratorGenerationCache { mcp_cache_key: Option, catalog: OnceCell, diff --git a/codex-rs/ext/skills/src/tools/mod.rs b/codex-rs/ext/skills/src/tools/mod.rs index 286b043cd8d2..53b07eb842c9 100644 --- a/codex-rs/ext/skills/src/tools/mod.rs +++ b/codex-rs/ext/skills/src/tools/mod.rs @@ -77,6 +77,7 @@ impl SkillToolContext { include_bundled_skills: false, include_orchestrator_skills: true, mcp_resources: self.mcp_resources.clone(), + executor_capability_discovery: None, }), ) .await diff --git a/codex-rs/ext/skills/tests/executor_file_system_authority.rs b/codex-rs/ext/skills/tests/executor_file_system_authority.rs index 83445c745c85..75c1ab0d59c0 100644 --- a/codex-rs/ext/skills/tests/executor_file_system_authority.rs +++ b/codex-rs/ext/skills/tests/executor_file_system_authority.rs @@ -10,6 +10,7 @@ use codex_core_skills::loader::load_skills_from_roots; use codex_exec_server::CopyOptions; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::EnvironmentManager; +use codex_exec_server::ExecutorCapabilityDiscoveryCache; use codex_exec_server::ExecutorFileSystem; use codex_exec_server::ExecutorFileSystemFuture; use codex_exec_server::FileMetadata; @@ -23,6 +24,7 @@ use codex_protocol::protocol::SkillScope; use codex_skills_extension::ExecutorSkillProvider; use codex_skills_extension::provider::SkillListQuery; use codex_skills_extension::provider::SkillProvider; +use codex_skills_extension::provider::SkillReadRequest; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; use pretty_assertions::assert_eq; @@ -267,6 +269,7 @@ async fn selected_root_id_distinguishes_identical_executor_paths() { include_bundled_skills: true, include_orchestrator_skills: false, mcp_resources: None, + executor_capability_discovery: None, }) .await .expect("list executor skills"); @@ -301,6 +304,57 @@ async fn selected_root_id_distinguishes_identical_executor_paths() { std::fs::remove_dir_all(test_root).expect("remove skill directory"); } +#[tokio::test] +async fn high_level_discovery_reuses_materialized_skill_contents_for_reads() { + let test_root = create_local_skill_root("materialized").expect("create local skill root"); + let manager = Arc::new(EnvironmentManager::default_for_tests()); + let provider = ExecutorSkillProvider::new_with_restriction_product( + Arc::clone(&manager), + /*restriction_product*/ None, + ); + let executor_roots = vec![SelectedCapabilityRoot { + id: "materialized-root".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: "local".to_string(), + path: PathUri::from_host_native_path(&test_root).expect("skill root URI"), + }, + }]; + let executor_capability_discovery = ExecutorCapabilityDiscoveryCache::new(manager) + .snapshot(&executor_roots) + .await; + let catalog = provider + .list(SkillListQuery { + turn_id: "turn-1".to_string(), + executor_roots, + host_snapshot: None, + include_host_skills: false, + include_bundled_skills: true, + include_orchestrator_skills: false, + mcp_resources: None, + executor_capability_discovery: Some(executor_capability_discovery), + }) + .await + .expect("list executor skills"); + let [entry] = catalog.entries.as_slice() else { + panic!("expected exactly one skill"); + }; + let request = SkillReadRequest { + authority: entry.authority.clone(), + package: entry.id.clone(), + resource: entry.main_prompt.clone(), + host_snapshot: None, + mcp_resources: None, + }; + + std::fs::remove_dir_all(&test_root).expect("remove skill directory after discovery"); + let read = provider + .read(request) + .await + .expect("read materialized executor skill"); + + assert_eq!(read.contents, SKILL_CONTENTS); +} + fn create_local_skill_root(label: &str) -> io::Result { let id = NEXT_TEST_ROOT_ID.fetch_add(1, Ordering::Relaxed); let test_root = std::env::temp_dir().join(format!( diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index d9c3d4d746c9..4e139f5959ee 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -208,6 +208,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach turn_id: "turn-1", environments: std::slice::from_ref(&turn_environment), ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, session_store: &session_store, thread_store: &thread_store, turn_store: &turn_store, @@ -260,6 +261,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach turn_id: "turn-2", environments: &[], ready_selected_capability_roots: &[], + executor_capability_discovery: None, session_store: &session_store, thread_store: &thread_store, turn_store: &unavailable_turn_store, @@ -282,6 +284,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach turn_id: "turn-3", environments: &[turn_environment], ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, session_store: &session_store, thread_store: &thread_store, turn_store: &restored_turn_store, @@ -309,6 +312,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach turn_id: "turn-4", environments: &[], ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, session_store: &session_store, thread_store: &thread_store, turn_store: &listing_disabled_turn_store, @@ -599,6 +603,7 @@ async fn root_qualified_locator_selects_only_the_matching_executor_skill() -> Te workspace_roots: Vec::new(), }], ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, session_store: &session_store, thread_store: &thread_store, turn_store: &turn_store, diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 6c9b9fb64644..0139f17edf6f 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -170,6 +170,8 @@ pub enum Feature { ToolSuggest, /// Enable plugins. Plugins, + /// Discover selected-root plugin and skill manifests through one high-level exec-server RPC. + ExecutorCapabilityDiscovery, /// Removed compatibility flag for plugin-bundled lifecycle hooks. PluginHooks, /// Allow the in-app browser pane in desktop apps. @@ -1116,6 +1118,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::Stable, default_enabled: true, }, + FeatureSpec { + id: Feature::ExecutorCapabilityDiscovery, + key: "executor_capability_discovery", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::PluginHooks, key: "plugin_hooks", diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index f622b5e2c66a..6e028fa98444 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -27,6 +27,19 @@ fn under_development_features_are_disabled_by_default() { } } +#[test] +fn executor_capability_discovery_is_an_opt_in_map_feature() { + let mut features = Features::with_defaults(); + assert!(!features.enabled(Feature::ExecutorCapabilityDiscovery)); + + features.apply_map(&BTreeMap::from([( + "executor_capability_discovery".to_string(), + true, + )])); + + assert!(features.enabled(Feature::ExecutorCapabilityDiscovery)); +} + #[test] fn default_enabled_features_are_stable() { for spec in crate::FEATURES { diff --git a/codex-rs/utils/plugins/Cargo.toml b/codex-rs/utils/plugins/Cargo.toml index cdfb600c8c20..1a4576d59d0f 100644 --- a/codex-rs/utils/plugins/Cargo.toml +++ b/codex-rs/utils/plugins/Cargo.toml @@ -15,6 +15,7 @@ workspace = true [dependencies] codex-exec-server = { workspace = true } +codex-exec-server-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-path-uri = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/codex-rs/utils/plugins/src/lib.rs b/codex-rs/utils/plugins/src/lib.rs index 5cb590549545..6b203955b2bf 100644 --- a/codex-rs/utils/plugins/src/lib.rs +++ b/codex-rs/utils/plugins/src/lib.rs @@ -7,7 +7,7 @@ pub mod mcp_connector; pub mod mention_syntax; pub mod plugin_namespace; -pub use plugin_namespace::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; +pub use codex_exec_server_protocol::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; pub use plugin_namespace::find_plugin_manifest_path; pub use plugin_namespace::plugin_namespace_for_root_uri; pub use plugin_namespace::plugin_namespace_for_skill_path; diff --git a/codex-rs/utils/plugins/src/plugin_namespace.rs b/codex-rs/utils/plugins/src/plugin_namespace.rs index 6a949cae1f85..1407c93b2c1f 100644 --- a/codex-rs/utils/plugins/src/plugin_namespace.rs +++ b/codex-rs/utils/plugins/src/plugin_namespace.rs @@ -1,18 +1,12 @@ //! Resolve plugin namespace from skill file paths by walking ancestors for `plugin.json`. use codex_exec_server::ExecutorFileSystem; +use codex_exec_server_protocol::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; use std::path::Path; use std::path::PathBuf; -/// Ordered plugin manifest paths recognized beneath a plugin root. -pub const DISCOVERABLE_PLUGIN_MANIFEST_PATHS: &[&str] = &[ - ".codex-plugin/plugin.json", - ".claude-plugin/plugin.json", - ".cursor-plugin/plugin.json", -]; - pub fn find_plugin_manifest_path(plugin_root: &Path) -> Option { DISCOVERABLE_PLUGIN_MANIFEST_PATHS .iter()