diff --git a/codex-rs/core-plugins/src/provider.rs b/codex-rs/core-plugins/src/provider.rs index e59a9311e6..3ebabcc961 100644 --- a/codex-rs/core-plugins/src/provider.rs +++ b/codex-rs/core-plugins/src/provider.rs @@ -1,19 +1,22 @@ -//! Plugin catalog listing interface and the MCP context used to request snapshots. +//! Plugin catalog listing interface and source inputs used to request snapshots. use crate::ExecutorPluginProviderError; use crate::PluginCatalog; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_mcp::McpResourceClient; use std::future::Future; use std::pin::Pin; use std::sync::Arc; use thiserror::Error; -/// Turn identifiers and MCP client used to request one catalog snapshot. +/// Identifiers, MCP client, and executor inventories used to request one catalog snapshot. #[derive(Clone, Debug)] pub struct PluginListQuery { pub thread_id: String, pub turn_id: String, pub mcp_resources: Option>, + /// Already-acquired V2 inventories; empty for cloud listing. + pub snapshots: Vec, } /// Failure to list plugin metadata. diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index fc22ace946..9e8e684394 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -925,6 +925,9 @@ "executor_capability_discovery": { "type": "boolean" }, + "executor_capability_discovery_v2": { + "type": "boolean" + }, "experimental_use_unified_exec_tool": { "type": "boolean" }, @@ -7253,6 +7256,9 @@ "executor_capability_discovery": { "type": "boolean" }, + "executor_capability_discovery_v2": { + "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 0eb7bdc3ad..c28e7b86f7 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -9,6 +9,7 @@ use codex_connectors::ConnectorRuntimeManager; use codex_connectors::PluginConnectorSource; use codex_core_plugins::PluginsManager; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; @@ -156,6 +157,7 @@ impl McpManager { .config } + #[allow(clippy::too_many_arguments)] #[tracing::instrument(name = "mcp.runtime_config.project_for_step", skip_all)] pub(crate) async fn runtime_config_for_step( &self, @@ -165,6 +167,7 @@ impl McpManager { identity: McpThreadIdentity<'_>, ready_selected_capability_roots: &[SelectedCapabilityRoot], executor_capability_discovery: Option<&ExecutorCapabilityDiscoverySnapshot>, + executor_capability_discovery_v2: Option<&[ExecutorCapabilityDiscoveryV2Snapshot]>, ) -> McpRuntimeProjection { let mut context = McpServerContributionContext::for_step( config, @@ -173,6 +176,7 @@ impl McpManager { identity.originator, ready_selected_capability_roots, executor_capability_discovery, + executor_capability_discovery_v2, ) .with_session_source(identity.session_source) .with_auth_changed(identity.auth_changed); diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index d2599a4d5c..4b00bfd292 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -3,8 +3,11 @@ use super::*; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::environment_selection::combine_selected_capability_roots; use crate::mcp_tool_call::McpToolApprovalContext; +use codex_exec_server::DiscoverV2CapabilitiesRequest; use codex_exec_server::ExecutorCapabilityDiscoveryCache; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Cache; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::MAX_SELECTED_CAPABILITY_ROOTS; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; @@ -122,6 +125,9 @@ impl Session { .await; let ready_selected_capability_roots = Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery_v2 = self + .executor_capability_discovery_for_step_v2(config, &environments) + .await; let executor_capability_discovery = self .executor_capability_discovery_for_step( config, @@ -147,6 +153,7 @@ impl Session { }, &ready_selected_capability_roots, executor_capability_discovery.as_deref(), + executor_capability_discovery_v2.as_deref(), ) .await; let mcp_config = self @@ -229,6 +236,9 @@ impl Session { .await; let ready_selected_capability_roots = Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery_v2 = self + .executor_capability_discovery_for_step_v2(&desired.config, &desired.environments) + .await; let executor_capability_discovery = self .executor_capability_discovery_for_step( &desired.config, @@ -255,6 +265,7 @@ impl Session { }, &ready_selected_capability_roots, executor_capability_discovery.as_deref(), + executor_capability_discovery_v2.as_deref(), ) .await; self.publish_mcp_runtime( @@ -302,6 +313,9 @@ impl Session { .await; let ready_selected_capability_roots = Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery_v2 = self + .executor_capability_discovery_for_step_v2(&desired.config, &desired.environments) + .await; let executor_capability_discovery = self .executor_capability_discovery_for_step( &desired.config, @@ -328,6 +342,7 @@ impl Session { }, &ready_selected_capability_roots, executor_capability_discovery.as_deref(), + executor_capability_discovery_v2.as_deref(), ) .await; let mcp_projection = self @@ -499,6 +514,66 @@ impl Session { Some(Arc::new(discovery)) } + /// Acquires V2 inventories from the thread's current cache. + pub(super) async fn executor_capability_discovery_for_step_v2( + &self, + config: &Config, + environments: &TurnEnvironmentSnapshot, + ) -> Option> { + if !config + .features + .enabled(Feature::ExecutorCapabilityDiscoveryV2) + { + return None; + } + // Retain this cache if a new turn replaces the thread's current cache. + let cache = self + .services + .thread_extension_data + .get_or_init(ExecutorCapabilityDiscoveryV2Cache::default); + let remote_environment_count = environments + .turn_environments() + .filter(|selected| selected.environment.is_remote()) + .count(); + cache.ensure_capacity(remote_environment_count); + let mut snapshots = Vec::new(); + for selected in environments.turn_environments() { + if !selected.environment.is_remote() { + continue; + } + let environment_id = &selected.selection.environment_id; + let info = match selected.environment.info().await { + Ok(info) => info, + Err(_) => { + tracing::warn!( + environment_id, + "could not read executor discovery V2 support" + ); + continue; + } + }; + if !info.capabilities.capability_discovery_v2 { + continue; + } + let request = DiscoverV2CapabilitiesRequest { + cwd: selected.cwd().clone(), + sandbox: Some(selected.sandbox_context(/*additional_permissions*/ None)), + }; + let result = cache + .discover(environment_id, &selected.environment, request) + .await; + snapshots.push(ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: environment_id.clone(), + result, + }); + } + if cache.take_recovered_discovery() { + // Refresh MCP after discovery recovers. + self.mark_mcp_runtime_dirty(); + } + Some(snapshots) + } + pub(crate) async fn resolve_selected_capability_roots_for_step( &self, environments: &TurnEnvironmentSnapshot, @@ -715,6 +790,9 @@ impl Session { let environments = self.services.turn_environments.snapshot().await; let environment_selections = environments.all_selections(); let desired = self.latest_mcp_desired_state(auth, environments).await; + let executor_capability_discovery_v2 = self + .executor_capability_discovery_for_step_v2(&desired.config, &desired.environments) + .await; let executor_capability_discovery = self .executor_capability_discovery_for_step( &desired.config, @@ -741,6 +819,7 @@ impl Session { }, &ready_selected_capability_roots, executor_capability_discovery.as_deref(), + executor_capability_discovery_v2.as_deref(), ) .await; self.publish_mcp_runtime( diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 0424f1adca..2cb1657a5d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3923,6 +3923,9 @@ impl Session { .await; let ready_selected_capability_roots = Self::ready_selected_capability_roots(&selected_capability_roots); + let executor_capability_discovery_v2 = self + .executor_capability_discovery_for_step_v2(&turn_context.config, &environments) + .await; let executor_capability_discovery = self .executor_capability_discovery_for_step( &turn_context.config, @@ -3932,6 +3935,9 @@ impl Session { .await; let extension_data = codex_extension_api::ExtensionData::new(turn_context.sub_id.clone()); + if let Some(snapshots) = &executor_capability_discovery_v2 { + extension_data.insert(snapshots.clone()); + } if let Some(messages) = settings .model_info .model_messages @@ -3969,6 +3975,7 @@ impl Session { &turn_context.originator, &ready_selected_capability_roots, executor_capability_discovery.as_deref(), + executor_capability_discovery_v2.as_deref(), ) .with_session_source(&turn_context.session_source) .with_selected_environments(&environments.all_selections()), @@ -3990,6 +3997,7 @@ impl Session { Ok::<_, CodexErr>(( selected_capability_roots, executor_capability_discovery, + executor_capability_discovery_v2, mcp, tool_router, extension_data, @@ -4005,6 +4013,7 @@ impl Session { let ( selected_capability_roots, executor_capability_discovery, + executor_capability_discovery_v2, mcp, tool_router, extension_data, @@ -4024,6 +4033,7 @@ impl Session { selected_capability_roots, executor_capability_discovery, extension_data, + executor_capability_discovery_v2, mcp, tool_router, loaded_agents_md, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index a617d9367b..725f213d2b 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1208,6 +1208,7 @@ impl Session { }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .await; (auth, mcp_projection) @@ -1931,6 +1932,7 @@ impl Session { }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .await } else { diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index a426437014..d90a02044e 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/codex-rs/core/src/session/step_context.rs @@ -11,6 +11,7 @@ use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnEnvironment; use crate::tools::router::ToolRouter; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_extension_api::ExtensionData; use codex_file_system::EnvironmentAccess; @@ -40,6 +41,7 @@ pub(crate) struct StepContext { pub(crate) executor_capability_discovery: Option>, /// Keeps the extension inputs used to build this step's tools for its World State as well. pub(crate) extension_data: ExtensionData, + pub(crate) executor_capability_discovery_v2: Option>, /// The exact MCP connections, configuration, and catalog captured for this step. pub(crate) mcp: Arc, /// The finalized tool plan advertised and executed for this exact sampling request. diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 2cde379539..e0370838c6 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -292,6 +292,7 @@ impl StepContext { selected_capability_roots: Vec::new(), executor_capability_discovery: None, extension_data: codex_extension_api::ExtensionData::new(turn.sub_id.clone()), + executor_capability_discovery_v2: None, mcp: Arc::new(codex_mcp::McpBinding::empty(mcp_config_for_test( &turn.config, ))), diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index 37dda837fc..9d59defb79 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -335,6 +335,9 @@ impl Session { for contributor in self.services.extensions.context_contributors() { for section in contributor .contribute_world_state(WorldStateContributionInput { + executor_capability_discovery_v2: step_context + .executor_capability_discovery_v2 + .as_deref(), thread_id: self.thread_id(), turn_id: turn_context.sub_id.as_str(), model_info: &step_context.settings.model_info, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 5612045b97..6a3e95ee80 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -9,6 +9,7 @@ use std::time::Duration; use std::time::Instant; use codex_diagnostics::Gauge; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Cache; use codex_extension_api::ThreadIdleCause; use futures::future::BoxFuture; use tokio::select; @@ -290,6 +291,18 @@ impl Session { input: Vec, task: T, ) { + // Replace the cache; in-flight acquisitions retain their previous Arc. + self.services + .thread_extension_data + .insert(ExecutorCapabilityDiscoveryV2Cache::default()); + if turn_context + .config + .features + .enabled(Feature::ExecutorCapabilityDiscoveryV2) + { + // MCP contributors must project the new inventory even if their other inputs match. + self.request_mcp_runtime_reprojection(); + } self.activate_plugin_selection(&turn_context).await; // Inherited or recovered roots are applied before task start. Otherwise this // task owns its turn, including background work. Later mail cannot change it. diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index df585fa7d9..0c2d6dc6c2 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1779,6 +1779,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .await; let second_session = &second_thread.thread.session; @@ -1801,6 +1802,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .await; @@ -1917,6 +1919,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .await; assert_eq!( diff --git a/codex-rs/core/tests/suite/executor_discovery_v2.rs b/codex-rs/core/tests/suite/executor_discovery_v2.rs new file mode 100644 index 0000000000..f8e278ec56 --- /dev/null +++ b/codex-rs/core/tests/suite/executor_discovery_v2.rs @@ -0,0 +1,596 @@ +//! Verifies gated, inert V2 acquisition and cache reuse across Core turns and idle callers. + +use anyhow::Context; +use anyhow::Result; +use codex_core::config::Config; +use codex_core::windows_sandbox::WindowsSandboxLevelExt; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; +use codex_exec_server::HttpRequestBodyDeltaNotification; +use codex_exec_server::HttpRequestParams; +use codex_extension_api::ExtensionFuture; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::McpServerContribution; +use codex_extension_api::McpServerContributionContext; +use codex_extension_api::McpServerContributor; +use codex_features::Feature; +use codex_protocol::config_types::WindowsSandboxLevel; +use core_test_support::responses; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use futures::SinkExt; +use futures::StreamExt; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use tokio::net::TcpListener; +use tokio::sync::Mutex; +use tokio_tungstenite::accept_async; +use tokio_tungstenite::tungstenite::Message; +use tokio_util::task::AbortOnDropHandle; + +struct DiscoveryRecorder { + snapshots: Arc>>, +} + +impl McpServerContributor for DiscoveryRecorder { + fn id(&self) -> &'static str { + "discovery-recorder" + } + + fn contribute<'a>( + &'a self, + context: McpServerContributionContext<'a, Config>, + ) -> ExtensionFuture<'a, Vec> { + Box::pin(async move { + let selected_roots = context + .ready_selected_capability_roots() + .unwrap_or_default(); + if !selected_roots.is_empty() { + let legacy = context.executor_capability_discovery(); + assert_eq!( + legacy.map(|snapshot| snapshot.roots().len()), + Some(selected_roots.len()) + ); + assert!(legacy.is_some_and(|snapshot| { + snapshot.roots().iter().all(|root| root.result.is_ok()) + })); + } + if let Some(snapshots) = context.executor_capability_discovery_v2() { + self.snapshots.lock().await.extend_from_slice(snapshots); + } + Vec::new() + }) + } +} + +struct Executor { + url: String, + requests: Arc>>, + _task: AbortOnDropHandle>, +} + +#[derive(Clone, Copy)] +enum DiscoveryFailure { + None, + Server, + Disconnect, +} + +impl Executor { + async fn start(supports_v2: bool, failure: DiscoveryFailure) -> Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("ws://{}", listener.local_addr()?); + let requests = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&requests); + let discovery_attempts = Arc::new(AtomicUsize::new(0)); + let http = codex_exec_server::Environment::create_for_tests(/*exec_server_url*/ None)? + .get_http_client(); + let task = tokio::spawn(async move { + let mut connections = tokio::task::JoinSet::new(); + while let Ok((stream, _)) = listener.accept().await { + let recorded = Arc::clone(&recorded); + let discovery_attempts = Arc::clone(&discovery_attempts); + let http = Arc::clone(&http); + connections.spawn(async move { + let mut socket = accept_async(stream).await?; + while let Some(Ok(message)) = socket.next().await { + let request: Value = match message { + Message::Text(text) => serde_json::from_str(&text)?, + Message::Binary(bytes) => serde_json::from_slice(&bytes)?, + Message::Close(_) => break, + _ => continue, + }; + if request.get("id").is_none() { continue; } + recorded.lock().await.push(request.clone()); + if request["method"] == "http/request" { + let params: HttpRequestParams = serde_json::from_value(request["params"].clone())?; + let request_id = params.request_id.clone(); + let streaming = params.stream_response; + let mut response = http.http_request(params).await?; + let delta = streaming.then(|| HttpRequestBodyDeltaNotification { + request_id, seq: 1, + delta: std::mem::replace(&mut response.body, Vec::new().into()), + done: true, error: None, + }); + socket.send(Message::Text(json!({"id":request["id"],"result":response}).to_string().into())).await?; + if let Some(delta) = delta { + socket.send(Message::Text(json!({"method":"http/request/bodyDelta","params":delta}).to_string().into())).await?; + } + continue; + } + if request["method"] == "capabilities/discoverV2" { + let attempt = discovery_attempts.fetch_add(1, Ordering::Relaxed); + match failure { + // Exhaust the environment's one transport replay so the cache sees Err. + DiscoveryFailure::Disconnect if attempt < 2 => break, + DiscoveryFailure::Server if attempt == 0 => { + socket.send(Message::Text(json!({"id":request["id"],"error":{"code":-32004,"message":"fixture discovery failure"}}).to_string().into())).await?; + continue; + } + DiscoveryFailure::None | DiscoveryFailure::Server | DiscoveryFailure::Disconnect => {} + } + } + let result = match request["method"].as_str() { + Some("initialize") => Some(json!({"sessionId":"acquisition-only", "environmentInfo": { + "shell":{"name":"bash","path":"/bin/bash"}, + "capabilities":{"sandboxedFileStreaming":true,"capabilityDiscoveryV2":supports_v2} + }})), + Some("environment/info") => Some(json!({ + "shell":{"name":"bash","path":"/bin/bash"}, + "capabilities":{"sandboxedFileStreaming":true,"capabilityDiscoveryV2":supports_v2} + })), + Some("capabilityRoots/discoverV1") => Some(json!({ + "roots": request["params"]["roots"].as_array().context("V1 roots")? + .iter().map(|root| json!({"id":root["id"],"path":root["path"]})).collect::>() + })), + Some("capabilities/discoverV2") => Some(json!({ + "plugins":[{"id":"inert-sentinel@market","remotePluginId":null, + "version":"1","root":"file:///inert-sentinel", + "files":{"manifest":{"path":"file:///inert-sentinel/.codex-plugin/plugin.json", + "contents":"{\"name\":\"inert-sentinel\",\"mcpServers\":{\"inert-sentinel\":{\"command\":\"never-run\"}}}"}, + "mcpConfig":null,"appsConfig":null},"skills":[]}], + "skills":[{"name":"inert-sentinel","description":"must not enter the prompt", + "shortDescription":null,"path":"file:///inert-sentinel/SKILL.md","metadata":null}], + "warnings":[] + })), + Some("fs/canonicalize") => Some(json!({"path":request["params"]["path"]})), + Some("fs/walk") => Some(json!({"entries":[],"errors":[],"truncated":false})), + _ => None, + }; + let response = match result { + Some(result) => json!({"id":request["id"],"result":result}), + None => json!({"id":request["id"],"error":{"code":-32004,"message":"fixture discovery failure"}}), + }; + if socket.send(Message::Text(response.to_string().into())).await.is_err() { break; } + } + anyhow::Ok(()) + }); + } + anyhow::Ok(()) + }); + Ok(Self { + url, + requests, + _task: AbortOnDropHandle::new(task), + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn flag_off_never_requests_v2() -> Result<()> { + check_acquisition(/*enabled*/ false, /*supported*/ true).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn v2_inventory_reaches_mcp_projection_without_loading_skills() -> Result<()> { + check_acquisition(/*enabled*/ true, /*supported*/ true).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unsupported_executor_keeps_existing_behavior() -> Result<()> { + check_acquisition(/*enabled*/ true, /*supported*/ false).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovery_failures_retry_or_remain_cached_by_error_kind() -> Result<()> { + skip_if_no_network!(Ok(())); + for failure in [DiscoveryFailure::Server, DiscoveryFailure::Disconnect] { + let executor = Executor::start(/*supports_v2*/ true, failure).await?; + let manager = codex_exec_server::EnvironmentManager::create_for_tests( + Some(executor.url.clone()), + /*local_runtime_paths*/ None, + ) + .await; + let environment = manager + .default_environment() + .context("remote environment")?; + let cache = codex_exec_server::ExecutorCapabilityDiscoveryV2Cache::default(); + let request = codex_exec_server::DiscoverV2CapabilitiesRequest { + cwd: codex_utils_path_uri::PathUri::parse("file:///repo")?, + sandbox: None, + }; + assert!( + cache + .discover("executor", &environment, request.clone()) + .await + .is_err() + ); + assert!(!cache.take_recovered_discovery()); + let (first, concurrent) = tokio::join!( + cache.discover("executor", &environment, request.clone()), + cache.discover("executor", &environment, request.clone()), + ); + let expected_calls = match failure { + DiscoveryFailure::Server => { + assert!(first.is_err()); + assert!(concurrent.is_err()); + let next_turn = codex_exec_server::ExecutorCapabilityDiscoveryV2Cache::default(); + next_turn + .discover("executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(!next_turn.take_recovered_discovery()); + assert!(!cache.take_recovered_discovery()); + 2 + } + DiscoveryFailure::Disconnect => { + let first = first.map_err(anyhow::Error::msg)?; + let concurrent = concurrent.map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&first, &concurrent)); + assert!(cache.take_recovered_discovery()); + assert!(!cache.take_recovered_discovery()); + let reused = cache + .discover("executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&first, &reused)); + assert!(!cache.take_recovered_discovery()); + // Two disconnected attempts, then one shared successful retry in the same cache. + 3 + } + DiscoveryFailure::None => unreachable!("the test only exercises failures"), + }; + assert_eq!( + executor + .requests + .lock() + .await + .iter() + .filter(|request| request["method"] == "capabilities/discoverV2") + .count(), + expected_calls, + ); + } + Ok(()) +} + +async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> { + skip_if_no_network!(Ok(())); + let executor = Executor::start(supported, DiscoveryFailure::None).await?; + let model = responses::start_mock_server().await; + let observed = Arc::new(Mutex::new(Vec::new())); + let mut extensions = ExtensionRegistryBuilder::new(); + let recorder = Arc::new(DiscoveryRecorder { + snapshots: Arc::clone(&observed), + }); + extensions.mcp_server_contributor(recorder); + let test = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_exec_server_url(executor.url.clone()) + .with_config(move |config| { + config.update_plan_enabled = true; + assert!( + config + .features + .enable(Feature::ExecutorCapabilityDiscovery) + .is_ok() + ); + if enabled { + assert!( + config + .features + .enable(Feature::ExecutorCapabilityDiscoveryV2) + .is_ok() + ); + } + }) + .build_with_auto_env(&model) + .await?; + let calls = || async { + executor + .requests + .lock() + .await + .iter() + .filter(|request| request["method"] == "capabilities/discoverV2") + .count() + }; + let acquires_v2 = enabled && supported; + if acquires_v2 { + observed.lock().await.clear(); + let _ = test.codex.current_mcp_config_and_runtime_context().await; + let initial_calls = calls().await; + let _ = test.codex.current_mcp_config_and_runtime_context().await; + assert!(initial_calls > 0); + assert_eq!( + calls().await, + initial_calls, + "idle acquisitions reuse the cache" + ); + let observed = observed.lock().await; + assert!(!observed.is_empty()); + assert!(observed.iter().all(|snapshot| snapshot.result.is_ok())); + } + let initial_calls = calls().await; + let mut selection = test.executor_environment().selection().clone(); + selection.environment_id = codex_exec_server::REMOTE_ENVIRONMENT_ID.to_string(); + let cwd = selection.cwd.clone(); + selection.workspace_roots = vec![cwd.clone()]; + selection.config = + codex_protocol::protocol::EnvironmentConfigState::Ready(codex_core::EnvironmentConfig { + allow_login_shell: false, + workspace_roots: selection.workspace_roots.clone(), + permission_profile: codex_protocol::models::PermissionProfileSnapshot::legacy( + test.config.permissions.permission_profile().clone(), + ), + shell_environment_policy: Default::default(), + windows_sandbox_level: WindowsSandboxLevel::from_config(&test.config), + windows_sandbox_type: test.config.permissions.windows_sandbox_type, + use_legacy_landlock: test.config.features.use_legacy_landlock(), + exec_policy: None, + mcp_policy: None, + network_policy: None, + selected_capability_roots: vec![codex_protocol::capabilities::SelectedCapabilityRoot { + id: "legacy-root".to_string(), + location: codex_protocol::capabilities::CapabilityRootLocation::Environment { + environment_id: selection.environment_id.clone(), + path: cwd.clone(), + }, + }], + }); + // Only the successful lifecycle case needs two turns and multiple model steps. + let turns = if acquires_v2 { 2 } else { 1 }; + let mut previous_inventory = None; + for turn in 0..turns { + observed.lock().await.clear(); + let before_turn = calls().await; + let mut bodies = Vec::new(); + if acquires_v2 { + bodies.push(responses::sse(vec![ + responses::ev_response_created("plan"), + responses::ev_function_call( + "plan-call", + "update_plan", + r#"{"plan":[{"step":"Say hello","status":"completed"}]}"#, + ), + responses::ev_completed("plan"), + ])); + } + bodies.push(responses::sse(vec![ + responses::ev_response_created("response"), + responses::ev_completed("response"), + ])); + let mut responses = Vec::new(); + for body in bodies { + responses.push(responses::mount_sse_once(&model, body).await); + } + test.submit_turn_with_environments("Say hello.", Some(vec![selection.clone()])) + .await?; + let model_requests = responses + .iter() + .flat_map(core_test_support::responses::ResponseMock::requests) + .collect::>(); + assert_eq!(model_requests.len(), if acquires_v2 { 2 } else { 1 }); + for request in model_requests { + assert!(!request.body_json().to_string().contains("inert-sentinel")); + } + let turn_calls = calls().await; + if acquires_v2 { + assert!( + turn_calls > before_turn, + "a new turn must acquire fresh inventory" + ); + if turn > 0 { + assert_eq!( + turn_calls, + before_turn + 1, + "steps share the new turn's cache" + ); + } + let observed = observed.lock().await; + let inventory = observed + .last() + .context("ordinary turn must project its V2 inventory into MCP")? + .result + .as_ref() + .map_err(|error| anyhow::anyhow!("{error}"))?; + assert_eq!( + inventory.plugins.first().context("inert plugin")?.id, + "inert-sentinel@market" + ); + if let Some(previous) = &previous_inventory { + assert!( + !Arc::ptr_eq(previous, inventory), + "MCP contributors must receive the new turn's inventory" + ); + } + previous_inventory = Some(Arc::clone(inventory)); + } else { + assert_eq!(turn_calls, 0); + assert!(observed.lock().await.is_empty()); + } + } + let requests = executor.requests.lock().await.clone(); + assert!( + requests + .iter() + .any(|request| request["method"] == "capabilityRoots/discoverV1") + ); + let discovery = requests + .iter() + .filter(|request| request["method"] == "capabilities/discoverV2") + .collect::>(); + for request in discovery.iter().skip(initial_calls) { + assert_eq!(request["params"]["cwd"], json!(cwd)); + } + for request in discovery { + assert!(request["params"]["sandbox"].is_object()); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovery_cache_coalesces_requests_and_separates_cwds_sandboxes_and_executors() +-> Result<()> { + skip_if_no_network!(Ok(())); + let root = tempfile::tempdir()?; + let executor = Executor::start(/*supports_v2*/ true, DiscoveryFailure::None).await?; + let manager = codex_exec_server::EnvironmentManager::create_for_tests( + Some(executor.url.clone()), + /*local_runtime_paths*/ None, + ) + .await; + let environment = manager + .default_environment() + .context("remote environment")?; + let cache = codex_exec_server::ExecutorCapabilityDiscoveryV2Cache::default(); + let request = codex_exec_server::DiscoverV2CapabilitiesRequest { + cwd: codex_utils_path_uri::PathUri::from_host_native_path(root.path())?, + sandbox: None, + }; + let calls = || async { + executor + .requests + .lock() + .await + .iter() + .filter(|request| request["method"] == "capabilities/discoverV2") + .count() + }; + assert_eq!( + calls().await, + 0, + "creating a cache must not request discovery" + ); + let (first, concurrent) = tokio::join!( + cache.discover("executor", &environment, request.clone()), + cache.discover("executor", &environment, request.clone()), + ); + let first = first.map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq( + &first, + &concurrent.map_err(anyhow::Error::msg)? + )); + assert_eq!(calls().await, 1); + let restricted = codex_exec_server::DiscoverV2CapabilitiesRequest { + sandbox: Some( + codex_exec_server::FileSystemSandboxContext::from_legacy_sandbox_policy( + codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), + request.cwd.clone(), + )?, + ), + ..request.clone() + }; + cache + .discover("executor", &environment, restricted) + .await + .map_err(anyhow::Error::msg)?; + let original = cache + .discover("executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&first, &original)); + assert_eq!(calls().await, 2); + cache + .discover( + "executor", + &environment, + codex_exec_server::DiscoverV2CapabilitiesRequest { + cwd: request.cwd.join("other")?, + ..request.clone() + }, + ) + .await + .map_err(anyhow::Error::msg)?; + assert_eq!( + calls().await, + 3, + "another cwd must not reuse the same inventory" + ); + let other = cache + .discover("other-executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&first, &other)); + assert_eq!(calls().await, 3); + let replacement_manager = codex_exec_server::EnvironmentManager::create_for_tests( + Some(executor.url.clone()), + /*local_runtime_paths*/ None, + ) + .await; + let replacement = replacement_manager + .default_environment() + .context("replacement environment")?; + cache + .discover("executor", &replacement, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert_eq!(calls().await, 4); + let other_reused = cache + .discover("other-executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&other, &other_reused)); + assert_eq!( + calls().await, + 4, + "a new environment instance must not invalidate the original instance" + ); + + // Four entries remain; reserve enough capacity for a selection larger than 16. + cache.ensure_capacity(/*minimum*/ 17); + for index in 0..13 { + cache + .discover( + "executor", + &environment, + codex_exec_server::DiscoverV2CapabilitiesRequest { + cwd: request.cwd.join(&format!("context-{index}"))?, + ..request.clone() + }, + ) + .await + .map_err(anyhow::Error::msg)?; + } + let oldest = cache + .discover("other-executor", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&other, &oldest)); + assert_eq!(calls().await, 17); + + cache + .discover( + "executor", + &environment, + codex_exec_server::DiscoverV2CapabilitiesRequest { + cwd: request.cwd.join("overflow")?, + ..request.clone() + }, + ) + .await + .map_err(anyhow::Error::msg)?; + // Reading the oldest entry did not promote it: eviction is FIFO, not LRU. + let reloaded = cache + .discover("other-executor", &environment, request) + .await + .map_err(anyhow::Error::msg)?; + assert_eq!(calls().await, 19); + assert!(!Arc::ptr_eq(&other, &reloaded)); + assert_eq!(other.as_ref(), reloaded.as_ref()); + + Ok(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index dccb224dbc..bc2f075304 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -257,5 +257,6 @@ mod window_headers; mod workspace_roots; mod worktree_trust; +mod executor_discovery_v2; #[path = "guardian_sender_messages_tests.rs"] mod guardian_sender_messages; diff --git a/codex-rs/exec-server-protocol/src/capabilities.rs b/codex-rs/exec-server-protocol/src/capabilities.rs index 2583b9f25f..3adb4538f8 100644 --- a/codex-rs/exec-server-protocol/src/capabilities.rs +++ b/codex-rs/exec-server-protocol/src/capabilities.rs @@ -5,13 +5,21 @@ use codex_utils_path_uri::PathUri; use serde::Deserialize; use serde::Serialize; use std::fmt; +use std::sync::Arc; use crate::CapabilityTextFile; use crate::DiscoveredPluginFiles; pub const CAPABILITIES_DISCOVER_V2_METHOD: &str = "capabilities/discoverV2"; -/// Discovers installed capabilities on the executor. +/// Caller-bound discovery result shared by capability consumers within a model step. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExecutorCapabilityDiscoveryV2Snapshot { + pub environment_id: String, + pub result: Result, String>, +} + +/// Discover installed capabilities for a thread, without client-supplied installation paths. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct DiscoverV2CapabilitiesRequest { diff --git a/codex-rs/exec-server/src/capability_discovery_v2_cache.rs b/codex-rs/exec-server/src/capability_discovery_v2_cache.rs new file mode 100644 index 0000000000..c8e7d6eda1 --- /dev/null +++ b/codex-rs/exec-server/src/capability_discovery_v2_cache.rs @@ -0,0 +1,121 @@ +//! Bounded lazy V2 responses, cached per environment, executor session, cwd, and sandbox. +//! Reports recovery from retryable failures so Core can refresh dependent MCP projections. + +use std::sync::Arc; +use std::sync::Weak; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use tokio::sync::Mutex; +use tokio::sync::OnceCell; + +use crate::DiscoverV2CapabilitiesRequest; +use crate::DiscoverV2CapabilitiesResponse; +use crate::Environment; + +const MIN_CACHED_DISCOVERY_CONTEXTS: usize = 16; + +/// Shared by turn and idle consumers; Core replaces its current cache at each turn start. +#[derive(Default)] +pub struct ExecutorCapabilityDiscoveryV2Cache { + entries: Mutex>>, + capacity: AtomicUsize, + recovered_discovery: AtomicBool, +} + +struct CachedEnvironmentDiscovery { + request: DiscoverV2CapabilitiesRequest, + environment: Weak, + session_id: Option, + response: OnceCell, String>>, + retryable: Arc, +} + +impl ExecutorCapabilityDiscoveryV2Cache { + /// Fits a complete environment selection; never shrinks until the cache is replaced. + pub fn ensure_capacity(&self, minimum: usize) { + self.capacity.fetch_max(minimum, Ordering::Relaxed); + } + + /// Reports whether a failed discovery recovered since the last observation. + pub fn take_recovered_discovery(&self) -> bool { + self.recovered_discovery.swap(false, Ordering::AcqRel) + } + + /// Reuses successes and non-retryable failures using V1's error classification. + /// Retryable failures and cancelled loads leave the cell available for retry. + pub async fn discover( + &self, + environment_id: &str, + environment: &Arc, + request: DiscoverV2CapabilitiesRequest, + ) -> Result, String> { + // Resolve startup/replacement before lookup; a resumed session keeps its cache entries. + let session_id = environment.executor_session_id().await.map_err(|_| { + tracing::warn!(environment_id, "executor discovery V2 session unavailable"); + "executor discovery V2 failed".to_string() + })?; + let entry = { + let mut entries = self.entries.lock().await; + let environment = Arc::downgrade(environment); + if let Some(entry) = entries.iter().find(|entry| { + entry.environment.ptr_eq(&environment) + && entry.session_id == session_id + && entry.request == request + }) { + Arc::clone(entry) + } else { + // Recovery belongs to the environment/request, not one executor session. + let retryable = entries + .iter() + .find(|entry| { + entry.environment.ptr_eq(&environment) && entry.request == request + }) + .map(|entry| Arc::clone(&entry.retryable)) + .unwrap_or_default(); + let entry = Arc::new(CachedEnvironmentDiscovery { + request: request.clone(), + environment, + session_id, + response: OnceCell::new(), + retryable, + }); + let capacity = self + .capacity + .load(Ordering::Relaxed) + .max(MIN_CACHED_DISCOVERY_CONTEXTS); + if entries.len() == capacity { + // Evict the oldest context (FIFO). In-flight requests retain their own Arc. + entries.remove(0); + } + entries.push(Arc::clone(&entry)); + entry + } + }; + let response = entry + .response + .get_or_try_init(|| async { + match environment.discover_capabilities_v2(request).await { + Ok(response) => Ok(Ok(Arc::new(response))), + Err(error) => { + let retryable = crate::client::can_retry_connection_attempt(&error); + tracing::warn!(environment_id, "executor discovery V2 failed"); + let error = "executor discovery V2 failed".to_string(); + if retryable { + entry.retryable.store(true, Ordering::Release); + Err(error) + } else { + Ok(Err(error)) + } + } + } + }) + .await? + .clone(); + if response.is_ok() && entry.retryable.swap(false, Ordering::AcqRel) { + self.recovered_discovery.store(true, Ordering::Release); + } + response + } +} diff --git a/codex-rs/exec-server/src/client_refresh_tests.rs b/codex-rs/exec-server/src/client_refresh_tests.rs index 5e391a602a..ac79cb8c40 100644 --- a/codex-rs/exec-server/src/client_refresh_tests.rs +++ b/codex-rs/exec-server/src/client_refresh_tests.rs @@ -20,6 +20,7 @@ use super::*; use crate::ExecServerRuntimeOptions; use crate::NoiseChannelIdentity; use crate::ProcessId; +use crate::discover_v2::capability_locations::tests as location_fixtures; use crate::noise_relay::stream_handler::NoiseOutboundMessage; use crate::noise_relay::stream_handler::NoiseStreamConnection; use crate::noise_relay::stream_handler::NoiseStreamHandler; @@ -285,14 +286,57 @@ async fn refresh_cancels_old_recovery_and_connects_without_resuming_old_session( #[tokio::test] async fn refresh_retires_a_still_connected_old_executor() -> Result<()> { let _clock = freeze_clock(); + let directory = tempfile::tempdir()?; + let root = directory.path().canonicalize()?; + for name in ["old", "new"] { + location_fixtures::write( + &root.join(name), + &format!("codex/skills/{name}/SKILL.md"), + &format!("---\nname: {name}\ndescription: {name} executor\n---\n"), + )?; + } let old = Executor::start(Validator::default()).await?; let new = Executor::start(Validator::default()).await?; + old.processor + .seed_capability_locations(location_fixtures::request(&root.join("old"))?) + .await?; + new.processor + .seed_capability_locations(location_fixtures::request(&root.join("new"))?) + .await?; let (client, registry) = old.client()?; let original = client.get().await?; + let environment = Arc::new(crate::Environment::remote_with_client( + client.clone(), + /*local_runtime_paths*/ None, + ScopedSkillsConfig::default(), + )); + let cache = crate::ExecutorCapabilityDiscoveryV2Cache::default(); + let request = crate::DiscoverV2CapabilitiesRequest { + cwd: codex_utils_path_uri::PathUri::from_host_native_path(&root)?, + sandbox: None, + }; + let inventory = cache + .discover("environment", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(inventory.skills.iter().any(|skill| skill.name == "old")); + environment.refresh_connection().await?; + let reused = cache + .discover("environment", &environment, request.clone()) + .await + .map_err(anyhow::Error::msg)?; + assert!(Arc::ptr_eq(&inventory, &reused)); *registry.target.lock().unwrap() = new.target.clone(); - client.refresh_connection().await?; + environment.refresh_connection().await?; assert!(original.is_disconnected()); assert_ne!(original.session_id(), client.get().await?.session_id()); + let refreshed = cache + .discover("environment", &environment, request) + .await + .map_err(anyhow::Error::msg)?; + assert!(!Arc::ptr_eq(&inventory, &refreshed)); + assert!(refreshed.skills.iter().any(|skill| skill.name == "new")); + assert!(!refreshed.skills.iter().any(|skill| skill.name == "old")); Ok(()) } @@ -375,6 +419,7 @@ async fn registration_renewal_recovers_the_session_and_running_process() -> Resu ); let original = client.get().await?; let session_id = original.session_id(); + assert_eq!(environment.executor_session_id().await?, session_id); let process = original .start_process( @@ -421,6 +466,7 @@ async fn registration_renewal_recovers_the_session_and_running_process() -> Resu let recovered = client.get().await?; assert!(Arc::ptr_eq(&original.inner, &recovered.inner)); assert_eq!(recovered.session_id(), session_id); + assert_eq!(environment.executor_session_id().await?, session_id); assert_eq!( process.write(b"after-renewal\n".to_vec()).await?.status, crate::protocol::WriteStatus::Accepted diff --git a/codex-rs/exec-server/src/discoverV2/capability_locations.rs b/codex-rs/exec-server/src/discoverV2/capability_locations.rs index 5f5a397400..12c443af94 100644 --- a/codex-rs/exec-server/src/discoverV2/capability_locations.rs +++ b/codex-rs/exec-server/src/discoverV2/capability_locations.rs @@ -8,7 +8,6 @@ use std::io; use crate::LocalFileSystem; use crate::protocol::FsOpenMode; -use codex_config::loader::system_config_toml_file; use codex_file_system::ExecutorFileSystem; use codex_file_system::WalkEntryKind; use codex_file_system::WalkOptions; @@ -24,11 +23,12 @@ use codex_core_plugin_common::installed::parse_remote_plugin_id; use codex_core_plugin_common::plugin_id::PluginId; use tokio::io::AsyncReadExt; -/// Home paths are supplied by the executor, not read from the sandbox helper's sanitized env. +/// Fixed paths are supplied by the executor, not read from the sandbox helper's sanitized env. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct CapabilityLocationRequest { pub codex_home: PathUri, pub user_home: Option, + pub system_skills_root: Option, } const MAX_LOCATIONS: usize = 512; @@ -73,8 +73,8 @@ pub(crate) async fn prewarm_locations( if let Some(home) = user_home { skill_roots.push(home.join(AGENTS_SKILLS_SUFFIX)); } - if let Some(system_root) = system_config_toml_file()?.parent() { - skill_roots.push(system_root.join("skills")); + if let Some(system_root) = &request.system_skills_root { + skill_roots.push(system_root.to_abs_path()?); } CandidateCapabilityLocations::resolve( fs, @@ -211,4 +211,4 @@ async fn read_remote_plugin_id(path: &std::path::Path) -> io::Result, ) -> io::Result<&(CapabilityLocationRequest, CapabilityLocations)> { @@ -66,6 +67,9 @@ impl CapabilityManager { codex_home: PathUri::from_abs_path(&codex_utils_home_dir::find_codex_home()?), user_home: dirs::home_dir() .and_then(|home| PathUri::from_host_native_path(home).ok()), + system_skills_root: system_config_toml_file()? + .parent() + .map(|root| PathUri::from_abs_path(&root.join("skills"))), }; let locations = prewarm_locations(&self.file_system, &request).await?; Ok::<_, io::Error>((request, locations)) diff --git a/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs b/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs index 5f6ce6b8fa..280b1b60ba 100644 --- a/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs +++ b/codex-rs/exec-server/src/discoverV2/tests/capability_locations_tests.rs @@ -14,12 +14,15 @@ use crate::LocalFileSystem; pub(crate) const MANIFEST: &str = r#"{"name":"demo","version":"1.10.0"}"#; pub(crate) fn request(root: &Path) -> anyhow::Result { - for path in ["codex", "home", "project"] { + for path in ["codex", "home", "project", "etc/codex"] { std::fs::create_dir_all(root.join(path))?; } Ok(CapabilityLocationRequest { codex_home: PathUri::from_host_native_path(root.join("codex"))?, user_home: Some(PathUri::from_host_native_path(root.join("home"))?), + system_skills_root: Some(PathUri::from_host_native_path( + root.join("etc/codex/skills"), + )?), }) } @@ -42,6 +45,7 @@ async fn prewarming_resolves_global_locations_without_cwd_or_config() -> anyhow: for root in [ "codex/skills", "home/.agents/skills", + "etc/codex/skills", "project/.agents/skills", ] { write(temp.path(), &format!("{root}/demo/SKILL.md"), "unparsed")?; @@ -69,7 +73,7 @@ async fn prewarming_resolves_global_locations_without_cwd_or_config() -> anyhow: remote_plugin_id: Some("plugins~demo".to_string()), }), }]; - for root in ["codex/skills", "home/.agents/skills"] { + for root in ["codex/skills", "home/.agents/skills", "etc/codex/skills"] { expected.push(CapabilityLocation { root: PathUri::from_host_native_path(temp.path().join(root))?, plugin: None, diff --git a/codex-rs/exec-server/src/environment/discovery_v2.rs b/codex-rs/exec-server/src/environment/discovery_v2.rs index ec927bae8c..71daf0e3c4 100644 --- a/codex-rs/exec-server/src/environment/discovery_v2.rs +++ b/codex-rs/exec-server/src/environment/discovery_v2.rs @@ -6,6 +6,14 @@ use crate::DiscoverV2CapabilitiesResponse; use crate::ExecServerError; impl Environment { + /// Identifies the initialized executor session; transport recovery preserves this ID. + pub(crate) async fn executor_session_id(&self) -> Result, ExecServerError> { + match &self.remote_client { + Some(client) => Ok(client.get().await?.session_id()), + None => Ok(None), + } + } + /// Returns metadata for the caller's cwd from the remote session's V2 lifecycle. /// Local environments and older executors retain the V1 selected-root API. pub async fn discover_capabilities_v2( diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index adca2b271c..93d7c98fbd 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -1,6 +1,7 @@ mod arg0_exec_helper; mod capability_discovery; mod capability_discovery_cache; +mod capability_discovery_v2_cache; mod client; mod client_api; mod client_error; @@ -71,6 +72,7 @@ pub use arg0_exec_helper::main as run_arg0_exec_helper_main; pub use capability_discovery::CapabilityDiscoveryError; pub use capability_discovery::discover_capability_roots; pub use capability_discovery_cache::ExecutorCapabilityDiscoveryCache; +pub use capability_discovery_v2_cache::ExecutorCapabilityDiscoveryV2Cache; pub use client::ExecServerClient; pub use client::ExecServerError; pub use client::http_client::HttpResponseBodyStream; @@ -178,6 +180,7 @@ pub use protocol::ExecResponse; pub use protocol::ExecServerNetworkPolicyDecision; pub use protocol::ExecServerNetworkPolicyRequest; pub use protocol::ExecServerNetworkProtocol; +pub use protocol::ExecutorCapabilityDiscoveryV2Snapshot; pub use protocol::ExecutorPlugin; pub use protocol::ExecutorSkill; pub use protocol::FsCanonicalizeParams; diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 0db3bc38a3..b708ce32c4 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -327,6 +327,16 @@ mod tests { use crate::server::session_registry::SessionRegistry; impl super::ConnectionProcessor { + // Discovery tests supply fixture homes without changing the process environment. + pub(crate) async fn seed_capability_locations( + &self, + request: crate::discover_v2::capability_locations::CapabilityLocationRequest, + ) -> std::io::Result<()> { + self.capability_manager + .seed_prewarmed_locations(request) + .await + } + // Connection-only tests do not prewarm or scan the real user home. pub(crate) fn new(runtime_paths: ExecServerRuntimeOptions) -> Self { let telemetry = crate::ExecServerTelemetry::default(); diff --git a/codex-rs/ext/extension-api/src/contributors/mcp.rs b/codex-rs/ext/extension-api/src/contributors/mcp.rs index 699048ddee..bec685a921 100644 --- a/codex-rs/ext/extension-api/src/contributors/mcp.rs +++ b/codex-rs/ext/extension-api/src/contributors/mcp.rs @@ -1,5 +1,6 @@ use codex_config::McpServerConfig; use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server_protocol::ExecutorCapabilityDiscoveryV2Snapshot; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TurnEnvironmentSelection; @@ -32,6 +33,8 @@ pub struct McpServerContributionContext<'a, C> { 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>, + /// V2 inventories supplied by the host for this contribution. + executor_capability_discovery_v2: Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]>, } impl Clone for McpServerContributionContext<'_, C> { @@ -55,6 +58,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { selected_environments: None, ready_selected_capability_roots: None, executor_capability_discovery: None, + executor_capability_discovery_v2: None, } } @@ -66,6 +70,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { originator: &'a str, ready_selected_capability_roots: &'a [SelectedCapabilityRoot], executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>, + executor_capability_discovery_v2: Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]>, ) -> Self { Self { config, @@ -77,6 +82,7 @@ impl<'a, C> McpServerContributionContext<'a, C> { selected_environments: None, ready_selected_capability_roots: Some(ready_selected_capability_roots), executor_capability_discovery, + executor_capability_discovery_v2, } } @@ -145,6 +151,13 @@ impl<'a, C> McpServerContributionContext<'a, C> { self.ready_selected_capability_roots } + /// Returns the V2 inventories supplied by the host, when available. + pub fn executor_capability_discovery_v2( + &self, + ) -> Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]> { + self.executor_capability_discovery_v2 + } + /// 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 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 f5571bb8d8..19dbb91001 100644 --- a/codex-rs/ext/extension-api/src/contributors/world_state.rs +++ b/codex-rs/ext/extension-api/src/contributors/world_state.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server_protocol::ExecutorCapabilityDiscoveryV2Snapshot; use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::openai_models::ModelInfo; @@ -21,6 +22,8 @@ pub struct WorldStateContributionInput<'a> { 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>, + /// Opted-in executor inventories captured for this exact step. + pub executor_capability_discovery_v2: Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]>, /// Metrics bound to the captured model for this sampling step. pub extension_metrics: Option>, pub session_store: &'a ExtensionData, diff --git a/codex-rs/ext/extension-api/tests/registry.rs b/codex-rs/ext/extension-api/tests/registry.rs index a54c83108c..8b35456b87 100644 --- a/codex-rs/ext/extension-api/tests/registry.rs +++ b/codex-rs/ext/extension-api/tests/registry.rs @@ -59,6 +59,7 @@ fn mcp_contribution_context_identifies_the_running_thread() { "codex_work_cca", &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .with_session_source(&session_source); diff --git a/codex-rs/ext/mcp/src/cloud_plugin.rs b/codex-rs/ext/mcp/src/cloud_plugin.rs index 2d10b403d2..9cabe53f4b 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin.rs @@ -102,6 +102,7 @@ impl TurnLifecycleContributor for PluginContributor { thread_id: input.thread_store.level_id().to_string(), turn_id: input.turn_id.to_string(), mcp_resources: resources.map(Arc::new), + snapshots: Vec::new(), }) .await { diff --git a/codex-rs/ext/mcp/src/cloud_plugin_tests.rs b/codex-rs/ext/mcp/src/cloud_plugin_tests.rs index dac29686f2..385d59b7c9 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin_tests.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin_tests.rs @@ -169,6 +169,7 @@ async fn disabling_plugins_clears_cloud_catalog_and_skips_discovery() -> anyhow: "test", &[], /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, )) .await; if let Some(state) = thread.get::() { @@ -235,6 +236,7 @@ async fn account_change_clears_hosted_plugins_before_waiting_for_executor() -> a let init = codex_extension_api::ExtensionDataInit::default(); let context = McpServerContributionContext::for_step( &config, &init, &thread, "test", &roots, /*executor_capability_discovery*/ None, + /*executor_capability_discovery_v2*/ None, ) .with_auth_changed(/*auth_changed*/ true); let mut loading = std::pin::pin!(contributor.selected_plugins(context, &config.plugins)); diff --git a/codex-rs/ext/mcp/src/lib_tests.rs b/codex-rs/ext/mcp/src/lib_tests.rs index a6c526acf5..6ce1f07f04 100644 --- a/codex-rs/ext/mcp/src/lib_tests.rs +++ b/codex-rs/ext/mcp/src/lib_tests.rs @@ -29,6 +29,7 @@ async fn hosted_plugin_runtime_forwards_thread_originator() -> Result<(), Box TestResu "test_originator", &selected_roots, discovery.as_ref(), + /*executor_capability_discovery_v2*/ None, ), &config.plugins, ) @@ -661,6 +662,10 @@ async fn raw_selected_plugin_contributions( None }; + let v2 = [codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: LOCAL_ENVIRONMENT_ID.to_string(), + result: Err("V2 must not change existing contributions".to_string()), + }]; let selected = registry.mcp_server_contributors()[0] .selected_plugins( McpServerContributionContext::for_step( @@ -670,6 +675,7 @@ async fn raw_selected_plugin_contributions( "test_originator", &selected_capability_roots, executor_capability_discovery.as_ref(), + Some(&v2), ), &config.plugins, ) diff --git a/codex-rs/ext/skills/src/cloud_cache_tests.rs b/codex-rs/ext/skills/src/cloud_cache_tests.rs index 5844dc4c4b..de05327814 100644 --- a/codex-rs/ext/skills/src/cloud_cache_tests.rs +++ b/codex-rs/ext/skills/src/cloud_cache_tests.rs @@ -89,6 +89,7 @@ fn query(client: Option>) -> SkillListQuery { include_cloud_skills: true, mcp_resources: client, executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), } } diff --git a/codex-rs/ext/skills/src/cloud_skill.rs b/codex-rs/ext/skills/src/cloud_skill.rs index 6adbefbae2..3f491e769f 100644 --- a/codex-rs/ext/skills/src/cloud_skill.rs +++ b/codex-rs/ext/skills/src/cloud_skill.rs @@ -53,6 +53,7 @@ impl TurnLifecycleContributor for SkillsExtension { .get::() .and_then(|state| state.mcp_resources.clone()), executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }; if let Err(error) = state.refresh_cloud_catalog(&self.providers, query).await { self.emit_warning( diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index 2e85c98cea..c38a9d82df 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -4,6 +4,7 @@ use crate::HostSkillsSnapshot; use crate::InjectedHostSkillPrompts; use codex_analytics::InvocationType; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::LOCAL_ENVIRONMENT_ID; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_extension_api::ConfigContributor; @@ -210,6 +211,7 @@ where .get::() .and_then(|state| state.mcp_resources.clone()), executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }, &thread_state, ) @@ -308,6 +310,10 @@ where executor_capability_discovery: step_store .get::() .map(|discovery| discovery.as_ref().clone()), + executor_capability_discovery_v2: step_store + .get::>() + .map(|discovery| discovery.as_ref().clone()) + .unwrap_or_default(), }); self.build_skill_tools( session_store, @@ -382,6 +388,7 @@ where include_cloud_skills: thread_state.cloud_skill_enabled(), mcp_resources: mcp_resources.clone(), executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }; let mut catalog = turn_store .get::() diff --git a/codex-rs/ext/skills/src/provider.rs b/codex-rs/ext/skills/src/provider.rs index c2a1210cdb..0532c51a02 100644 --- a/codex-rs/ext/skills/src/provider.rs +++ b/codex-rs/ext/skills/src/provider.rs @@ -8,6 +8,7 @@ mod host; use crate::HostSkillsSnapshot; use codex_exec_server::EnvironmentAccess; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; +use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::ResolvedSelectedCapabilityRoot; use codex_mcp::McpResourceClient; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -38,6 +39,8 @@ pub struct SkillListQuery { pub mcp_resources: Option>, /// Present only when the opt-in high-level executor discovery path is selected. pub executor_capability_discovery: Option, + /// Inert V2 inventories; empty when none were supplied. Existing discovery is unchanged. + pub executor_capability_discovery_v2: Vec, } /// A skill read with the source access supplied by its callback. diff --git a/codex-rs/ext/skills/src/render_tests.rs b/codex-rs/ext/skills/src/render_tests.rs index 58aa7ae014..fa67d777c6 100644 --- a/codex-rs/ext/skills/src/render_tests.rs +++ b/codex-rs/ext/skills/src/render_tests.rs @@ -547,6 +547,7 @@ async fn host_alias_roots_follow_core_discovery_order() -> Result<(), Box>(); @@ -915,6 +916,7 @@ async fn singleton_plugin_versions_share_the_marketplace_alias_root() include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }) .await?; let entries = catalog.entries.iter().collect::>(); diff --git a/codex-rs/ext/skills/src/world_state_catalogs.rs b/codex-rs/ext/skills/src/world_state_catalogs.rs index 944f3b4467..8950896060 100644 --- a/codex-rs/ext/skills/src/world_state_catalogs.rs +++ b/codex-rs/ext/skills/src/world_state_catalogs.rs @@ -168,6 +168,11 @@ impl<'a> CatalogContext<'a> { .get::() .and_then(|state| state.mcp_resources.clone()), executor_capability_discovery: self.input.executor_capability_discovery.cloned(), + executor_capability_discovery_v2: self + .input + .executor_capability_discovery_v2 + .map(Vec::from) + .unwrap_or_default(), }; let cloud = self.cloud_catalog_contribution(&query); @@ -244,6 +249,7 @@ impl<'a> CatalogContext<'a> { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }) .await; self.input 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 cdf7f416ff..c2ab52f9fe 100644 --- a/codex-rs/ext/skills/tests/executor_file_system_authority.rs +++ b/codex-rs/ext/skills/tests/executor_file_system_authority.rs @@ -417,6 +417,7 @@ async fn selected_root_id_distinguishes_identical_executor_paths() { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }) .await .expect("list executor skills"); @@ -507,6 +508,7 @@ async fn executor_discovery_preserves_posix_and_windows_locator_alias_roots() { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: Some(discovery), + executor_capability_discovery_v2: Vec::new(), }) .await .expect("list executor skills"); @@ -624,6 +626,12 @@ async fn executor_discovery_routes_produce_equivalent_catalog_metadata() { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery, + executor_capability_discovery_v2: vec![ + codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot { + environment_id: "local".to_string(), + result: Err("V2 must not change V1 or direct discovery".to_string()), + }, + ], }; let direct = provider @@ -776,6 +784,7 @@ async fn pre_discovered_executor_catalog_snapshot() { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: Some(executor_capability_discovery), + executor_capability_discovery_v2: Vec::new(), }) .await .expect("list pre-discovered executor skills"); @@ -893,6 +902,7 @@ async fn direct_executor_discovery_preserves_hidden_nested_and_probed_metadata() include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: None, + executor_capability_discovery_v2: Vec::new(), }) .await .expect("list directly discovered executor skills"); @@ -953,6 +963,7 @@ async fn high_level_discovery_reuses_materialized_skill_contents_for_reads() { include_cloud_skills: false, mcp_resources: None, executor_capability_discovery: Some(executor_capability_discovery), + executor_capability_discovery_v2: Vec::new(), }) .await .expect("list executor skills"); diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index e03d694b18..1d1c46b401 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -134,6 +134,7 @@ async fn skill_world_state_fragments( let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id, @@ -322,6 +323,7 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -349,6 +351,7 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -389,6 +392,7 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -463,6 +467,7 @@ async fn persisted_host_snapshot_deduplicates_warning_after_reinitialization() - let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &model_info, thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -504,6 +509,7 @@ async fn persisted_host_snapshot_deduplicates_warning_after_reinitialization() - let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &model_info, thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-2", @@ -597,6 +603,7 @@ async fn executor_cloud_and_host_share_catalog_world_state_flow() -> TestResult let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -708,6 +715,7 @@ async fn nonempty_executor_empty_host_records_catalog_metrics() -> TestResult { let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -790,6 +798,7 @@ async fn host_world_state_uses_provider_catalog_with_core_compatible_rendering() let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -873,6 +882,7 @@ async fn shadow_selection_uses_host_catalog_when_instructions_are_disabled() -> let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -1167,6 +1177,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach let available_sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -1232,6 +1243,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach let unavailable_sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: Some(&available_world_state), + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-2", @@ -1271,6 +1283,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach let restored_sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: Some(&unavailable_world_state), + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-3", @@ -1330,6 +1343,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id, @@ -1358,6 +1372,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach let listing_disabled_sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-4", @@ -1487,6 +1502,7 @@ async fn overlapping_executor_catalogs_render_their_own_roots() -> TestResult { environments: &[], ready_selected_capability_roots: &[root], executor_capability_discovery: None, + executor_capability_discovery_v2: None, extension_metrics: None, session_store: &session_store, thread_store: &thread_store, @@ -1762,6 +1778,7 @@ async fn catalog_rebalances_only_to_avoid_omissions_and_retains_the_allocation() environments: &[], ready_selected_capability_roots: if ready { &selected_roots } else { &[] }, executor_capability_discovery: None, + executor_capability_discovery_v2: None, extension_metrics: None, session_store: &session_store, thread_store: &thread_store, @@ -2382,6 +2399,7 @@ async fn root_qualified_locator_selects_only_the_matching_executor_skill() -> Te registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -2519,6 +2537,7 @@ async fn model_context_window_scales_executor_and_cloud_catalogs() -> TestResult let sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &model_info, thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -2536,6 +2555,7 @@ async fn model_context_window_scales_executor_and_cloud_catalogs() -> TestResult let _repeated_sections = registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &model_info, thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", @@ -2636,6 +2656,7 @@ async fn executor_catalog_emits_at_most_four_warnings() -> TestResult { registry.context_contributors()[0] .contribute_world_state(WorldStateContributionInput { previous_world_state: None, + executor_capability_discovery_v2: None, model_info: &catalog_model_info(), thread_id: codex_protocol::ThreadId::new(), turn_id: "turn-1", diff --git a/codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs b/codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs index c67d715ff1..1c9d62a628 100644 --- a/codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs +++ b/codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs @@ -119,6 +119,7 @@ async fn cloud_preference_preserves_aliases_reads_and_executor_fallback() -> Tes environments: &[], ready_selected_capability_roots: if ready { &roots } else { &[] }, executor_capability_discovery: None, + executor_capability_discovery_v2: None, extension_metrics: None, session_store: &session_store, thread_store: &thread_store, diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 13ff893c1c..0928655c33 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -275,6 +275,8 @@ pub enum Feature { Plugins, /// Discover selected-root plugin and skill manifests through one high-level exec-server RPC. ExecutorCapabilityDiscovery, + /// Acquire V2 executor inventories without publishing or activating their capabilities. + ExecutorCapabilityDiscoveryV2, /// Skip host skill snapshots when no registered contributor requires them. SkipHostSkillDiscovery, /// Removed compatibility flag for plugin-bundled lifecycle hooks. @@ -1628,6 +1630,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::ExecutorCapabilityDiscoveryV2, + key: "executor_capability_discovery_v2", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::SkipHostSkillDiscovery, key: "skip_host_skill_discovery",