Harden core test fixture startup assertions (#41194)

## Why

Test fixture setup can take more than five seconds on loaded CI workers, and
failures returned from setup can be obscured when downstream mocks verify their
expectations during teardown.

## What changed

- Add a shared `expect_startup` helper with a 30-second timeout that panics on
  either timeout or setup failure before mock teardown runs.
- Use the helper for remote-environment and unified-exec fixture setup, and
  allow the same startup window for the initial exec-server requests.

## Testing

Add paused-time tests covering successful startup after five seconds, timeout
reporting, and propagation of setup errors in the presence of unmet mocks.

GitOrigin-RevId: 39db69bde2983acd023a1911a6fad106674e169a

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

@@ -29,6 +29,7 @@ pub mod context_snapshot;
 pub mod hooks;
 pub mod process;
 pub mod responses;
+pub mod startup;
 pub mod streaming_sse;
 pub mod test_codex;
 pub mod test_codex_exec;

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

@@ -0,0 +1,22 @@
+//! Bounded startup assertions that preserve fixture failures during teardown.
+
+use anyhow::Result;
+use std::future::Future;
+use std::time::Duration;
+use tokio::time::timeout;
+
+/// Allow local configuration and state-database setup on loaded CI workers.
+pub const STARTUP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30);
+
+/// Fail before the caller's mock server drops and verifies downstream requests.
+/// Returning an error instead would let an unmet mock expectation mask its cause.
+pub async fn expect_startup<T>(startup: impl Future<Output = Result<T>>) -> T {
+    timeout(STARTUP_TIMEOUT, startup)
+        .await
+        .expect("test fixture startup timed out")
+        .expect("test fixture startup failed")
+}
+
+#[cfg(test)]
+#[path = "startup_tests.rs"]
+mod tests;

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

@@ -0,0 +1,43 @@
+//! Keeps startup failures visible even when downstream mocks have unmet expectations.
+
+use super::expect_startup;
+use crate::responses::ev_completed;
+use crate::responses::mount_sse_sequence;
+use crate::responses::sse;
+use crate::responses::start_mock_server;
+use anyhow::Result;
+use anyhow::anyhow;
+use pretty_assertions::assert_eq;
+use std::future::pending;
+use std::future::ready;
+use std::time::Duration;
+use tokio::time::sleep;
+
+#[tokio::test(start_paused = true)]
+async fn startup_can_complete_after_five_seconds() {
+    let value = expect_startup(async {
+        sleep(Duration::from_secs(/*secs*/ 6)).await;
+        Ok("ready")
+    })
+    .await;
+    assert_eq!(value, "ready");
+}
+
+#[tokio::test(start_paused = true)]
+#[should_panic(expected = "test fixture startup timed out")]
+async fn startup_timeout_is_not_masked_by_mock_expectations() {
+    let server = start_mock_server().await;
+    mount_sse_sequence(&server, vec![sse(vec![ev_completed("unused")])]).await;
+
+    expect_startup(pending::<Result<()>>()).await;
+}
+
+#[tokio::test(start_paused = true)]
+#[should_panic(expected = "injected startup failure")]
+async fn startup_error_is_not_masked_by_mock_expectations() {
+    let server = start_mock_server().await;
+    mount_sse_sequence(&server, vec![sse(vec![ev_completed("unused")])]).await;
+    let startup: Result<()> = Err(anyhow!("injected startup failure"));
+
+    expect_startup(ready(startup)).await;
+}

codex-rs/core/tests/suite/remote_env.rs

@@ -97,6 +97,7 @@ use core_test_support::responses::start_mock_server;
 use core_test_support::skip_if_no_network;
 use core_test_support::skip_if_no_remote_env;
 use core_test_support::skip_if_target_windows;
+use core_test_support::startup::expect_startup;
 use core_test_support::submit_thread_settings;
 use core_test_support::test_codex::TestCodex;
 use core_test_support::test_codex::TestCodexBuilder;
@@ -2133,9 +2134,7 @@ async fn deferred_executor_stays_pending_after_materialization() -> Result<()> {
     let mut builder = test_codex_with_wait_for_environment().with_config(|config| {
         assert!(config.features.enable(Feature::DeferredExecutor).is_ok());
     });
-    let test = timeout(Duration::from_secs(5), builder.build(&server))
-        .await
-        .context("thread startup should not wait for the remote environment")??;
+    let test = expect_startup(builder.build(&server)).await;
     let environment_manager = test.thread_manager.environment_manager();
     let provider = Arc::new(FailingNoiseConnectProvider::default());
     environment_manager.materialize_pending_noise_environment(
@@ -2275,12 +2274,7 @@ async fn deferred_executor_spawn_agent_inherits_ready_step_environments(
         attach_rx,
         shutdown_rx,
     ));
-    let test = timeout(
-        Duration::from_secs(5),
-        builder.build_with_remote_and_local_env(&server),
-    )
-    .await
-    .context("thread startup should not wait for the remote environment")??;
+    let test = expect_startup(builder.build_with_remote_and_local_env(&server)).await;
     let owner_active_profile = ActivePermissionProfile::new("owner-read-only");
     let owner_profile_workspace_root = test.config.cwd.join("owner-profile-root");
     let remote_selection = TurnEnvironmentSelection {
@@ -2440,12 +2434,7 @@ async fn deferred_executor_guardian_uses_newly_ready_step_environment() -> Resul
         attach_rx,
         shutdown_rx,
     ));
-    let test = timeout(
-        Duration::from_secs(5),
-        builder.build_with_remote_and_local_env(&server),
-    )
-    .await
-    .context("thread startup should not wait for the remote environment")??;
+    let test = expect_startup(builder.build_with_remote_and_local_env(&server)).await;
     let remote_cwd = test.cwd.path().join("guardian-remote").abs();
     let local_cwd = test.cwd.path().abs();
     fs::create_dir_all(remote_cwd.as_path())?;
@@ -2591,9 +2580,7 @@ async fn deferred_executor_loads_agents_md_when_environment_becomes_ready() -> R
         attach_rx,
         shutdown_rx,
     ));
-    let test = timeout(Duration::from_secs(5), builder.build(&server))
-        .await
-        .context("thread startup should not wait for the remote environment")??;
+    let test = expect_startup(builder.build(&server)).await;
 
     test.codex
         .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
@@ -2699,9 +2686,7 @@ async fn deferred_executor_compaction_preserves_then_updates_environment_once()
             config.model_context_window = Some(100);
             config.model_auto_compact_token_limit = Some(90);
         });
-    let test = timeout(Duration::from_secs(5), builder.build(&server))
-        .await
-        .context("thread startup should not wait for the remote environment")??;
+    let test = expect_startup(builder.build(&server)).await;
 
     test.codex
         .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {

codex-rs/core/tests/suite/unified_exec_process_events.rs

@@ -34,6 +34,8 @@ use core_test_support::responses::ev_response_created;
 use core_test_support::responses::mount_sse_sequence;
 use core_test_support::responses::sse;
 use core_test_support::responses::start_mock_server;
+use core_test_support::startup::STARTUP_TIMEOUT;
+use core_test_support::startup::expect_startup;
 use core_test_support::test_codex::test_codex;
 use core_test_support::test_codex::turn_permission_fields;
 use futures::SinkExt;
@@ -87,9 +89,12 @@ struct PushedExecServerResult {
     process_start: Value,
 }
 
-async fn read_exec_server_json(websocket: &mut WebSocketStream<TcpStream>) -> Value {
+async fn read_exec_server_json(
+    websocket: &mut WebSocketStream<TcpStream>,
+    wait: Duration,
+) -> Value {
     loop {
-        match timeout(Duration::from_secs(5), websocket.next())
+        match timeout(wait, websocket.next())
             .await
             .expect("websocket read should not time out")
             .expect("websocket should stay open")
@@ -118,7 +123,7 @@ async fn accept_initialized_exec_server(listener: TcpListener) -> WebSocketStrea
     let (stream, _) = listener.accept().await.expect("connection");
     let mut websocket = accept_async(stream).await.expect("websocket handshake");
 
-    let initialize = read_exec_server_json(&mut websocket).await;
+    let initialize = read_exec_server_json(&mut websocket, Duration::from_secs(/*secs*/ 5)).await;
     assert_eq!(initialize["method"], "initialize");
     send_exec_server_json(
         &mut websocket,
@@ -128,7 +133,7 @@ async fn accept_initialized_exec_server(listener: TcpListener) -> WebSocketStrea
         }),
     )
     .await;
-    let initialized = read_exec_server_json(&mut websocket).await;
+    let initialized = read_exec_server_json(&mut websocket, Duration::from_secs(/*secs*/ 5)).await;
     assert_eq!(initialized["method"], "initialized");
 
     websocket
@@ -138,7 +143,7 @@ async fn send_environment_info(
     websocket: &mut WebSocketStream<TcpStream>,
     scenario: PushedExecScenario,
 ) {
-    let info = read_exec_server_json(websocket).await;
+    let info = read_exec_server_json(websocket, STARTUP_TIMEOUT).await;
     assert_eq!(info["method"], "environment/info");
     respond_environment_info(websocket, &info["id"], scenario).await;
 }
@@ -179,7 +184,8 @@ async fn serve_exec_with_pushed_events(
     send_environment_info(&mut websocket, scenario).await;
 
     let process_start = loop {
-        let request = read_exec_server_json(&mut websocket).await;
+        // The runtime may still be finishing local setup before its first tool call.
+        let request = read_exec_server_json(&mut websocket, STARTUP_TIMEOUT).await;
         match request["method"].as_str() {
             Some("process/start") => break request,
             Some("environment/info") => {
@@ -428,7 +434,7 @@ async fn serve_exec_with_pushed_events(
 
     let mut process_read_requests = 0;
     loop {
-        let request = read_exec_server_json(&mut websocket).await;
+        let request = read_exec_server_json(&mut websocket, Duration::from_secs(/*secs*/ 5)).await;
         match request["method"].as_str() {
             Some("process/read") => {
                 process_read_requests += 1;
@@ -687,9 +693,7 @@ timeout = 900
             config.bypass_hook_trust = true;
         }
     });
-    let test = timeout(Duration::from_secs(5), builder.build(&server))
-        .await
-        .context("thread startup should connect to the fake exec-server")??;
+    let test = expect_startup(builder.build(&server)).await;
 
     let turn_permission_profile = if managed_network_configured {
         test.session_configured.permission_profile.clone()