diff --git a/codex-rs/exec-server-protocol/src/lib.rs b/codex-rs/exec-server-protocol/src/lib.rs index 069696d87d..969da3f9d9 100644 --- a/codex-rs/exec-server-protocol/src/lib.rs +++ b/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"; diff --git a/codex-rs/exec-server/testing/run_version_skew.sh b/codex-rs/exec-server/testing/run_version_skew.sh new file mode 100755 index 0000000000..f0e9536def --- /dev/null +++ b/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 diff --git a/codex-rs/exec-server/tests/relay.rs b/codex-rs/exec-server/tests/relay.rs index 1b03096df6..2cfa7b62a8 100644 --- a/codex-rs/exec-server/tests/relay.rs +++ b/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; diff --git a/codex-rs/exec-server/tests/relay/version_skew.rs b/codex-rs/exec-server/tests/relay/version_skew.rs new file mode 100644 index 0000000000..bfac106312 --- /dev/null +++ b/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(¤t, &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, ¤t).await +} + +fn version_skew_binaries() -> Result> { + 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(®istry) + .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(®istry) + .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(®istry).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(®istry) + .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, ®istry_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>, + notifications: &mut Vec, + id: u64, + method: &str, + params: Value, +) -> Result { + 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 { + 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) -> Result { + 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() +}