From 910cf49269cc07eb97bc551f13b47b61b4d10232 Mon Sep 17 00:00:00 2001 From: Charley Cunningham Date: Tue, 24 Mar 2026 10:14:21 -0700 Subject: [PATCH] [codex] Stabilize second compaction history test (#15605) ## Summary - replace the second-compaction test fixtures with a single ordered `/responses` sequence - assert against the real recorded request order instead of aggregating per-mock captures - realign the second-summary assertion to the first post-compaction user turn where the summary actually appears ## Root cause `compact_resume_after_second_compaction_preserves_history` collected requests from multiple `mount_sse_once_match` recorders. Overlapping matchers could record the same HTTP request more than once, so the test indexed into a duplicated synthetic list rather than the true request stream. That made the summary assertion depend on matcher evaluation order and platform-specific behavior. ## Impact - makes the flaky test deterministic by removing duplicate request capture from the assertion path - keeps the change scoped to the test only ## Validation - `just fmt` - `just argument-comment-lint` - `env -u CODEX_SANDBOX_NETWORK_DISABLED cargo test -p codex-core compact_resume_after_second_compaction_preserves_history -- --nocapture` - repeated the same targeted test 10 times --------- Co-authored-by: Codex --- codex-rs/core/src/codex_tests.rs | 30 ++++++++ .../core/tests/suite/compact_resume_fork.rs | 74 ++++++++++--------- 2 files changed, 69 insertions(+), 35 deletions(-) diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 5d6769e0e..2b805b68c 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -38,6 +38,7 @@ use crate::protocol::NetworkApprovalProtocol; use crate::protocol::RateLimitSnapshot; use crate::protocol::RateLimitWindow; use crate::protocol::ResumedHistory; +use crate::protocol::RolloutItem; use crate::protocol::TokenCountEvent; use crate::protocol::TokenUsage; use crate::protocol::TokenUsageInfo; @@ -1166,6 +1167,35 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result< }) .await?; wait_for_event(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + // The parent rollout writer drains asynchronously after turn completion. + // Wait until the persisted JSONL includes the source user turn before forking from it. + let mut source_history_persisted = false; + for _ in 0..100 { + let history = RolloutRecorder::get_rollout_history(&rollout_path).await; + source_history_persisted = history.ok().is_some_and(|history| { + history.get_rollout_items().into_iter().any(|item| { + matches!( + item, + RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. }) + if role == "user" + && content.iter().any(|content_item| { + matches!( + content_item, + ContentItem::InputText { text } if text == "fork seed" + ) + }) + ) + }) + }); + if source_history_persisted { + break; + } + sleep(StdDuration::from_millis(10)).await; + } + assert!( + source_history_persisted, + "source rollout should contain the completed pre-fork user turn before forking" + ); let mut fork_config = initial.config.clone(); fork_config.permissions.approval_policy = diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index f4f89281d..069cd22c7 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -310,10 +310,10 @@ async fn compact_resume_after_second_compaction_preserves_history() -> Result<() return Ok(()); } - // 1. Arrange mocked SSE responses for the initial flow plus the second compact. + // 1. Arrange mocked SSE responses as a single ordered stream so assertions + // observe the real request sequence instead of per-mock duplicate captures. let server = MockServer::start().await; - let mut request_log = mount_initial_flow(&server).await; - request_log.extend(mount_second_compact_flow(&server).await); + let request_log = mount_second_compact_sequence(&server).await; // 2. Drive the conversation through compact -> resume -> fork -> compact -> resume. let (_home, config, manager, base) = start_test_conversation(&server, None).await; @@ -349,7 +349,12 @@ async fn compact_resume_after_second_compaction_preserves_history() -> Result<() let resumed_again = resume_conversation(&manager, &config, forked_path).await; user_turn(&resumed_again, AFTER_SECOND_RESUME).await; - let mut requests = gather_request_bodies(&request_log); + let mut requests = request_log + .requests() + .into_iter() + .map(|request| request.body_json()) + .collect::>(); + requests.iter_mut().for_each(normalize_line_endings); normalize_compact_prompts(&mut requests); let input_after_compact = json!(requests[requests.len() - 2]["input"]); let input_after_resume = json!(requests[requests.len() - 1]["input"]); @@ -382,20 +387,18 @@ async fn compact_resume_after_second_compaction_preserves_history() -> Result<() ); let seeded_user_prefix = &first_request_user_texts[..first_turn_user_index]; let summary_after_second_compact = - extract_summary_user_text(&requests[requests.len() - 3], SUMMARY_TEXT); + extract_summary_user_text(&requests[requests.len() - 2], SUMMARY_TEXT); let mut expected_after_second_compact_user_texts = vec![ "hello world".to_string(), "AFTER_COMPACT".to_string(), "AFTER_RESUME".to_string(), "AFTER_FORK".to_string(), - summary_after_second_compact, + summary_after_second_compact.clone(), ]; expected_after_second_compact_user_texts.extend_from_slice(seeded_user_prefix); expected_after_second_compact_user_texts.push("AFTER_COMPACT_2".to_string()); - let mut expected_fork_local_user_texts = vec![ - "AFTER_FORK".to_string(), - expected_after_second_compact_user_texts[4].clone(), - ]; + let mut expected_fork_local_user_texts = + vec!["AFTER_FORK".to_string(), summary_after_second_compact]; expected_fork_local_user_texts.extend_from_slice(seeded_user_prefix); expected_fork_local_user_texts.push("AFTER_COMPACT_2".to_string()); let final_user_texts = json_message_input_texts(&requests[requests.len() - 1], "user"); @@ -403,11 +406,16 @@ async fn compact_resume_after_second_compaction_preserves_history() -> Result<() .split_last() .unwrap_or_else(|| panic!("after-second-resume request missing user messages")); assert_eq!(final_last, AFTER_SECOND_RESUME); - let matched_prefix_len = if final_prefix.starts_with(&expected_after_second_compact_user_texts) + let matched_prefix_len = if let Some(start) = final_prefix + .windows(expected_after_second_compact_user_texts.len()) + .position(|window| window == expected_after_second_compact_user_texts) { - expected_after_second_compact_user_texts.len() - } else if final_prefix.starts_with(&expected_fork_local_user_texts) { - expected_fork_local_user_texts.len() + start + expected_after_second_compact_user_texts.len() + } else if let Some(start) = final_prefix + .windows(expected_fork_local_user_texts.len()) + .position(|window| window == expected_fork_local_user_texts) + { + start + expected_fork_local_user_texts.len() } else { panic!("after-second-resume user texts should preserve post-compact user history prefix"); }; @@ -749,33 +757,29 @@ async fn mount_initial_flow(server: &MockServer) -> Vec { vec![first, compact, after_compact, after_resume, after_fork] } -async fn mount_second_compact_flow(server: &MockServer) -> Vec { +async fn mount_second_compact_sequence(server: &MockServer) -> ResponseMock { + let sse1 = sse(vec![ + ev_assistant_message("m1", FIRST_REPLY), + ev_completed("r1"), + ]); + let sse2 = sse(vec![ + ev_assistant_message("m2", SUMMARY_TEXT), + ev_completed("r2"), + ]); + let sse3 = sse(vec![ + ev_assistant_message("m3", "AFTER_COMPACT_REPLY"), + ev_completed("r3"), + ]); + let sse4 = sse(vec![ev_completed("r4")]); + let sse5 = sse(vec![ev_completed("r5")]); let sse6 = sse(vec![ ev_assistant_message("m4", SUMMARY_TEXT), ev_completed("r6"), ]); let sse7 = sse(vec![ev_completed("r7")]); + let sse8 = sse(vec![ev_completed("r8")]); - // Keep this matcher broad enough to survive prompt-shape differences across - // platforms/config (history may include either marker text or compact prompt - // fragments), but explicitly exclude the final resume turn so these two - // one-shot mocks cannot race for the same request. - let match_second_compact = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - (body.contains("AFTER_FORK") - || body_contains_text(body, SUMMARIZATION_PROMPT) - || body.contains(&json_fragment(FIRST_REPLY))) - && !body.contains(&format!("\"text\":\"{AFTER_SECOND_RESUME}\"")) - }; - let second_compact = mount_sse_once_match(server, match_second_compact, sse6).await; - - let match_after_second_resume = |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body.contains(&format!("\"text\":\"{AFTER_SECOND_RESUME}\"")) - }; - let after_second_resume = mount_sse_once_match(server, match_after_second_resume, sse7).await; - - vec![second_compact, after_second_resume] + mount_sse_sequence(server, vec![sse1, sse2, sse3, sse4, sse5, sse6, sse7, sse8]).await } async fn start_test_conversation(