tests: harden review and pending-input probes

Co-authored-by: Codex <noreply@openai.com>

codex-rs/app-server/tests/suite/v2/review.rs

@@ -242,35 +242,48 @@ async fn review_start_exec_approval_item_id_matches_command_execution_item() ->
         }
     }
 
-    let read_id = mcp
-        .send_thread_read_request(ThreadReadParams {
-            thread_id: thread_id.clone(),
-            include_turns: true,
-        })
-        .await?;
-    let read_resp: JSONRPCResponse = timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
-    )
-    .await??;
-    let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
-    let turn = thread
-        .turns
-        .iter()
-        .find(|turn| turn.id == turn_id.as_str())
-        .expect("did not observe completed review turn in thread/read");
-    let command_execution_item_ids = turn
-        .items
-        .iter()
-        .filter_map(|item| match item {
-            ThreadItem::CommandExecution { id, .. } => Some(id.clone()),
-            _ => None,
-        })
-        .collect::<Vec<_>>();
-    eprintln!(
-        "review approval probe observed persisted command execution items: {command_execution_item_ids:?}"
+    let expected_command_execution_item_ids = vec![approval_item_id.clone()];
+    let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
+    let mut command_execution_item_ids = None;
+    while command_execution_item_ids.is_none() {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let read_id = mcp
+            .send_thread_read_request(ThreadReadParams {
+                thread_id: thread_id.clone(),
+                include_turns: true,
+            })
+            .await?;
+        let read_resp: JSONRPCResponse = timeout(
+            remaining,
+            mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
+        )
+        .await??;
+        let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
+        let maybe_turn = thread.turns.iter().find(|turn| turn.id == turn_id.as_str());
+        let persisted_item_ids = maybe_turn
+            .map(|turn| {
+                turn.items
+                    .iter()
+                    .filter_map(|item| match item {
+                        ThreadItem::CommandExecution { id, .. } => Some(id.clone()),
+                        _ => None,
+                    })
+                    .collect::<Vec<_>>()
+            })
+            .unwrap_or_default();
+        eprintln!(
+            "review approval probe thread/read persisted command execution items: {persisted_item_ids:?}"
+        );
+        if persisted_item_ids == expected_command_execution_item_ids {
+            command_execution_item_ids = Some(persisted_item_ids);
+        } else {
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    }
+    assert_eq!(
+        command_execution_item_ids.expect("missing persisted command execution items"),
+        expected_command_execution_item_ids
     );
-    assert_eq!(command_execution_item_ids, vec![approval_item_id]);
 
     Ok(())
 }

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

@@ -46,6 +46,12 @@ fn message_input_texts(body: &Value, role: &str) -> Vec<String> {
         .collect()
 }
 
+fn has_user_text(body: &Value, text: &str) -> bool {
+    message_input_texts(body, "user")
+        .iter()
+        .any(|candidate| candidate == text)
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn injected_user_input_triggers_follow_up_request_with_deltas() {
     let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
@@ -172,14 +178,27 @@ async fn injected_user_input_triggers_follow_up_request_with_deltas() {
 
     let first_body: Value = serde_json::from_slice(&requests[0]).expect("parse first request");
     let second_body: Value = serde_json::from_slice(&requests[1]).expect("parse second request");
+    eprintln!("pending input probe request[0] body: {first_body}");
+    eprintln!("pending input probe request[1] body: {second_body}");
 
-    let first_texts = message_input_texts(&first_body, "user");
-    assert!(first_texts.iter().any(|text| text == "first prompt"));
-    assert!(!first_texts.iter().any(|text| text == "second prompt"));
+    let request_bodies = [&first_body, &second_body];
+    let initial_request_matches = request_bodies
+        .iter()
+        .filter(|body| has_user_text(body, "first prompt") && !has_user_text(body, "second prompt"))
+        .count();
+    let follow_up_request_matches = request_bodies
+        .iter()
+        .filter(|body| has_user_text(body, "first prompt") && has_user_text(body, "second prompt"))
+        .count();
 
-    let second_texts = message_input_texts(&second_body, "user");
-    assert!(second_texts.iter().any(|text| text == "first prompt"));
-    assert!(second_texts.iter().any(|text| text == "second prompt"));
+    assert_eq!(
+        initial_request_matches, 1,
+        "expected exactly one initial request with only the first prompt, bodies: {request_bodies:?}"
+    );
+    assert_eq!(
+        follow_up_request_matches, 1,
+        "expected exactly one follow-up request with both prompts, bodies: {request_bodies:?}"
+    );
 
     server.shutdown().await;
 }