test: reduce time dependency on test harness (#5053)

Tightened the CLI integration tests to stop relying on wall-clock
sleeps—new fs watcher helper waits for session files instead of timing
out, and SSE mocks/fixtures make the flows deterministic.

codex-rs/Cargo.lock

@@ -1577,10 +1577,12 @@ dependencies = [
  "anyhow",
  "assert_cmd",
  "codex-core",
+ "notify",
  "regex-lite",
  "serde_json",
  "tempfile",
  "tokio",
+ "walkdir",
  "wiremock",
 ]
 
@@ -2371,6 +2373,15 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fsevent-sys"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "futures"
 version = "0.3.31"
@@ -3057,6 +3068,26 @@ version = "2.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd"
 
+[[package]]
+name = "inotify"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f37dccff2791ab604f9babef0ba14fbe0be30bd368dc541e2b08d07c8aa908f3"
+dependencies = [
+ "bitflags 2.9.1",
+ "inotify-sys",
+ "libc",
+]
+
+[[package]]
+name = "inotify-sys"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "inout"
 version = "0.1.4"
@@ -3257,6 +3288,26 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "kqueue"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eac30106d7dce88daf4a3fcb4879ea939476d5074a9b7ddd0fb97fa4bed5596a"
+dependencies = [
+ "kqueue-sys",
+ "libc",
+]
+
+[[package]]
+name = "kqueue-sys"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed9625ffda8729b85e45cf04090035ac368927b8cebc34898e7c120f52e4838b"
+dependencies = [
+ "bitflags 1.3.2",
+ "libc",
+]
+
 [[package]]
 name = "lalrpop"
 version = "0.19.12"
@@ -3656,6 +3707,30 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61807f77802ff30975e01f4f071c8ba10c022052f98b3294119f3e615d13e5be"
 
+[[package]]
+name = "notify"
+version = "8.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3"
+dependencies = [
+ "bitflags 2.9.1",
+ "fsevent-sys",
+ "inotify",
+ "kqueue",
+ "libc",
+ "log",
+ "mio",
+ "notify-types",
+ "walkdir",
+ "windows-sys 0.60.2",
+]
+
+[[package]]
+name = "notify-types"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e0826a989adedc2a244799e823aece04662b66609d96af8dff7ac6df9a8925d"
+
 [[package]]
 name = "nu-ansi-term"
 version = "0.50.1"

codex-rs/Cargo.toml

@@ -122,6 +122,7 @@ log = "0.4"
 maplit = "1.0.2"
 mime_guess = "2.0.5"
 multimap = "0.10.0"
+notify = "8.2.0"
 nucleo-matcher = "0.3.1"
 openssl-sys = "*"
 opentelemetry = "0.30.0"

codex-rs/core/tests/common/Cargo.toml

@@ -10,8 +10,10 @@ path = "lib.rs"
 anyhow = { workspace = true }
 assert_cmd = { workspace = true }
 codex-core = { workspace = true }
+notify = { workspace = true }
 regex-lite = { workspace = true }
 serde_json = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["time"] }
+walkdir = { workspace = true }
 wiremock = { workspace = true }

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

@@ -164,6 +164,149 @@ pub fn sandbox_network_env_var() -> &'static str {
     codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR
 }
 
+pub mod fs_wait {
+    use anyhow::Result;
+    use anyhow::anyhow;
+    use notify::RecursiveMode;
+    use notify::Watcher;
+    use std::path::Path;
+    use std::path::PathBuf;
+    use std::sync::mpsc;
+    use std::sync::mpsc::RecvTimeoutError;
+    use std::time::Duration;
+    use std::time::Instant;
+    use tokio::task;
+    use walkdir::WalkDir;
+
+    pub async fn wait_for_path_exists(
+        path: impl Into<PathBuf>,
+        timeout: Duration,
+    ) -> Result<PathBuf> {
+        let path = path.into();
+        task::spawn_blocking(move || wait_for_path_exists_blocking(path, timeout)).await?
+    }
+
+    pub async fn wait_for_matching_file(
+        root: impl Into<PathBuf>,
+        timeout: Duration,
+        predicate: impl FnMut(&Path) -> bool + Send + 'static,
+    ) -> Result<PathBuf> {
+        let root = root.into();
+        task::spawn_blocking(move || {
+            let mut predicate = predicate;
+            blocking_find_matching_file(root, timeout, &mut predicate)
+        })
+        .await?
+    }
+
+    fn wait_for_path_exists_blocking(path: PathBuf, timeout: Duration) -> Result<PathBuf> {
+        if path.exists() {
+            return Ok(path);
+        }
+
+        let watch_root = nearest_existing_ancestor(&path);
+        let (tx, rx) = mpsc::channel();
+        let mut watcher = notify::recommended_watcher(move |res| {
+            let _ = tx.send(res);
+        })?;
+        watcher.watch(&watch_root, RecursiveMode::Recursive)?;
+
+        let deadline = Instant::now() + timeout;
+        loop {
+            if path.exists() {
+                return Ok(path.clone());
+            }
+            let now = Instant::now();
+            if now >= deadline {
+                break;
+            }
+            let remaining = deadline.saturating_duration_since(now);
+            match rx.recv_timeout(remaining) {
+                Ok(Ok(_event)) => {
+                    if path.exists() {
+                        return Ok(path.clone());
+                    }
+                }
+                Ok(Err(err)) => return Err(err.into()),
+                Err(RecvTimeoutError::Timeout) => break,
+                Err(RecvTimeoutError::Disconnected) => break,
+            }
+        }
+
+        if path.exists() {
+            Ok(path)
+        } else {
+            Err(anyhow!("timed out waiting for {:?}", path))
+        }
+    }
+
+    fn blocking_find_matching_file(
+        root: PathBuf,
+        timeout: Duration,
+        predicate: &mut impl FnMut(&Path) -> bool,
+    ) -> Result<PathBuf> {
+        let root = wait_for_path_exists_blocking(root, timeout)?;
+
+        if let Some(found) = scan_for_match(&root, predicate) {
+            return Ok(found);
+        }
+
+        let (tx, rx) = mpsc::channel();
+        let mut watcher = notify::recommended_watcher(move |res| {
+            let _ = tx.send(res);
+        })?;
+        watcher.watch(&root, RecursiveMode::Recursive)?;
+
+        let deadline = Instant::now() + timeout;
+
+        while Instant::now() < deadline {
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            match rx.recv_timeout(remaining) {
+                Ok(Ok(_event)) => {
+                    if let Some(found) = scan_for_match(&root, predicate) {
+                        return Ok(found);
+                    }
+                }
+                Ok(Err(err)) => return Err(err.into()),
+                Err(RecvTimeoutError::Timeout) => break,
+                Err(RecvTimeoutError::Disconnected) => break,
+            }
+        }
+
+        if let Some(found) = scan_for_match(&root, predicate) {
+            Ok(found)
+        } else {
+            Err(anyhow!("timed out waiting for matching file in {:?}", root))
+        }
+    }
+
+    fn scan_for_match(root: &Path, predicate: &mut impl FnMut(&Path) -> bool) -> Option<PathBuf> {
+        for entry in WalkDir::new(root).into_iter().filter_map(Result::ok) {
+            let path = entry.path();
+            if !entry.file_type().is_file() {
+                continue;
+            }
+            if predicate(path) {
+                return Some(path.to_path_buf());
+            }
+        }
+        None
+    }
+
+    fn nearest_existing_ancestor(path: &Path) -> PathBuf {
+        let mut current = path;
+        loop {
+            if current.exists() {
+                return current.to_path_buf();
+            }
+            match current.parent() {
+                Some(parent) => current = parent,
+                None => return PathBuf::from("."),
+            }
+        }
+    }
+}
+
 #[macro_export]
 macro_rules! skip_if_sandbox {
     () => {{

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

@@ -1,12 +1,11 @@
 use assert_cmd::Command as AssertCommand;
 use codex_core::RolloutRecorder;
 use codex_core::protocol::GitInfo;
+use core_test_support::fs_wait;
 use core_test_support::skip_if_no_network;
 use std::time::Duration;
-use std::time::Instant;
 use tempfile::TempDir;
 use uuid::Uuid;
-use walkdir::WalkDir;
 use wiremock::Mock;
 use wiremock::MockServer;
 use wiremock::ResponseTemplate;
@@ -211,12 +210,12 @@ async fn responses_api_stream_cli() {
 
 /// End-to-end: create a session (writes rollout), verify the file, then resume and confirm append.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn integration_creates_and_checks_session_file() {
+async fn integration_creates_and_checks_session_file() -> anyhow::Result<()> {
     // Honor sandbox network restrictions for CI parity with the other tests.
-    skip_if_no_network!();
+    skip_if_no_network!(Ok(()));
 
     // 1. Temp home so we read/write isolated session files.
-    let home = TempDir::new().unwrap();
+    let home = TempDir::new()?;
 
     // 2. Unique marker we'll look for in the session log.
     let marker = format!("integration-test-{}", Uuid::new_v4());
@@ -254,63 +253,20 @@ async fn integration_creates_and_checks_session_file() {
 
     // Wait for sessions dir to appear.
     let sessions_dir = home.path().join("sessions");
-    let dir_deadline = Instant::now() + Duration::from_secs(5);
+    fs_wait::wait_for_path_exists(&sessions_dir, Duration::from_secs(5)).await?;
-    while !sessions_dir.exists() && Instant::now() < dir_deadline {
-        std::thread::sleep(Duration::from_millis(50));
-    }
-    assert!(sessions_dir.exists(), "sessions directory never appeared");
 
     // Find the session file that contains `marker`.
-    let deadline = Instant::now() + Duration::from_secs(10);
+    let marker_clone = marker.clone();
-    let mut matching_path: Option<std::path::PathBuf> = None;
+    let path = fs_wait::wait_for_matching_file(&sessions_dir, Duration::from_secs(10), move |p| {
-    while Instant::now() < deadline && matching_path.is_none() {
+        if p.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
-        for entry in WalkDir::new(&sessions_dir) {
+            return false;
-            let entry = match entry {
-                Ok(e) => e,
-                Err(_) => continue,
-            };
-            if !entry.file_type().is_file() {
-                continue;
-            }
-            if !entry.file_name().to_string_lossy().ends_with(".jsonl") {
-                continue;
-            }
-            let path = entry.path();
-            let Ok(content) = std::fs::read_to_string(path) else {
-                continue;
-            };
-            let mut lines = content.lines();
-            if lines.next().is_none() {
-                continue;
-            }
-            for line in lines {
-                if line.trim().is_empty() {
-                    continue;
-                }
-                let item: serde_json::Value = match serde_json::from_str(line) {
-                    Ok(v) => v,
-                    Err(_) => continue,
-                };
-                if item.get("type").and_then(|t| t.as_str()) == Some("response_item")
-                    && let Some(payload) = item.get("payload")
-                    && payload.get("type").and_then(|t| t.as_str()) == Some("message")
-                    && let Some(c) = payload.get("content")
-                    && c.to_string().contains(&marker)
-                {
-                    matching_path = Some(path.to_path_buf());
-                    break;
-                }
-            }
         }
-        if matching_path.is_none() {
+        let Ok(content) = std::fs::read_to_string(p) else {
-            std::thread::sleep(Duration::from_millis(50));
+            return false;
-        }
+        };
-    }
+        content.contains(&marker_clone)
-
+    })
-    let path = match matching_path {
+    .await?;
-        Some(p) => p,
-        None => panic!("No session file containing the marker was found"),
-    };
 
     // Basic sanity checks on location and metadata.
     let rel = match path.strip_prefix(&sessions_dir) {
@@ -418,42 +374,25 @@ async fn integration_creates_and_checks_session_file() {
     assert!(output2.status.success(), "resume codex-cli run failed");
 
     // Find the new session file containing the resumed marker.
-    let deadline = Instant::now() + Duration::from_secs(10);
+    let marker2_clone = marker2.clone();
-    let mut resumed_path: Option<std::path::PathBuf> = None;
+    let resumed_path =
-    while Instant::now() < deadline && resumed_path.is_none() {
+        fs_wait::wait_for_matching_file(&sessions_dir, Duration::from_secs(10), move |p| {
-        for entry in WalkDir::new(&sessions_dir) {
+            if p.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
-            let entry = match entry {
+                return false;
-                Ok(e) => e,
-                Err(_) => continue,
-            };
-            if !entry.file_type().is_file() {
-                continue;
             }
-            if !entry.file_name().to_string_lossy().ends_with(".jsonl") {
+            std::fs::read_to_string(p)
-                continue;
+                .map(|content| content.contains(&marker2_clone))
-            }
+                .unwrap_or(false)
-            let p = entry.path();
+        })
-            let Ok(c) = std::fs::read_to_string(p) else {
+        .await?;
-                continue;
-            };
-            if c.contains(&marker2) {
-                resumed_path = Some(p.to_path_buf());
-                break;
-            }
-        }
-        if resumed_path.is_none() {
-            std::thread::sleep(Duration::from_millis(50));
-        }
-    }
 
-    let resumed_path = resumed_path.expect("No resumed session file found containing the marker2");
     // Resume should write to the existing log file.
     assert_eq!(
         resumed_path, path,
         "resume should create a new session file"
     );
 
-    let resumed_content = std::fs::read_to_string(&resumed_path).unwrap();
+    let resumed_content = std::fs::read_to_string(&resumed_path)?;
     assert!(
         resumed_content.contains(&marker),
         "resumed file missing original marker"
@@ -462,6 +401,7 @@ async fn integration_creates_and_checks_session_file() {
         resumed_content.contains(&marker2),
         "resumed file missing resumed marker"
     );
+    Ok(())
 }
 
 /// Integration test to verify git info is collected and recorded in session files.

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

@@ -5,6 +5,7 @@ use std::os::unix::fs::PermissionsExt;
 use codex_core::protocol::EventMsg;
 use codex_core::protocol::InputItem;
 use codex_core::protocol::Op;
+use core_test_support::fs_wait;
 use core_test_support::responses;
 use core_test_support::skip_if_no_network;
 use core_test_support::test_codex::TestCodex;
@@ -17,8 +18,7 @@ use responses::ev_assistant_message;
 use responses::ev_completed;
 use responses::sse;
 use responses::start_mock_server;
-use tokio::time::Duration;
+use std::time::Duration;
-use tokio::time::sleep;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn summarize_context_three_requests_and_instructions() -> anyhow::Result<()> {
@@ -60,14 +60,7 @@ echo -n "${@: -1}" > $(dirname "${0}")/notify.txt"#,
     wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
 
     // We fork the notify script, so we need to wait for it to write to the file.
-    for _ in 0..100u32 {
+    fs_wait::wait_for_path_exists(&notify_file, Duration::from_secs(5)).await?;
-        if notify_file.exists() {
-            break;
-        }
-        sleep(Duration::from_millis(100)).await;
-    }
-
-    assert!(notify_file.exists());
 
     Ok(())
 }