Test exec-server compatibility across Codex versions (#35990)

## Why

The executor protocol supports Codex releases back to `0.145.0`, so compatibility needs to hold when either side of the app-server/exec-server connection is upgraded first.

## What changed

- Define `MINIMUM_SUPPORTED_CODEX_VERSION` in `codex-exec-server-protocol`.
- Add a Unix test harness that runs current-to-released and released-to-current command execution over authenticated Noise connections.
- Test the current binary against itself, the latest release, and the minimum supported release by default, while allowing explicit release versions.
- Verify that the remote command runs successfully and relay payloads remain encrypted, and increase the relay test timeout to accommodate the end-to-end scenarios.

GitOrigin-RevId: faea8d44fce161f40ed15170876a1282a6de4c22

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

@@ -7,3 +7,6 @@ pub use network_policy::*;
 pub use process_id::ProcessId;
 pub use protocol::*;
 pub use rpc::*;
+
+/// Oldest Codex release supported by the executor protocol.
+pub const MINIMUM_SUPPORTED_CODEX_VERSION: &str = "0.145.0";

codex-rs/exec-server/testing/run_version_skew.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+minimum_supported_release="$(
+  sed -n 's/^pub const MINIMUM_SUPPORTED_CODEX_VERSION: &str = "\([^"]*\)";$/\1/p' \
+    "${repo_root}/codex-rs/exec-server-protocol/src/lib.rs"
+)"
+: "${minimum_supported_release:?minimum supported Codex release is missing}"
+release_directory="$(mktemp -d "${TMPDIR:-/tmp}/codex-exec-server-skew.XXXXXX")"
+trap 'rm -rf "${release_directory:?}"' EXIT
+
+if [[ $# -eq 0 ]]; then
+  releases=(latest "${minimum_supported_release}")
+else
+  releases=("$@")
+fi
+
+case "$(uname -s):$(uname -m)" in
+  Darwin:arm64) target="aarch64-apple-darwin" ;;
+  Darwin:x86_64) target="x86_64-apple-darwin" ;;
+  Linux:aarch64 | Linux:arm64) target="aarch64-unknown-linux-musl" ;;
+  Linux:x86_64) target="x86_64-unknown-linux-musl" ;;
+  *)
+    echo "Unsupported platform: $(uname -s) $(uname -m)" >&2
+    exit 1
+    ;;
+esac
+
+asset="codex-${target}.tar.gz"
+cd "${repo_root}/codex-rs"
+cargo build -p codex-cli --bin codex
+export CODEX_TEST_CURRENT_CODEX="${CARGO_TARGET_DIR:-${repo_root}/codex-rs/target}/debug/codex"
+
+echo "Testing current Codex compatibility through authenticated Noise"
+export CODEX_TEST_RELEASED_CODEX="${CODEX_TEST_CURRENT_CODEX}"
+just test -p codex-exec-server --test relay version_skew --test-threads 1
+
+tested_release_version=""
+for release in "${releases[@]}"; do
+  release="${release#rust-v}"
+  if [[ "${release}" == "${tested_release_version}" ]]; then
+    echo "Skipping Codex ${release}; this release was already tested"
+    continue
+  fi
+
+  if [[ "${release}" == "latest" ]]; then
+    release_url="https://github.com/openai/codex/releases/latest/download/${asset}"
+  else
+    release_url="https://github.com/openai/codex/releases/download/rust-v${release}/${asset}"
+  fi
+
+  binary_directory="${release_directory}/${release}"
+  mkdir -p "${binary_directory}"
+  echo "Downloading released Codex from ${release_url}"
+  curl -fsSL "${release_url}" -o "${binary_directory}/${asset}"
+  tar -xzf "${binary_directory}/${asset}" -C "${binary_directory}"
+
+  export CODEX_TEST_RELEASED_CODEX="${binary_directory}/codex-${target}"
+  release_output="$("${CODEX_TEST_RELEASED_CODEX}" --version)"
+  echo "${release_output}"
+  tested_release_version="${release_output##* }"
+
+  just test -p codex-exec-server --test relay version_skew --test-threads 1
+done

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

@@ -1,5 +1,9 @@
 mod common;
 
+#[cfg(unix)]
+#[path = "relay/version_skew.rs"]
+mod version_skew;
+
 #[path = "../src/proto/codex.exec_server.relay.v1.rs"]
 mod relay_proto;
 
@@ -70,7 +74,7 @@ const ENVIRONMENT_ID: &str = "env-noise-relay-test";
 const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
 const HARNESS_KEY_AUTHORIZATION: &str = "harness-key-authorization";
 const REGISTRY_TOKEN: &str = "registry-token";
-const TEST_TIMEOUT: Duration = Duration::from_secs(10);
+const TEST_TIMEOUT: Duration = Duration::from_secs(30);
 
 #[derive(Debug)]
 struct StaticRegistryAuthProvider;

codex-rs/exec-server/tests/relay/version_skew.rs

@@ -0,0 +1,381 @@
+use std::path::Path;
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use anyhow::Context;
+use anyhow::Result;
+use codex_exec_server::CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR;
+use codex_exec_server::CODEX_EXEC_SERVER_NOISE_ENVIRONMENT_ID_ENV_VAR;
+use codex_exec_server::CODEX_EXEC_SERVER_NOISE_REGISTRY_URL_ENV_VAR;
+use codex_exec_server::CODEX_EXEC_SERVER_URL_ENV_VAR;
+use pretty_assertions::assert_eq;
+use serde_json::Value;
+use serde_json::json;
+use tempfile::TempDir;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::Lines;
+use tokio::net::TcpListener;
+use tokio::process::ChildStdin;
+use tokio::process::ChildStdout;
+use tokio::process::Command;
+use tokio::time::timeout;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::header;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+
+use super::ENVIRONMENT_ID;
+use super::EXECUTOR_REGISTRATION_ID;
+use super::HARNESS_KEY_AUTHORIZATION;
+use super::REGISTRY_TOKEN;
+use super::accept_websocket;
+use super::assert_relay_data_is_encrypted;
+use super::proxy_relay_frames;
+use super::registered_executor_public_key;
+
+const RELEASED_CODEX_ENV_VAR: &str = "CODEX_TEST_RELEASED_CODEX";
+const CURRENT_CODEX_ENV_VAR: &str = "CODEX_TEST_CURRENT_CODEX";
+const EXECUTOR_MARKER_ENV_VAR: &str = "CODEX_EXECUTOR_VERSION_SKEW_MARKER";
+const VERSION_SKEW_TIMEOUT: Duration = Duration::from_secs(30);
+const EXPECTED_OUTPUT: &str = "executor-version-skew-ok";
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn current_app_server_runs_commands_on_released_exec_server_over_noise() -> Result<()> {
+    let Some((current, released)) = version_skew_binaries()? else {
+        return Ok(());
+    };
+    assert_noise_version_skew(&current, &released).await
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn released_app_server_runs_commands_on_current_exec_server_over_noise() -> Result<()> {
+    let Some((current, released)) = version_skew_binaries()? else {
+        return Ok(());
+    };
+    assert_noise_version_skew(&released, &current).await
+}
+
+fn version_skew_binaries() -> Result<Option<(PathBuf, PathBuf)>> {
+    let Some(released) = std::env::var_os(RELEASED_CODEX_ENV_VAR) else {
+        return Ok(None);
+    };
+    let current = std::env::var_os(CURRENT_CODEX_ENV_VAR)
+        .with_context(|| format!("{CURRENT_CODEX_ENV_VAR} must name the current Codex binary"))?;
+    let current = PathBuf::from(current);
+    let released = PathBuf::from(released);
+    anyhow::ensure!(
+        current.is_file(),
+        "current Codex does not exist: {}",
+        current.display()
+    );
+    anyhow::ensure!(
+        released.is_file(),
+        "released Codex does not exist: {}",
+        released.display()
+    );
+    Ok(Some((current, released)))
+}
+
+async fn assert_noise_version_skew(app_binary: &Path, executor_binary: &Path) -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let model = mock_model(codex_home.path()).await?;
+    let model_url = model.uri();
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "danger-full-access"
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider"
+base_url = "{model_url}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )?;
+
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let rendezvous_url = format!("ws://{}", listener.local_addr()?);
+    let registry = MockServer::start().await;
+    let registry_url = registry.uri();
+    Mock::given(method("POST"))
+        .and(path(format!(
+            "/cloud/environment/{ENVIRONMENT_ID}/register"
+        )))
+        .and(header("authorization", format!("Bearer {REGISTRY_TOKEN}")))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "environment_id": ENVIRONMENT_ID,
+            "url": format!("{rendezvous_url}/relay?role=environment"),
+            "security_profile": "noise_hybrid_ik_v1",
+            "executor_registration_id": EXECUTOR_REGISTRATION_ID,
+        })))
+        .expect(1)
+        .mount(&registry)
+        .await;
+    Mock::given(method("POST"))
+        .and(path(format!(
+            "/cloud/environment/{ENVIRONMENT_ID}/validate"
+        )))
+        .and(header("authorization", format!("Bearer {REGISTRY_TOKEN}")))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"valid": true})))
+        .expect(1)
+        .mount(&registry)
+        .await;
+
+    let mut executor = Command::new(executor_binary)
+        .args([
+            "exec-server",
+            "--remote",
+            registry_url.as_str(),
+            "--environment-id",
+            ENVIRONMENT_ID,
+        ])
+        .current_dir(codex_home.path())
+        .env("CODEX_HOME", codex_home.path())
+        .env("CODEX_API_KEY", REGISTRY_TOKEN)
+        .env(EXECUTOR_MARKER_ENV_VAR, EXPECTED_OUTPUT)
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::inherit())
+        .kill_on_drop(true)
+        .spawn()
+        .with_context(|| format!("start remote executor from {}", executor_binary.display()))?;
+
+    let environment_websocket = accept_websocket(&listener, "environment").await?;
+    let executor_public_key = registered_executor_public_key(&registry).await?;
+    Mock::given(method("POST"))
+        .and(path(format!("/cloud/environment/{ENVIRONMENT_ID}/connect")))
+        .and(header("authorization", format!("Bearer {REGISTRY_TOKEN}")))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "environment_id": ENVIRONMENT_ID,
+            "url": format!("{rendezvous_url}/relay?role=harness"),
+            "security_profile": "noise_hybrid_ik_v1",
+            "executor_registration_id": EXECUTOR_REGISTRATION_ID,
+            "executor_public_key": executor_public_key,
+            "harness_key_authorization": HARNESS_KEY_AUTHORIZATION,
+        })))
+        .expect(1)
+        .mount(&registry)
+        .await;
+
+    let captured_frames = Arc::new(Mutex::new(Vec::new()));
+    let captured_relay_frames = Arc::clone(&captured_frames);
+    let relay = tokio::spawn(async move {
+        let harness_websocket = accept_websocket(&listener, "harness").await?;
+        proxy_relay_frames(
+            environment_websocket,
+            harness_websocket,
+            captured_relay_frames,
+        )
+        .await
+    });
+
+    let mut app_server = Command::new(app_binary)
+        .arg("app-server")
+        .current_dir(codex_home.path())
+        .env("CODEX_HOME", codex_home.path())
+        .env("CODEX_API_KEY", REGISTRY_TOKEN)
+        .env(
+            "CODEX_APP_SERVER_MANAGED_CONFIG_PATH",
+            codex_home.path().join("managed_config.toml"),
+        )
+        .env(CODEX_EXEC_SERVER_NOISE_REGISTRY_URL_ENV_VAR, &registry_url)
+        .env(
+            CODEX_EXEC_SERVER_NOISE_ENVIRONMENT_ID_ENV_VAR,
+            ENVIRONMENT_ID,
+        )
+        .env(CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, REGISTRY_TOKEN)
+        .env_remove(CODEX_EXEC_SERVER_URL_ENV_VAR)
+        .env_remove(EXECUTOR_MARKER_ENV_VAR)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .kill_on_drop(true)
+        .spawn()
+        .with_context(|| format!("start app-server from {}", app_binary.display()))?;
+    let mut stdin = app_server.stdin.take().context("app-server stdin")?;
+    let stdout = app_server.stdout.take().context("app-server stdout")?;
+    let mut stdout = BufReader::new(stdout).lines();
+    let mut notifications = Vec::new();
+
+    app_server_request(
+        &mut stdin,
+        &mut stdout,
+        &mut notifications,
+        /*id*/ 1,
+        "initialize",
+        json!({
+            "clientInfo": {"name": "noise-version-skew", "version": "0.1.0"},
+            "capabilities": {"experimentalApi": true},
+        }),
+    )
+    .await?;
+    stdin.write_all(b"{\"method\":\"initialized\"}\n").await?;
+
+    let thread = app_server_request(
+        &mut stdin,
+        &mut stdout,
+        &mut notifications,
+        /*id*/ 2,
+        "thread/start",
+        json!({"cwd": codex_home.path()}),
+    )
+    .await?;
+    let thread_id = thread["thread"]["id"]
+        .as_str()
+        .context("thread/start should return a thread id")?;
+    app_server_request(
+        &mut stdin,
+        &mut stdout,
+        &mut notifications,
+        /*id*/ 3,
+        "turn/start",
+        json!({
+            "threadId": thread_id,
+            "input": [{
+                "type": "text",
+                "text": "run the Noise compatibility command",
+                "textElements": [],
+            }],
+        }),
+    )
+    .await?;
+
+    while !notifications
+        .iter()
+        .any(|notification: &Value| notification["method"] == "turn/completed")
+    {
+        let line = timeout(VERSION_SKEW_TIMEOUT, stdout.next_line())
+            .await
+            .context("waiting for turn/completed")??
+            .context("app-server exited before turn/completed")?;
+        notifications.push(serde_json::from_str(&line)?);
+    }
+
+    assert_eq!(
+        std::fs::read_to_string(codex_home.path().join("version-skew-output.txt"))?,
+        EXPECTED_OUTPUT
+    );
+    assert_relay_data_is_encrypted(&captured_frames)?;
+    registry.verify().await;
+    model.verify().await;
+
+    let _ = app_server.start_kill();
+    let _ = executor.start_kill();
+    relay.abort();
+    Ok(())
+}
+
+async fn app_server_request(
+    stdin: &mut ChildStdin,
+    stdout: &mut Lines<BufReader<ChildStdout>>,
+    notifications: &mut Vec<Value>,
+    id: u64,
+    method: &str,
+    params: Value,
+) -> Result<Value> {
+    let request = json!({"id": id, "method": method, "params": params});
+    stdin.write_all(request.to_string().as_bytes()).await?;
+    stdin.write_all(b"\n").await?;
+
+    loop {
+        let line = timeout(VERSION_SKEW_TIMEOUT, stdout.next_line())
+            .await
+            .with_context(|| format!("waiting for {method} response"))??
+            .with_context(|| format!("app-server exited before {method} response"))?;
+        let response: Value = serde_json::from_str(&line)?;
+        if response["id"] == id {
+            anyhow::ensure!(
+                response.get("error").is_none(),
+                "{method} failed: {}",
+                response["error"]
+            );
+            return response.get("result").cloned().context("missing result");
+        }
+        notifications.push(response);
+    }
+}
+
+async fn mock_model(codex_home: &Path) -> Result<MockServer> {
+    let server = MockServer::start().await;
+    let arguments = serde_json::to_string(&json!({
+        "cmd": format!("printf '%s' \"${EXECUTOR_MARKER_ENV_VAR}\" > version-skew-output.txt"),
+        "workdir": codex_home,
+        "yield_time_ms": 5_000,
+    }))?;
+    let completed = |id| {
+        json!({
+            "type": "response.completed",
+            "response": {
+                "id": id,
+                "usage": {
+                    "input_tokens": 0,
+                    "input_tokens_details": null,
+                    "output_tokens": 0,
+                    "output_tokens_details": null,
+                    "total_tokens": 0,
+                },
+            },
+        })
+    };
+    let responses = vec![
+        event_stream(vec![
+            json!({"type": "response.created", "response": {"id": "response-1"}}),
+            json!({
+                "type": "response.output_item.done",
+                "item": {
+                    "type": "function_call",
+                    "call_id": "noise-version-skew-command",
+                    "name": "exec_command",
+                    "arguments": arguments,
+                },
+            }),
+            completed("response-1"),
+        ])?,
+        event_stream(vec![
+            json!({"type": "response.created", "response": {"id": "response-2"}}),
+            json!({
+                "type": "response.output_item.done",
+                "item": {
+                    "type": "message",
+                    "role": "assistant",
+                    "id": "message-1",
+                    "content": [{"type": "output_text", "text": "done"}],
+                },
+            }),
+            completed("response-2"),
+        ])?,
+    ];
+    for response in responses {
+        Mock::given(method("POST"))
+            .and(path("/v1/responses"))
+            .respond_with(ResponseTemplate::new(200).set_body_raw(response, "text/event-stream"))
+            .up_to_n_times(1)
+            .expect(1)
+            .mount(&server)
+            .await;
+    }
+    Ok(server)
+}
+
+fn event_stream(events: Vec<Value>) -> Result<String> {
+    events
+        .into_iter()
+        .map(|event| {
+            let event_type = event["type"].as_str().context("SSE event type")?;
+            Ok(format!("event: {event_type}\ndata: {event}\n\n"))
+        })
+        .collect()
+}