Gate capability discovery V2 on executor version (#57192)

## Why

Executors older than `0.164.0` can advertise V2 discovery support without implementing the final contract. Enabling V2 also suppressed V1 discovery for incompatible executors, hiding selected capability roots and skills.

## What changed

Require both the V2 capability flag and an executor version of at least `0.164.0` before issuing `capabilities/discoverV2`. Share this compatibility check between session discovery and the exec-server client. Fall back to V1 when incompatible executors provide no V2 snapshots, clearing any previously stored V2 inventory.

## Testing

Add coverage for version boundaries, prereleases, missing and malformed versions, and transport recovery for compatible executors. Verify that an older executor retains selected plugins and skill instructions through V1 with V2 enabled.

GitOrigin-RevId: 93adf52fc91f02b58fd14f8ff3b093712b3346b8

codex-rs/core/src/session/mcp.rs

@@ -10,6 +10,7 @@ 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_exec_server::supports_capability_discovery_v2;
 use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
 use codex_mcp::ElicitationReviewRequest;
 use codex_mcp::ElicitationReviewer;
@@ -133,6 +134,7 @@ impl Session {
                 config,
                 &selected_capability_roots,
                 &environments,
+                executor_capability_discovery_v2.as_deref(),
             )
             .await;
         if inventory_changed || discovery_recovered {
@@ -247,6 +249,7 @@ impl Session {
                     &desired.config,
                     &selected_capability_roots,
                     &desired.environments,
+                    executor_capability_discovery_v2.as_deref(),
                 )
                 .await;
             let mcp_projection = self
@@ -329,6 +332,7 @@ impl Session {
                 &desired.config,
                 &selected_capability_roots,
                 &desired.environments,
+                executor_capability_discovery_v2.as_deref(),
             )
             .await;
         let mcp_projection = self
@@ -481,11 +485,9 @@ impl Session {
         config: &Config,
         resolved_roots: &[ResolvedSelectedCapabilityRoot],
         environments: &TurnEnvironmentSnapshot,
+        v2_snapshots: Option<&[ExecutorCapabilityDiscoveryV2Snapshot]>,
     ) -> (Option<Arc<ExecutorCapabilityDiscoverySnapshot>>, bool) {
-        if config
-            .features
-            .enabled(Feature::ExecutorCapabilityDiscoveryV2)
-        {
+        if v2_snapshots.is_some() {
             return (None, false);
         }
         let restricted_file_system = environments.primary().map_or_else(
@@ -577,6 +579,7 @@ impl Session {
             .count();
         cache.ensure_capacity(remote_environment_count);
         let mut snapshots = Vec::new();
+        let mut skipped_incompatible_executor = false;
         for selected in environments.turn_environments() {
             if !selected.environment.is_remote() {
                 continue;
@@ -592,7 +595,8 @@ impl Session {
                     continue;
                 }
             };
-            if !info.capabilities.capability_discovery_v2 {
+            if !supports_capability_discovery_v2(&info) {
+                skipped_incompatible_executor = true;
                 continue;
             }
             let request = DiscoverV2CapabilitiesRequest {
@@ -607,6 +611,14 @@ impl Session {
                 result,
             });
         }
+        if snapshots.is_empty() && skipped_incompatible_executor {
+            let changed = self
+                .services
+                .thread_extension_data
+                .remove::<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>()
+                .is_some();
+            return (None, changed);
+        }
         let previous = self
             .services
             .thread_extension_data
@@ -870,6 +882,7 @@ impl Session {
                 &desired.config,
                 &selected_capability_roots,
                 &desired.environments,
+                executor_capability_discovery_v2.as_deref(),
             )
             .await;
         let mcp_projection = self

codex-rs/core/src/session/mod.rs

@@ -4059,6 +4059,7 @@ impl Session {
                     &turn_context.config,
                     &selected_capability_roots,
                     &environments,
+                    executor_capability_discovery_v2.as_deref(),
                 )
                 .await
             };

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

@@ -10146,6 +10146,7 @@ async fn capability_discovery_uses_environment_permission_profile() {
             &turn_context.config,
             /*resolved_roots*/ &[],
             &turn_context.initial_environments,
+            /*v2_snapshots*/ None,
         )
         .await;
     let discovery = discovery.expect("restricted environment should trigger capability discovery");

codex-rs/core/tests/common/exec_server.rs

@@ -30,10 +30,32 @@ impl ExecServerProcess {
 
     pub async fn start_with_env(env: &[(&str, &str)]) -> Result<Self> {
         let codex_home = TempDir::new()?;
-        let mut child = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
+        let mut command = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?);
+        command.envs(env.iter().copied());
+        Self::start_command(command, codex_home).await
+    }
+
+    pub async fn start_with_version(version: &str) -> Result<Self> {
+        let codex_home = TempDir::new()?;
+        let package = codex_home.path().join("package");
+        let bin = package.join("bin");
+        std::fs::create_dir_all(&bin)?;
+        let executable = bin.join(format!("codex{}", std::env::consts::EXE_SUFFIX));
+        codex_utils_cargo_bin::copy_executable(
+            &codex_utils_cargo_bin::cargo_bin("codex")?,
+            &executable,
+        )?;
+        std::fs::write(
+            package.join("codex-package.json"),
+            serde_json::to_vec(&serde_json::json!({"version": version}))?,
+        )?;
+        Self::start_command(Command::new(executable), codex_home).await
+    }
+
+    async fn start_command(mut command: Command, codex_home: TempDir) -> Result<Self> {
+        let mut child = command
             .args(["exec-server", "--listen", "ws://127.0.0.1:0"])
             .env("CODEX_HOME", codex_home.path())
-            .envs(env.iter().copied())
             .stdin(Stdio::null())
             .stdout(Stdio::piped())
             .stderr(Stdio::inherit())

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

@@ -209,6 +209,14 @@ enum DiscoveryFailure {
 
 impl Executor {
     async fn start(supports_v2: bool, failure: DiscoveryFailure) -> Result<Self> {
+        Self::start_with_version(supports_v2, failure, "0.164.0").await
+    }
+
+    async fn start_with_version(
+        supports_v2: bool,
+        failure: DiscoveryFailure,
+        version: &'static str,
+    ) -> Result<Self> {
         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()));
@@ -267,15 +275,23 @@ impl Executor {
                         let result = match request["method"].as_str() {
                             Some("initialize") => Some(json!({"sessionId":"acquisition-only", "environmentInfo": {
                                 "shell":{"name":"bash","path":"/bin/bash"},
+                                "executorVersion":version,
                                 "capabilities":{"sandboxedFileStreaming":true,"capabilityDiscoveryV2":supports_v2}
                             }})),
                             Some("environment/info") => Some(json!({
                                 "shell":{"name":"bash","path":"/bin/bash"},
+                                "executorVersion":version,
                                 "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::<Vec<_>>()
+                                    .iter().map(|root| json!({
+                                        "id":root["id"],"path":root["path"],
+                                        "skills":[{"instructions":{
+                                            "path":format!("{}/inert-sentinel/SKILL.md", root["path"].as_str().expect("root URI").trim_end_matches('/')),
+                                            "contents":"---\nname: inert-sentinel\ndescription: Executor skill used by the acquisition test\n---\n# Sentinel instructions from the captured executor"
+                                        }}]
+                                    })).collect::<Vec<_>>()
                             })),
                             Some("capabilities/discoverV2") => Some(inventory.lock().await.clone().unwrap_or_else(|| json!({
                                 "plugins":[{"id":"inert-sentinel@market","displayName":"inert-sentinel","remotePluginId":null,
@@ -322,12 +338,12 @@ impl Executor {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn v2_disabled_skips_inventory_discovery() -> Result<()> {
-    check_acquisition(/*enabled*/ false, /*supported*/ true).await
+    check_acquisition(/*enabled*/ false, "0.164.0").await
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn v2_shares_inventory_with_mcp_and_skills_until_next_turn() -> Result<()> {
-    check_acquisition(/*enabled*/ true, /*supported*/ true).await
+    check_acquisition(/*enabled*/ true, "0.164.0").await
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -537,9 +553,15 @@ async fn v2_discovery_failures_retry_or_remain_cached_by_error_kind() -> Result<
     Ok(())
 }
 
-async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn v2_old_executor_retains_v1_selected_roots() -> Result<()> {
+    check_acquisition(/*enabled*/ true, "0.163.0").await
+}
+
+async fn check_acquisition(enabled: bool, version: &'static str) -> Result<()> {
     skip_if_no_network!(Ok(()));
-    let executor = Executor::start(supported, DiscoveryFailure::None).await?;
+    let executor =
+        Executor::start_with_version(/*supports_v2*/ true, DiscoveryFailure::None, version).await?;
     let model = responses::start_mock_server().await;
     let observed = Arc::new(Mutex::new(Vec::new()));
     let selected_plugins = Arc::new(Mutex::new(None));
@@ -601,7 +623,7 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
             .filter(|request| request["method"] == "capabilities/discoverV2")
             .count()
     };
-    let acquires_v2 = enabled && supported;
+    let acquires_v2 = enabled && version == "0.164.0";
     if acquires_v2 {
         observed.lock().await.clear();
         mcp_snapshots_per_step.lock().await.clear();
@@ -678,13 +700,11 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
         for body in bodies {
             responses.push(responses::mount_sse_once(&model, body).await);
         }
-        let prompt = if acquires_v2 {
-            "Use $inert-sentinel to say hello."
-        } else {
-            "Say hello."
-        };
-        test.submit_turn_with_environments(prompt, Some(vec![selection.clone()]))
-            .await?;
+        test.submit_turn_with_environments(
+            "Use $inert-sentinel to say hello.",
+            Some(vec![selection.clone()]),
+        )
+        .await?;
         if acquires_v2 {
             let counts = mcp_snapshots_per_step.lock().await;
             let first = counts
@@ -708,7 +728,7 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
             plugin_id: "inert-sentinel@market".to_string(),
         };
         // Both paths publish accepted Cloud packages and available installations only.
-        let mut expected_plugins = vec![
+        let expected_plugins = vec![
             (Some(executor_root), local_identity),
             (
                 None,
@@ -717,9 +737,6 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
                 },
             ),
         ];
-        if enabled && !supported {
-            expected_plugins.retain(|(root, _)| root.is_none());
-        }
         {
             let selected_plugins = selected_plugins.lock().await;
             let snapshot = selected_plugins.as_ref().context("step plugin snapshot")?;
@@ -739,20 +756,17 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
             .collect::<Vec<_>>();
         assert_eq!(model_requests.len(), if acquires_v2 { 2 } else { 1 });
         for request in model_requests {
-            if acquires_v2 {
-                assert!(
-                    request
-                        .body_json()
-                        .to_string()
-                        .contains("Sentinel instructions from the captured executor")
-                );
-            }
+            assert!(
+                request
+                    .body_json()
+                    .to_string()
+                    .contains("Sentinel instructions from the captured executor")
+            );
             let developer_text = request.message_input_texts("developer").join("\n");
-            assert_eq!(
+            assert!(
                 developer_text
                     .contains("- inert-sentinel: Executor skill used by the acquisition test"),
-                acquires_v2,
-                "only a successful V2 inventory renders the skill catalog",
+                "both discovery modes must render the skill catalog",
             );
         }
         let turn_calls = calls().await;
@@ -796,8 +810,8 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
         requests
             .iter()
             .any(|request| request["method"] == "capabilityRoots/discoverV1"),
-        !enabled,
-        "the V2 flag selects one protocol, including when no executor supplies a V2 snapshot",
+        !acquires_v2,
+        "incompatible executors retain V1 discovery even with the V2 flag enabled",
     );
     if acquires_v2 {
         let reads = requests

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

@@ -1409,7 +1409,8 @@ async fn skill_read_recovers_omitted_installation_and_reads_selected_environment
         }
     }
 
-    let executor = core_test_support::exec_server::ExecServerProcess::start().await?;
+    let executor =
+        core_test_support::exec_server::ExecServerProcess::start_with_version("0.164.0").await?;
     // Keep byte-budget decisions stable across host temporary-directory lengths.
     let temp_root = fs::canonicalize(std::env::temp_dir())?;
     let workspace = tempfile::Builder::new()

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

@@ -86,7 +86,8 @@ async fn frozen_cloud_skills_with_named_executor_request_history() -> Result<()>
         })
         .build(&server)
         .await?;
-    let executor = core_test_support::exec_server::ExecServerProcess::start().await?;
+    let executor =
+        core_test_support::exec_server::ExecServerProcess::start_with_version("0.164.0").await?;
     test.thread_manager
         .environment_manager()
         .upsert_environment(

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

@@ -135,7 +135,13 @@ impl NoiseStreamHandler for NotifyingProcessor {
         ConnectionProcessor::decode(payload)
     }
 
-    fn encode(message: Self::Outgoing) -> Result<NoiseOutboundMessage, ExecServerError> {
+    fn encode(mut message: Self::Outgoing) -> Result<NoiseOutboundMessage, ExecServerError> {
+        // Advertise the stable contract implemented by this test executor.
+        if let codex_exec_server_protocol::JSONRPCMessage::Response(response) = &mut message
+            && let Some(info) = response.result.get_mut("environmentInfo")
+        {
+            info["executorVersion"] = serde_json::json!("0.164.0");
+        }
         ConnectionProcessor::encode(message)
     }
 

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

@@ -1,6 +1,7 @@
 mod connect_options;
 mod discovery_recovery;
 mod discovery_v2;
+pub use discovery_v2::supports_capability_discovery_v2;
 
 use std::collections::HashMap;
 use std::collections::HashSet;

codex-rs/exec-server/src/environment/discovery_v2.rs

@@ -3,7 +3,16 @@
 use super::Environment;
 use crate::DiscoverV2CapabilitiesRequest;
 use crate::DiscoverV2CapabilitiesResponse;
+use crate::EnvironmentInfo;
 use crate::ExecServerError;
+use codex_build_info::BuildInfo;
+
+/// The final DiscoveryV2 contract starts at stable 0.164.0.
+pub fn supports_capability_discovery_v2(info: &EnvironmentInfo) -> bool {
+    let build = BuildInfo::from_version(&info.executor_version);
+    let minimum = BuildInfo::from_version("0.164.0");
+    info.capabilities.capability_discovery_v2 && build.version() >= minimum.version()
+}
 
 impl Environment {
     /// Identifies the initialized executor session; transport recovery preserves this ID.
@@ -28,12 +37,8 @@ impl Environment {
         let connection_state = client.subscribe_connection_state();
         let client = client.get().await?;
         let discover = || async {
-            if !client
-                .environment_info()
-                .await?
-                .capabilities
-                .capability_discovery_v2
-            {
+            let info = client.environment_info().await?;
+            if !supports_capability_discovery_v2(&info) {
                 return Err(ExecServerError::Protocol(
                     "capabilities/discoverV2 is unavailable in this environment session"
                         .to_string(),

codex-rs/exec-server/src/environment/discovery_v2_tests.rs

@@ -1,4 +1,4 @@
-//! Real transport recovery preserves a discovery request's cwd and retries it once.
+//! Version compatibility gates discovery RPCs; supported clients retain transport recovery.
 
 use super::DiscoverV2CapabilitiesRequest;
 use super::DiscoverV2CapabilitiesResponse;
@@ -21,8 +21,18 @@ use tokio_tungstenite::accept_async;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_util::task::AbortOnDropHandle;
 
+#[test_case::test_case("0.164.0", true; "minimum_stable")]
+#[test_case::test_case("0.168.0+build", true; "newer_stable")]
+#[test_case::test_case("0.163.0", false; "older")]
+#[test_case::test_case("0.164.0-alpha.1", false; "prerelease_before_cutoff")]
+#[test_case::test_case("0.165.0-alpha.1", true; "next_minor_prerelease")]
+#[test_case::test_case("", false; "missing")]
+#[test_case::test_case("invalid", false; "malformed")]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn discovery_retries_disconnect_within_the_same_request() -> anyhow::Result<()> {
+async fn discovery_checks_version_and_retries_disconnect(
+    version: &'static str,
+    supported: bool,
+) -> anyhow::Result<()> {
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let url = format!("ws://{}", listener.local_addr()?);
     let attempts = Arc::new(Mutex::new(Vec::new()));
@@ -38,10 +48,13 @@ async fn discovery_retries_disconnect_within_the_same_request() -> anyhow::Resul
                 if request.get("id").is_none() {
                     continue;
                 }
-                let info = json!({
+                let mut info = json!({
                     "shell":{"name":"bash","path":"/bin/bash"},
                     "capabilities":{"capabilityDiscoveryV2":true}
                 });
+                if !version.is_empty() {
+                    info["executorVersion"] = json!(version);
+                }
                 let result = match request["method"].as_str() {
                     Some("initialize") => {
                         // V2 support is advertised by the executor, not negotiated by the client.
@@ -87,8 +100,17 @@ async fn discovery_retries_disconnect_within_the_same_request() -> anyhow::Resul
         Duration::from_secs(12),
         environment.discover_capabilities_v2(params.clone()),
     )
-    .await??;
-    assert_eq!(response, DiscoverV2CapabilitiesResponse::default());
+    .await?;
+    if !supported {
+        assert!(
+            matches!(&response, Err(crate::ExecServerError::Protocol(message))
+                if message == "capabilities/discoverV2 is unavailable in this environment session"),
+            "unexpected discovery response: {response:?}"
+        );
+        assert_eq!(*attempts.lock().await, Vec::<serde_json::Value>::new());
+        return Ok(());
+    }
+    assert_eq!(response?, DiscoverV2CapabilitiesResponse::default());
     assert_eq!(
         *attempts.lock().await,
         vec![

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

@@ -125,6 +125,7 @@ pub use environment::LOCAL_ENVIRONMENT_ID;
 pub use environment::MAX_SELECTED_CAPABILITY_ROOTS;
 pub use environment::REMOTE_ENVIRONMENT_ID;
 pub use environment::RemoteEnvironmentOptions;
+pub use environment::supports_capability_discovery_v2;
 pub use environment_bootstrap::PreparedEnvironmentManager;
 pub use environment_provider::DefaultEnvironmentProvider;
 pub use environment_provider::EnvironmentProvider;