From 624c69e8406cd79e8afb29e4b268ca3fede8634a Mon Sep 17 00:00:00 2001 From: Thibault Sottiaux Date: Mon, 6 Apr 2026 08:18:46 -0700 Subject: [PATCH] [codex] add response proxy subagent header test (#16876) This adds end-to-end coverage for `responses-api-proxy` request dumps when Codex spawns a subagent and validates that the `x-codex-window-id` and `x-openai-subagent` are properly set. --- codex-rs/core/tests/suite/mod.rs | 1 + .../suite/responses_api_proxy_headers.rs | 239 ++++++++++++++++++ 2 files changed, 240 insertions(+) create mode 100644 codex-rs/core/tests/suite/responses_api_proxy_headers.rs diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 85dbe5e91..4d58537e0 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -122,6 +122,7 @@ mod request_permissions; #[cfg(not(target_os = "windows"))] mod request_permissions_tool; mod request_user_input; +mod responses_api_proxy_headers; mod resume; mod resume_warning; mod review; diff --git a/codex-rs/core/tests/suite/responses_api_proxy_headers.rs b/codex-rs/core/tests/suite/responses_api_proxy_headers.rs new file mode 100644 index 000000000..7dc888ffc --- /dev/null +++ b/codex-rs/core/tests/suite/responses_api_proxy_headers.rs @@ -0,0 +1,239 @@ +//! Exercises a real `responses-api-proxy` process with request dumping enabled, then verifies that +//! parent and spawned subagent requests carry the expected window, parent-thread, and subagent +//! identity headers in the dumped Responses API requests. + +use anyhow::Result; +use anyhow::anyhow; +use codex_features::Feature; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_sse_once_match; +use core_test_support::responses::sse; +use core_test_support::responses::start_mock_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::io::Write; +use std::path::Path; +use std::process::Child; +use std::process::Command as StdCommand; +use std::process::Stdio; +use std::time::Duration; +use std::time::Instant; +use tempfile::TempDir; + +const PARENT_PROMPT: &str = "spawn a subagent and report when it is started"; +const CHILD_PROMPT: &str = "child: say done"; +const SPAWN_CALL_ID: &str = "spawn-call-1"; +const PROXY_START_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 5); +const PROXY_POLL_INTERVAL: Duration = Duration::from_millis(/*millis*/ 20); + +struct ResponsesApiProxy { + child: Child, + port: u16, +} + +impl ResponsesApiProxy { + fn start(upstream_url: &str, dump_dir: &Path) -> Result { + let server_info = dump_dir.join("server-info.json"); + let mut child = StdCommand::new(codex_utils_cargo_bin::cargo_bin("codex")?) + .args(["responses-api-proxy", "--server-info"]) + .arg(&server_info) + .args(["--upstream-url", upstream_url, "--dump-dir"]) + .arg(dump_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + + child + .stdin + .take() + .ok_or_else(|| anyhow!("responses-api-proxy stdin was not piped"))? + .write_all(b"dummy\n")?; + + let deadline = Instant::now() + PROXY_START_TIMEOUT; + loop { + if let Ok(info) = std::fs::read_to_string(&server_info) { + let port = serde_json::from_str::(&info)? + .get("port") + .and_then(Value::as_u64) + .ok_or_else(|| anyhow!("proxy server info missing port"))?; + return Ok(Self { + child, + port: u16::try_from(port)?, + }); + } + if let Some(status) = child.try_wait()? { + return Err(anyhow!( + "responses-api-proxy exited before writing server info: {status}" + )); + } + if Instant::now() >= deadline { + return Err(anyhow!("timed out waiting for responses-api-proxy")); + } + std::thread::sleep(PROXY_POLL_INTERVAL); + } + } + + fn base_url(&self) -> String { + format!("http://127.0.0.1:{}/v1", self.port) + } +} + +impl Drop for ResponsesApiProxy { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn responses_api_proxy_dumps_parent_and_subagent_identity_headers() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let dump_dir = TempDir::new()?; + let proxy = + ResponsesApiProxy::start(&format!("{}/v1/responses", server.uri()), dump_dir.path())?; + + let spawn_args = serde_json::to_string(&json!({ "message": CHILD_PROMPT }))?; + mount_sse_once_match( + &server, + |req: &wiremock::Request| request_body_contains(req, PARENT_PROMPT), + sse(vec![ + ev_response_created("resp-parent-1"), + ev_function_call(SPAWN_CALL_ID, "spawn_agent", &spawn_args), + ev_completed("resp-parent-1"), + ]), + ) + .await; + mount_sse_once_match( + &server, + |req: &wiremock::Request| { + request_body_contains(req, CHILD_PROMPT) && !request_body_contains(req, SPAWN_CALL_ID) + }, + sse(vec![ + ev_response_created("resp-child-1"), + ev_assistant_message("msg-child-1", "child done"), + ev_completed("resp-child-1"), + ]), + ) + .await; + mount_sse_once_match( + &server, + |req: &wiremock::Request| request_body_contains(req, SPAWN_CALL_ID), + sse(vec![ + ev_response_created("resp-parent-2"), + ev_assistant_message("msg-parent-2", "parent done"), + ev_completed("resp-parent-2"), + ]), + ) + .await; + + let proxy_base_url = proxy.base_url(); + let mut builder = test_codex().with_config(move |config| { + config.model_provider.base_url = Some(proxy_base_url); + config + .features + .disable(Feature::EnableRequestCompression) + .expect("test config should allow feature update"); + }); + let test = builder.build(&server).await?; + test.submit_turn(PARENT_PROMPT).await?; + + let dumps = wait_for_proxy_request_dumps(dump_dir.path())?; + let parent = dumps + .iter() + .find(|dump| dump_body_contains(dump, PARENT_PROMPT)) + .ok_or_else(|| anyhow!("missing parent request dump"))?; + let child = dumps + .iter() + .find(|dump| { + dump_body_contains(dump, CHILD_PROMPT) && !dump_body_contains(dump, SPAWN_CALL_ID) + }) + .ok_or_else(|| anyhow!("missing child request dump"))?; + + let parent_window_id = header(parent, "x-codex-window-id") + .ok_or_else(|| anyhow!("parent request missing x-codex-window-id"))?; + let child_window_id = header(child, "x-codex-window-id") + .ok_or_else(|| anyhow!("child request missing x-codex-window-id"))?; + let (parent_thread_id, parent_generation) = split_window_id(parent_window_id)?; + let (child_thread_id, child_generation) = split_window_id(child_window_id)?; + + assert_eq!(parent_generation, 0); + assert_eq!(child_generation, 0); + assert!(child_thread_id != parent_thread_id); + assert_eq!(header(parent, "x-openai-subagent"), None); + assert_eq!(header(child, "x-openai-subagent"), Some("collab_spawn")); + assert_eq!( + header(child, "x-codex-parent-thread-id"), + Some(parent_thread_id) + ); + + Ok(()) +} + +fn request_body_contains(req: &wiremock::Request, text: &str) -> bool { + std::str::from_utf8(&req.body).is_ok_and(|body| body.contains(text)) +} + +fn wait_for_proxy_request_dumps(dump_dir: &Path) -> Result> { + let deadline = Instant::now() + Duration::from_secs(/*secs*/ 2); + loop { + let dumps = read_proxy_request_dumps(dump_dir).unwrap_or_default(); + if dumps.len() >= 3 + && dumps + .iter() + .any(|dump| dump_body_contains(dump, CHILD_PROMPT)) + { + return Ok(dumps); + } + if Instant::now() >= deadline { + return Err(anyhow!( + "timed out waiting for proxy request dumps, got {}", + dumps.len() + )); + } + std::thread::sleep(PROXY_POLL_INTERVAL); + } +} + +fn read_proxy_request_dumps(dump_dir: &Path) -> Result> { + let mut dumps = Vec::new(); + for entry in std::fs::read_dir(dump_dir)? { + let path = entry?.path(); + if path + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with("-request.json")) + { + dumps.push(serde_json::from_str(&std::fs::read_to_string(&path)?)?); + } + } + Ok(dumps) +} + +fn dump_body_contains(dump: &Value, text: &str) -> bool { + dump.get("body") + .is_some_and(|body| body.to_string().contains(text)) +} + +fn header<'a>(dump: &'a Value, name: &str) -> Option<&'a str> { + dump.get("headers")?.as_array()?.iter().find_map(|header| { + (header.get("name")?.as_str()?.eq_ignore_ascii_case(name)) + .then(|| header.get("value")?.as_str()) + .flatten() + }) +} + +fn split_window_id(window_id: &str) -> Result<(&str, u64)> { + let (thread_id, generation) = window_id + .rsplit_once(':') + .ok_or_else(|| anyhow!("invalid window id header: {window_id}"))?; + Ok((thread_id, generation.parse::()?)) +}