test: add delayed exec-server transport (#31427)

## Why

Macrobenchmarks benefit from having a way to exercise remote-executor
latency without depending on Docker.

This is a very minimal first cut, if we find that simulating network
conditions is useful we can always expand this scope or switch to a more
robust network shaping approach.

## What

- add a package-local exec-server binary for Cargo and Bazel test
fixtures
- add a host-local WebSocket exec-server fixture and fixed-delay
interposer
- let TestAppServer route its auto environment through that delayed
WebSocket transport
- cover the delayed thread/start path through the public app-server API

## Stack

1. [#31425 test: add TestAppServer
builder](https://github.com/openai/codex/pull/31425)
2. [#31427 test: add delayed exec-server
transport](https://github.com/openai/codex/pull/31427)
3. [#31295 bench: add cold skill load
macrobenchmark](https://github.com/openai/codex/pull/31295)
4. [#31428 bench: add e2e benchmark
entrypoints](https://github.com/openai/codex/pull/31428)
5. [#31429 ci: smoke Bazel e2e
benchmarks](https://github.com/openai/codex/pull/31429)

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",
 ]

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"

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<dyn std::error::Error + Send + Sync>> {
+    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,
+        ))
+}

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 }

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;

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<Self> {
+        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,
+            }
+        }
+    }
+}

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<Self> {
+        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<String> {
+    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<R, W>(
+    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::<DelayedChunk>(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<u8>,
+}
+
+#[cfg(test)]
+#[path = "rpc_delay_tests.rs"]
+mod tests;

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::<usize, anyhow::Error>(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<Vec<u8>>,
+    dropped: Arc<AtomicBool>,
+}
+
+impl PendingAfterChunkReader {
+    fn new(dropped: Arc<AtomicBool>) -> 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<io::Result<()>> {
+        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<io::Result<usize>> {
+        Poll::Ready(Err(io::Error::new(
+            io::ErrorKind::BrokenPipe,
+            "synthetic write failure",
+        )))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}

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<TestEnv>,
     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<TempDir>,
 }
 
@@ -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<PathBuf>,
     env_overrides: Vec<(String, Option<String>)>,
     args: Vec<String>,
+    exec_server_delay: Option<Duration>,
 }
 
 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<TestAppServer> {
@@ -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)
     }
 }

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;

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

@@ -128,14 +128,26 @@ pub struct TestEnv {
 
 impl TestEnv {
     pub async fn local() -> Result<Self> {
+        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<String>) -> Result<Self> {
         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),