diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 37ff0ca4f2..c57d575c7e 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6500,7 +6500,6 @@ impl CodexMessageProcessor { if let Some(review_model) = &config.review_model { config.model = Some(review_model.clone()); } - let NewThread { thread_id, thread: review_thread, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index fcea98468a..566b863536 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -589,10 +589,14 @@ impl ThreadManager { parent_trace: Option, ) -> CodexResult { let history = RolloutRecorder::get_rollout_history(&path).await?; - let snapshot_mid_turn = snapshot_ends_mid_turn(&history); + let snapshot_state = snapshot_turn_state(&history); let history = match snapshot { ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => { - truncate_before_nth_user_message(history, nth_user_message, snapshot_mid_turn) + truncate_before_nth_user_message( + history, + nth_user_message, + snapshot_state.ends_mid_turn, + ) } ForkSnapshot::Interrupted => { let history = match history { @@ -600,8 +604,8 @@ impl ThreadManager { InitialHistory::Forked(history) => InitialHistory::Forked(history), InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history), }; - if snapshot_mid_turn { - append_interrupted_boundary(history) + if snapshot_state.ends_mid_turn { + append_interrupted_boundary(history, snapshot_state.active_turn_id) } else { history } @@ -919,32 +923,47 @@ fn truncate_before_nth_user_message( } } -fn snapshot_ends_mid_turn(history: &InitialHistory) -> bool { +#[derive(Debug, Eq, PartialEq)] +struct SnapshotTurnState { + ends_mid_turn: bool, + active_turn_id: Option, +} + +fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState { let rollout_items = history.get_rollout_items(); let mut builder = ThreadHistoryBuilder::new(); for item in &rollout_items { builder.handle_rollout_item(item); } if builder.has_active_turn() { - return true; + return SnapshotTurnState { + ends_mid_turn: true, + active_turn_id: builder.active_turn_snapshot().map(|turn| turn.id), + }; } let Some(last_user_position) = truncation::user_message_positions_in_rollout(&rollout_items) .last() .copied() else { - return false; + return SnapshotTurnState { + ends_mid_turn: false, + active_turn_id: None, + }; }; // Synthetic fork/resume histories can contain user/assistant response items // without explicit turn lifecycle events. If the persisted snapshot has no // terminating boundary after its last user message, treat it as mid-turn. - !rollout_items[last_user_position + 1..].iter().any(|item| { - matches!( - item, - RolloutItem::EventMsg(EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) - ) - }) + SnapshotTurnState { + ends_mid_turn: !rollout_items[last_user_position + 1..].iter().any(|item| { + matches!( + item, + RolloutItem::EventMsg(EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) + ) + }), + active_turn_id: None, + } } /// Append the same persisted interrupt boundary used by the live interrupt path diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 668fd525e5..f698dfd4d2 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -255,6 +255,103 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() { ); } +#[tokio::test] +async fn interrupted_fork_snapshot_preserves_persisted_turn_id() { + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config(); + config.codex_home = temp_dir.path().join("codex-home"); + config.cwd = config.codex_home.clone(); + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + + let auth_manager = + AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); + let manager = ThreadManager::new( + &config, + auth_manager.clone(), + SessionSource::Exec, + CollaborationModesConfig::default(), + ); + + let source = manager + .resume_thread_with_history( + config.clone(), + InitialHistory::Forked(vec![ + RolloutItem::ResponseItem(user_msg("hello")), + RolloutItem::ResponseItem(assistant_msg("partial")), + ]), + auth_manager, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("create source thread from completed history"); + let source_path = source + .thread + .rollout_path() + .expect("source rollout path should exist"); + let source_history = RolloutRecorder::get_rollout_history(&source_path) + .await + .expect("read source rollout history"); + let source_snapshot_state = snapshot_turn_state(&source_history); + assert!(source_snapshot_state.ends_mid_turn); + let expected_turn_id = source_snapshot_state.active_turn_id.clone(); + assert!(expected_turn_id.is_some()); + + let forked = manager + .fork_thread( + ForkSnapshot::Interrupted, + config, + source_path, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("fork interrupted snapshot"); + let forked_path = forked + .thread + .rollout_path() + .expect("forked rollout path should exist"); + let history = RolloutRecorder::get_rollout_history(&forked_path) + .await + .expect("read forked rollout history"); + assert!(!snapshot_turn_state(&history).ends_mid_turn); + let rollout_items: Vec<_> = history + .get_rollout_items() + .into_iter() + .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) + .collect(); + let interrupted_marker_json = + serde_json::to_value(RolloutItem::ResponseItem(interrupted_turn_history_marker())) + .expect("serialize interrupted marker"); + let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg( + EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: expected_turn_id, + reason: TurnAbortReason::Interrupted, + }), + )) + .expect("serialize interrupted abort event"); + assert_eq!( + rollout_items + .iter() + .filter(|item| { + serde_json::to_value(item).expect("serialize rollout item") + == interrupted_marker_json + }) + .count(), + 1, + ); + assert_eq!( + rollout_items + .iter() + .filter(|item| { + serde_json::to_value(item).expect("serialize rollout item") + == interrupted_abort_json + }) + .count(), + 1, + ); +} + #[tokio::test] async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_source() { let temp_dir = tempdir().expect("tempdir"); @@ -292,7 +389,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ let source_history = RolloutRecorder::get_rollout_history(&source_path) .await .expect("read source rollout history"); - assert!(snapshot_ends_mid_turn(&source_history)); + assert!(snapshot_turn_state(&source_history).ends_mid_turn); manager.remove_thread(&source.thread_id).await; let forked = manager @@ -312,7 +409,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ let history = RolloutRecorder::get_rollout_history(&forked_path) .await .expect("read forked rollout history"); - assert!(!snapshot_ends_mid_turn(&history)); + assert!(!snapshot_turn_state(&history).ends_mid_turn); let forked_rollout_items: Vec<_> = history .get_rollout_items()