diff --git a/codex-rs/app-server/tests/suite/v2/review.rs b/codex-rs/app-server/tests/suite/v2/review.rs index 2f7397f0f2..83fa008f53 100644 --- a/codex-rs/app-server/tests/suite/v2/review.rs +++ b/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::(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::>(); - 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::(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::>() + }) + .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(()) } diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index 71947cd5f8..4ec1701440 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -46,6 +46,12 @@ fn message_input_texts(body: &Value, role: &str) -> Vec { .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; }