diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index f3e3d6e65c..89a93c2f8a 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -79,6 +79,7 @@ struct ActiveReplaySegment { turn_id: Option, counts_as_user_turn: bool, compaction_count: u64, + pre_user_compaction_count: u64, previous_turn_settings: Option, reference_context_item: TurnReferenceContextItem, base_replacement_history_index: Option, @@ -103,6 +104,8 @@ fn finalize_active_segment( // `EventMsg::UserMessage`. if *pending_rollback_turns > 0 { if active_segment.counts_as_user_turn { + *window_generation = + window_generation.saturating_add(active_segment.pre_user_compaction_count); *pending_rollback_turns -= 1; } return; @@ -168,6 +171,12 @@ impl Session { active_segment.get_or_insert_with(ActiveReplaySegment::default); active_segment.compaction_count = active_segment.compaction_count.saturating_add(1); + // A compaction seen after the user boundary in reverse replay occurred before + // the user input, so it survives rollback of that user turn. + if active_segment.counts_as_user_turn { + active_segment.pre_user_compaction_count = + active_segment.pre_user_compaction_count.saturating_add(1); + } // Looking backward, compaction clears any older baseline unless a newer // `TurnContextItem` in this same segment has already re-established it. if !reconstruction_complete diff --git a/codex-rs/core/tests/suite/window_headers.rs b/codex-rs/core/tests/suite/window_headers.rs index 5592f05f8c..fd6e965bb5 100644 --- a/codex-rs/core/tests/suite/window_headers.rs +++ b/codex-rs/core/tests/suite/window_headers.rs @@ -11,6 +11,7 @@ use codex_protocol::user_input::UserInput; use core_test_support::responses::ResponsesRequest; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_completed_with_tokens; use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; @@ -237,6 +238,84 @@ async fn window_id_rolls_back_across_compaction_and_persists_on_resume() -> Resu Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn window_id_keeps_pre_turn_compaction_after_rolling_back_its_user_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let request_log = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed_with_tokens( + "resp-1", /*total_tokens*/ 50, + )]), + sse(vec![ev_completed_with_tokens( + "resp-2", /*total_tokens*/ 150, + )]), + sse(vec![ + ev_assistant_message("msg-3", "summary"), + ev_completed_with_tokens("resp-3", /*total_tokens*/ 30), + ]), + sse(vec![ + ev_assistant_message("msg-4", "rolled back"), + ev_completed_with_tokens("resp-4", /*total_tokens*/ 30), + ]), + sse(vec![ev_completed("resp-5")]), + ], + ) + .await; + + let mut builder = test_codex().with_config(|config| { + config.model_provider.name = "Non-OpenAI Model provider".to_string(); + config.compact_prompt = Some(SUMMARIZATION_PROMPT.to_string()); + config.model_context_window = Some(1_000); + config.model_auto_compact_token_limit = Some(100); + }); + let initial = builder.build(&server).await?; + let initial_thread = Arc::clone(&initial.codex); + let rollout_path = initial + .session_configured + .rollout_path + .clone() + .expect("rollout path"); + + submit_user_turn(&initial_thread, "first").await?; + submit_user_turn(&initial_thread, "second").await?; + submit_user_turn(&initial_thread, "rolled back after pre-turn compact").await?; + initial_thread + .submit(Op::ThreadRollback { num_turns: 1 }) + .await?; + wait_for_event(&initial_thread, |event| { + matches!(event, EventMsg::ThreadRolledBack(_)) + }) + .await; + shutdown_thread(&initial_thread).await?; + + let resumed = builder + .resume(&server, initial.home.clone(), rollout_path) + .await?; + submit_user_turn(&resumed.codex, "after resume").await?; + shutdown_thread(&resumed.codex).await?; + + let requests = request_log.requests(); + assert_eq!(requests.len(), 5, "expected five model requests"); + + let window_ids = requests.iter().map(window_id_parts).collect::>(); + let thread_id = window_ids[0].0.clone(); + assert_eq!( + window_ids, + vec![ + (thread_id.clone(), 0), + (thread_id.clone(), 0), + (thread_id.clone(), 0), + (thread_id.clone(), 1), + (thread_id, 1), + ] + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn forked_compacted_history_inherits_effective_generation_on_resume() -> Result<()> { skip_if_no_network!(Ok(()));