diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 3d98479749..c947a4973d 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -409,11 +409,14 @@ dependencies = [ "codex-protocol", "codex-utils-cargo-bin", "core_test_support", + "pretty_assertions", "serde", "serde_json", "shlex", "tempfile", "tokio", + "tokio-util", + "url", "uuid", "wiremock", ] diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index bde8048e76..e32607bd00 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -12,6 +12,10 @@ path = "src/main.rs" name = "codex-app-server-test-notify-capture" path = "src/bin/notify_capture.rs" +[[bin]] +name = "exec-server" +path = "src/bin/exec_server.rs" + [lib] name = "codex_app_server" path = "src/lib.rs" diff --git a/codex-rs/app-server/src/bin/exec_server.rs b/codex-rs/app-server/src/bin/exec_server.rs new file mode 100644 index 0000000000..abf58a246b --- /dev/null +++ b/codex-rs/app-server/src/bin/exec_server.rs @@ -0,0 +1,31 @@ +//! Cargo entry point for the minimal exec-server integration-test fixture. +//! +//! This mirrors `//codex-rs/exec-server/testing:exec-server` so Cargo-backed +//! app-server integration tests can receive `CARGO_BIN_EXE_exec-server`. It +//! also handles the filesystem-helper argv mode because exec-server re-execs +//! `codex_self_exe` for sandboxed filesystem requests. + +use codex_exec_server::ExecServerRuntimePaths; +use std::ffi::OsStr; + +const CODEX_LINUX_SANDBOX_EXE_ENV_VAR: &str = "CODEX_TEST_LINUX_SANDBOX_EXE"; + +fn main() -> Result<(), Box> { + let mut args = std::env::args_os(); + let _ = args.next(); + if args.next().as_deref() == Some(OsStr::new(codex_exec_server::CODEX_FS_HELPER_ARG1)) { + codex_exec_server::run_fs_helper_main(); + } + + let current_exe = std::env::current_exe()?; + let codex_linux_sandbox_exe = + std::env::var_os(CODEX_LINUX_SANDBOX_EXE_ENV_VAR).map(std::path::PathBuf::from); + let runtime_paths = ExecServerRuntimePaths::new(current_exe, codex_linux_sandbox_exe)?; + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()? + .block_on(codex_exec_server::run_main( + "ws://127.0.0.1:0", + runtime_paths, + )) +} diff --git a/codex-rs/app-server/tests/common/Cargo.toml b/codex-rs/app-server/tests/common/Cargo.toml index 8123346dc6..8dd43051c7 100644 --- a/codex-rs/app-server/tests/common/Cargo.toml +++ b/codex-rs/app-server/tests/common/Cargo.toml @@ -6,7 +6,6 @@ license.workspace = true [lib] path = "lib.rs" -test = false doctest = false [lints] @@ -29,12 +28,22 @@ serde = { workspace = true } serde_json = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = [ + "io-util", "io-std", "macros", + "net", "process", "rt-multi-thread", + "sync", + "test-util", + "time", ] } +tokio-util = { workspace = true } +url = { workspace = true } uuid = { workspace = true } wiremock = { workspace = true } core_test_support = { path = "../../../core/tests/common" } shlex = { workspace = true } + +[dev-dependencies] +pretty_assertions = { workspace = true } diff --git a/codex-rs/app-server/tests/common/lib.rs b/codex-rs/app-server/tests/common/lib.rs index f9dee08a5b..d39340a899 100644 --- a/codex-rs/app-server/tests/common/lib.rs +++ b/codex-rs/app-server/tests/common/lib.rs @@ -4,10 +4,12 @@ mod analytics_server; mod auth_fixtures; mod config; mod json_logging; +mod local_websocket_exec_server; mod mock_model_server; mod models_cache; mod responses; mod rollout; +mod rpc_delay; mod test_app_server; pub use analytics_server::start_analytics_events_server; diff --git a/codex-rs/app-server/tests/common/local_websocket_exec_server.rs b/codex-rs/app-server/tests/common/local_websocket_exec_server.rs new file mode 100644 index 0000000000..77e17f8e4e --- /dev/null +++ b/codex-rs/app-server/tests/common/local_websocket_exec_server.rs @@ -0,0 +1,90 @@ +use std::path::Path; +use std::process::Stdio; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use tokio::io::AsyncBufReadExt; +use tokio::io::BufReader; +use tokio::process::Child; +use tokio::process::Command; + +const START_TIMEOUT: Duration = Duration::from_secs(10); +#[cfg(target_os = "linux")] +const CODEX_LINUX_SANDBOX_EXE_ENV_VAR: &str = "CODEX_TEST_LINUX_SANDBOX_EXE"; + +/// Host-local exec-server fixture that exposes a WebSocket URL. +/// +/// This is distinct from the ordinary local stdio executor: callers use it +/// when they need a socket transport they can interpose. +pub(crate) struct LocalWebsocketExecServer { + child: Child, + websocket_url: String, +} + +impl LocalWebsocketExecServer { + pub(crate) async fn start(codex_home: &Path, exec_server_program: &Path) -> Result { + let mut command = Command::new(exec_server_program); + command.stdin(Stdio::null()); + command.stdout(Stdio::piped()); + command.stderr(Stdio::inherit()); + command.current_dir(codex_home); + command.env("CODEX_HOME", codex_home); + #[cfg(target_os = "linux")] + command.env( + CODEX_LINUX_SANDBOX_EXE_ENV_VAR, + core_test_support::find_codex_linux_sandbox_exe() + .context("should find binary for delayed exec-server Linux sandbox helper")?, + ); + command.kill_on_drop(true); + let child = command.spawn().context("start local exec-server fixture")?; + let mut exec_server = Self { + child, + websocket_url: String::new(), + }; + let stdout = exec_server + .child + .stdout + .take() + .ok_or_else(|| anyhow!("local exec-server fixture stdout was not captured"))?; + let mut lines = BufReader::new(stdout).lines(); + let deadline = tokio::time::Instant::now() + START_TIMEOUT; + exec_server.websocket_url = loop { + let remaining = deadline + .checked_duration_since(tokio::time::Instant::now()) + .ok_or_else(|| anyhow!("timed out waiting for local exec-server listen URL"))?; + let line = tokio::time::timeout(remaining, lines.next_line()) + .await + .map_err(|_| anyhow!("timed out waiting for local exec-server listen URL"))?? + .ok_or_else(|| { + anyhow!("local exec-server exited before emitting its listen URL") + })?; + let listen_url = line.trim(); + if listen_url.starts_with("ws://") { + break listen_url.to_string(); + } + }; + Ok(exec_server) + } + + pub(crate) fn websocket_url(&self) -> &str { + &self.websocket_url + } +} + +impl Drop for LocalWebsocketExecServer { + fn drop(&mut self) { + let _ = self.child.start_kill(); + + let start = std::time::Instant::now(); + let timeout = Duration::from_secs(5); + while start.elapsed() < timeout { + match self.child.try_wait() { + Ok(Some(_)) => return, + Ok(None) => std::thread::sleep(Duration::from_millis(10)), + Err(_) => return, + } + } + } +} diff --git a/codex-rs/app-server/tests/common/rpc_delay.rs b/codex-rs/app-server/tests/common/rpc_delay.rs new file mode 100644 index 0000000000..9b702aa8f9 --- /dev/null +++ b/codex-rs/app-server/tests/common/rpc_delay.rs @@ -0,0 +1,157 @@ +use std::io; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use tokio::io::AsyncRead; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWrite; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio::task::JoinSet; +use tokio::time::Instant; +use tokio::time::sleep_until; +use tokio_util::task::AbortOnDropHandle; +use url::Host; +use url::Url; + +const FORWARD_BUFFER_BYTES: usize = 64 * 1024; +const FORWARD_QUEUE_CHUNKS: usize = 16; + +pub(crate) struct WebsocketDelayInterposer { + websocket_url: String, + accept_task: JoinHandle<()>, +} + +impl WebsocketDelayInterposer { + pub(crate) async fn start(upstream_url: &str, added_delay: Duration) -> Result { + let upstream = websocket_authority(upstream_url)?; + let listener = TcpListener::bind("127.0.0.1:0") + .await + .context("bind RPC delay interposer")?; + let websocket_url = format!("ws://{}", listener.local_addr()?); + let accept_task = tokio::spawn(async move { + let mut connections = JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((downstream, _peer)) = accepted else { + break; + }; + let upstream = upstream.clone(); + connections.spawn(async move { + let Ok(upstream) = TcpStream::connect(upstream).await else { + return; + }; + let _ = proxy_connection(downstream, upstream, added_delay).await; + }); + } + _ = connections.join_next(), if !connections.is_empty() => {} + } + } + }); + Ok(Self { + websocket_url, + accept_task, + }) + } + + pub(crate) fn websocket_url(&self) -> &str { + &self.websocket_url + } +} + +impl Drop for WebsocketDelayInterposer { + fn drop(&mut self) { + self.accept_task.abort(); + } +} + +fn websocket_authority(websocket_url: &str) -> Result { + let websocket_url = Url::parse(websocket_url).context("parse RPC delay upstream URL")?; + if websocket_url.scheme() != "ws" { + return Err(anyhow!("RPC delay requires a ws:// exec-server URL")); + } + let host = websocket_url + .host() + .ok_or_else(|| anyhow!("RPC delay exec-server URL has no host"))?; + let port = websocket_url + .port_or_known_default() + .ok_or_else(|| anyhow!("RPC delay exec-server URL has no port"))?; + let host = match host { + Host::Domain(host) => host.to_string(), + Host::Ipv4(host) => host.to_string(), + Host::Ipv6(host) => format!("[{host}]"), + }; + Ok(format!("{host}:{port}")) +} + +async fn proxy_connection( + downstream: TcpStream, + upstream: TcpStream, + added_delay: Duration, +) -> io::Result<()> { + let (downstream_read, downstream_write) = downstream.into_split(); + let (upstream_read, upstream_write) = upstream.into_split(); + let client_to_server = forward_direction(downstream_read, upstream_write, added_delay); + let server_to_client = forward_direction(upstream_read, downstream_write, added_delay); + tokio::try_join!(client_to_server, server_to_client)?; + Ok(()) +} + +async fn forward_direction( + mut reader: R, + mut writer: W, + added_delay: Duration, +) -> io::Result<()> +where + R: AsyncRead + Unpin + Send + 'static, + W: AsyncWrite + Unpin + Send + 'static, +{ + // tokio::io::copy would wait before reading the next chunk, turning a + // fixed propagation delay into a bandwidth limit. Timestamping reads into + // a bounded queue lets close-together chunks emerge close together after + // the same delay while still applying backpressure. + let (tx, mut rx) = mpsc::channel::(FORWARD_QUEUE_CHUNKS); + let reader_task = AbortOnDropHandle::new(tokio::spawn(async move { + loop { + let mut bytes = vec![0; FORWARD_BUFFER_BYTES]; + let read = reader.read(&mut bytes).await?; + if read == 0 { + break; + } + bytes.truncate(read); + let chunk = DelayedChunk { + deliver_at: Instant::now() + added_delay, + bytes, + }; + if tx.send(chunk).await.is_err() { + break; + } + } + Ok::<(), io::Error>(()) + })); + + while let Some(chunk) = rx.recv().await { + sleep_until(chunk.deliver_at).await; + writer.write_all(&chunk.bytes).await?; + } + writer.shutdown().await?; + reader_task + .await + .map_err(|err| io::Error::other(format!("RPC delay reader task failed: {err}")))??; + Ok(()) +} + +struct DelayedChunk { + deliver_at: Instant, + bytes: Vec, +} + +#[cfg(test)] +#[path = "rpc_delay_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/tests/common/rpc_delay_tests.rs b/codex-rs/app-server/tests/common/rpc_delay_tests.rs new file mode 100644 index 0000000000..6ae9e06229 --- /dev/null +++ b/codex-rs/app-server/tests/common/rpc_delay_tests.rs @@ -0,0 +1,181 @@ +use std::collections::VecDeque; +use std::io; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; +use std::time::Duration; + +use pretty_assertions::assert_eq; +use tokio::io::AsyncRead; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWrite; +use tokio::io::AsyncWriteExt; +use tokio::io::ReadBuf; +use tokio::io::duplex; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::time::advance; +use tokio::time::timeout; + +use super::WebsocketDelayInterposer; +use super::forward_direction; +use super::websocket_authority; + +#[tokio::test(start_paused = true)] +async fn delays_then_flushes_eof() -> anyhow::Result<()> { + let delay = Duration::from_millis(15); + let (mut input_writer, input_reader) = duplex(/*max_buf_size*/ 64); + let (output_writer, mut output_reader) = duplex(/*max_buf_size*/ 64); + let forward_task = tokio::spawn(forward_direction(input_reader, output_writer, delay)); + + input_writer.write_all(b"payload").await?; + input_writer.shutdown().await?; + tokio::task::yield_now().await; + + let mut before_delay = [0; 1]; + assert!( + timeout(Duration::ZERO, output_reader.read(&mut before_delay)) + .await + .is_err() + ); + + advance(delay).await; + let mut output = Vec::new(); + output_reader.read_to_end(&mut output).await?; + assert_eq!(output, b"payload"); + forward_task.await??; + Ok(()) +} + +#[tokio::test(start_paused = true)] +async fn burst_chunks_share_one_deadline() -> anyhow::Result<()> { + let delay = Duration::from_millis(15); + let (mut input_writer, input_reader) = duplex(/*max_buf_size*/ 1); + let (output_writer, mut output_reader) = duplex(/*max_buf_size*/ 64); + let forward_task = tokio::spawn(forward_direction(input_reader, output_writer, delay)); + + input_writer.write_all(b"ab").await?; + input_writer.shutdown().await?; + tokio::task::yield_now().await; + + advance(delay).await; + tokio::task::yield_now().await; + let mut output = [0; 2]; + timeout(Duration::ZERO, output_reader.read_exact(&mut output)).await??; + assert_eq!(output, *b"ab"); + forward_task.await??; + Ok(()) +} + +#[tokio::test] +async fn write_error_cancels_reader() { + let reader_dropped = Arc::new(AtomicBool::new(false)); + let reader = PendingAfterChunkReader::new(Arc::clone(&reader_dropped)); + + let error = forward_direction(reader, FailingWriter, Duration::ZERO) + .await + .expect_err("failing writer should fail forwarding"); + assert_eq!(error.kind(), io::ErrorKind::BrokenPipe); + + for _ in 0..10 { + if reader_dropped.load(Ordering::Acquire) { + break; + } + tokio::task::yield_now().await; + } + assert!(reader_dropped.load(Ordering::Acquire)); +} + +#[tokio::test] +async fn zero_delay_loopback_forwards_and_closes_active_sockets() -> anyhow::Result<()> { + let upstream_listener = TcpListener::bind("127.0.0.1:0").await?; + let upstream_url = format!("ws://{}", upstream_listener.local_addr()?); + let upstream_task = tokio::spawn(async move { + let (mut upstream, _) = upstream_listener.accept().await?; + let mut request = [0; 5]; + upstream.read_exact(&mut request).await?; + assert_eq!(request, *b"hello"); + upstream.write_all(b"world").await?; + + let mut after_drop = [0; 1]; + let read = timeout(Duration::from_secs(1), upstream.read(&mut after_drop)).await??; + Ok::(read) + }); + + let interposer = WebsocketDelayInterposer::start(&upstream_url, Duration::ZERO).await?; + let mut downstream = + TcpStream::connect(websocket_authority(interposer.websocket_url())?).await?; + downstream.write_all(b"hello").await?; + let mut response = [0; 5]; + downstream.read_exact(&mut response).await?; + assert_eq!(response, *b"world"); + + drop(interposer); + tokio::task::yield_now().await; + + let mut after_drop = [0; 1]; + let read = timeout(Duration::from_secs(1), downstream.read(&mut after_drop)).await??; + assert_eq!(read, 0); + assert_eq!(upstream_task.await??, 0); + Ok(()) +} + +struct PendingAfterChunkReader { + chunks: VecDeque>, + dropped: Arc, +} + +impl PendingAfterChunkReader { + fn new(dropped: Arc) -> Self { + Self { + chunks: VecDeque::from([b"x".to_vec()]), + dropped, + } + } +} + +impl AsyncRead for PendingAfterChunkReader { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let Some(chunk) = self.chunks.pop_front() else { + return Poll::Pending; + }; + buf.put_slice(&chunk); + Poll::Ready(Ok(())) + } +} + +impl Drop for PendingAfterChunkReader { + fn drop(&mut self) { + self.dropped.store(true, Ordering::Release); + } +} + +struct FailingWriter; + +impl AsyncWrite for FailingWriter { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &[u8], + ) -> Poll> { + Poll::Ready(Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "synthetic write failure", + ))) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index 1a74e1df5a..825ea46ab8 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -5,6 +5,7 @@ use std::process::ExitStatus; use std::process::Stdio; use std::sync::atomic::AtomicI64; use std::sync::atomic::Ordering; +use std::time::Duration; use tokio::io::AsyncBufReadExt; use tokio::io::AsyncWriteExt; use tokio::io::BufReader; @@ -121,12 +122,15 @@ 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 codex_login::default_client::CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR; +use core_test_support::is_remote_test_environment; use core_test_support::test_codex::TestEnv; use core_test_support::test_codex::test_env; use tempfile::TempDir; use tokio::process::Command; use crate::json_logging::JsonLogCapture; +use crate::local_websocket_exec_server::LocalWebsocketExecServer; +use crate::rpc_delay::WebsocketDelayInterposer; pub struct TestAppServer { next_request_id: AtomicI64, @@ -141,6 +145,9 @@ pub struct TestAppServer { auto_env: Option, json_logs: JsonLogCapture, codex_home: PathBuf, + // Fields drop in declaration order. Tear down the delayed child before + // removing an owned CODEX_HOME that may still be its cwd on Windows. + _delayed_exec_server: Option<(LocalWebsocketExecServer, WebsocketDelayInterposer)>, _owned_codex_home: Option, } @@ -159,6 +166,7 @@ impl TestAppServer { program: None, env_overrides: Vec::new(), args: vec![DISABLE_PLUGIN_STARTUP_TASKS_ARG.to_string()], + exec_server_delay: None, } } @@ -283,6 +291,7 @@ impl TestAppServer { auto_env: None, json_logs, codex_home: codex_home.to_path_buf(), + _delayed_exec_server: None, _owned_codex_home: None, }) } @@ -1692,6 +1701,7 @@ pub struct TestAppServerBuilder { program: Option, env_overrides: Vec<(String, Option)>, args: Vec, + exec_server_delay: Option, } enum TestAppServerEnvironment { @@ -1761,6 +1771,13 @@ impl TestAppServerBuilder { builder } + /// Adds this fixed one-way delay to the app-server/exec-server RPC stream. + /// A 15ms delay contributes roughly 30ms to a round trip. + pub fn with_exec_server_delay(mut self, exec_server_delay: Duration) -> Self { + self.exec_server_delay = Some(exec_server_delay); + self + } + /// Builds a server with a temporary CODEX_HOME and automatic environment /// by default. pub async fn build(self) -> anyhow::Result { @@ -1770,6 +1787,7 @@ impl TestAppServerBuilder { program, mut env_overrides, args, + exec_server_delay, } = self; let (codex_home, owned_codex_home) = match codex_home { Some(codex_home) => (codex_home, None), @@ -1781,7 +1799,7 @@ impl TestAppServerBuilder { ) } }; - let auto_env = match environment { + let (auto_env, delayed_exec_server) = match environment { TestAppServerEnvironment::Auto => { let environments_toml = codex_home.join("environments.toml"); ensure!( @@ -1792,7 +1810,34 @@ impl TestAppServerBuilder { "automatic environment cannot be used when {} exists", environments_toml.display() ); - let auto_env = test_env().await?; + let (auto_env, delayed_exec_server) = match exec_server_delay { + Some(added_delay) => { + ensure!( + !is_remote_test_environment(), + "TestAppServer exec-server delay only supports the local test environment" + ); + let exec_server_program = + codex_utils_cargo_bin::cargo_bin("exec-server") + .context("should find binary for delayed exec-server fixture")?; + // Local auto environments normally use stdio. Start a + // host-local WebSocket fixture so the delay interposer has a + // socket stream to wrap. + let local_websocket_exec_server = + LocalWebsocketExecServer::start(&codex_home, &exec_server_program) + .await?; + let interposer = WebsocketDelayInterposer::start( + local_websocket_exec_server.websocket_url(), + added_delay, + ) + .await?; + let auto_env = TestEnv::local_with_exec_server_url(Some( + interposer.websocket_url().to_string(), + )) + .await?; + (auto_env, Some((local_websocket_exec_server, interposer))) + } + None => (test_env().await?, None), + }; // Noise registry configuration takes precedence over the URL-based // provider, so clear inherited values to keep the selection hermetic. let mut auto_env_overrides = vec![ @@ -1816,9 +1861,15 @@ impl TestAppServerBuilder { ]; auto_env_overrides.append(&mut env_overrides); env_overrides = auto_env_overrides; - Some(auto_env) + (Some(auto_env), delayed_exec_server) + } + TestAppServerEnvironment::None => { + ensure!( + exec_server_delay.is_none(), + "exec-server delay requires the automatic test environment" + ); + (None, None) } - TestAppServerEnvironment::None => None, }; if !env_overrides .iter() @@ -1853,6 +1904,7 @@ impl TestAppServerBuilder { .await?; app_server.auto_env = auto_env; app_server._owned_codex_home = owned_codex_home; + app_server._delayed_exec_server = delayed_exec_server; Ok(app_server) } } diff --git a/codex-rs/app-server/tests/suite/v2/auto_env.rs b/codex-rs/app-server/tests/suite/v2/auto_env.rs index e3f3c30755..ccf80e37f1 100644 --- a/codex-rs/app-server/tests/suite/v2/auto_env.rs +++ b/codex-rs/app-server/tests/suite/v2/auto_env.rs @@ -10,14 +10,53 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::UserInput as V2UserInput; use core_test_support::responses; +use core_test_support::skip_if_host_windows; +use core_test_support::skip_if_remote; use pretty_assertions::assert_eq; use std::collections::BTreeMap; use std::time::Duration; +use std::time::Instant; use tempfile::TempDir; use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); +#[tokio::test] +async fn builder_interposes_fixed_delay_for_auto_env() -> Result<()> { + skip_if_host_windows!(Ok(())); + skip_if_remote!(Ok(()), "the fixed-delay fixture is local-only"); + + let codex_home = TempDir::new()?; + let requested_delay = Duration::from_secs(1); + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_exec_server_delay(requested_delay) + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + assert_eq!( + mcp.auto_env_params()?.environment_id, + codex_exec_server::REMOTE_ENVIRONMENT_ID + ); + + let thread_start = Instant::now(); + let request_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams::default()) + .await?; + let _: JSONRPCResponse = timeout( + Duration::from_secs(60), + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let elapsed = thread_start.elapsed(); + assert!( + elapsed >= requested_delay, + "thread/start completed in {elapsed:?}, below the requested {requested_delay:?} delay" + ); + + Ok(()) +} + #[tokio::test] async fn thread_start_with_auto_env_exposes_fixture_cwd_to_model() -> Result<()> { let server = responses::start_mock_server().await; diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 62624d9e3c..82f66dbb54 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -128,14 +128,26 @@ pub struct TestEnv { impl TestEnv { pub async fn local() -> Result { + Self::local_with_exec_server_url(/*exec_server_url*/ None).await + } + + /// Builds a host-local test environment, optionally using the provided + /// exec-server URL instead of the normal implicit local executor. + pub async fn local_with_exec_server_url(exec_server_url: Option) -> Result { let local_cwd_temp_dir = Arc::new(TempDir::new()?); let cwd = local_cwd_temp_dir.abs(); + let selection = match exec_server_url { + Some(_) => TurnEnvironmentSelection { + environment_id: codex_exec_server::REMOTE_ENVIRONMENT_ID.to_string(), + cwd: PathUri::from_abs_path(&cwd), + }, + None => local(cwd.clone()), + }; let environment = - codex_exec_server::Environment::create_for_tests(/*exec_server_url*/ None)?; - let selection = local(cwd.clone()); + codex_exec_server::Environment::create_for_tests(exec_server_url.clone())?; Ok(Self { environment, - exec_server_url: None, + exec_server_url, cwd, selection, local_cwd_temp_dir: Some(local_cwd_temp_dir),