Derive interrupted fork turn ids from persisted history

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

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,

codex-rs/core/src/thread_manager.rs

@@ -589,10 +589,14 @@ impl ThreadManager {
         parent_trace: Option<W3cTraceContext>,
     ) -> CodexResult<NewThread> {
         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<String>,
+}
+
+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

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()