feat(rollout): preserve fork references across replay

Preserve fork-reference replay behavior on the current origin/main base and collapse the branch back to a single commit for easier future restacks.

codex-rs/core/src/codex.rs

@@ -2164,6 +2164,11 @@ impl Session {
         state.clear_connector_selection();
     }
 
+    async fn set_connector_selection(&self, connector_ids: HashSet<String>) {
+        self.clear_connector_selection().await;
+        self.merge_connector_selection(connector_ids).await;
+    }
+
     async fn record_initial_history(&self, conversation_history: InitialHistory) {
         let turn_context = self.new_default_turn().await;
         let is_subagent = {
@@ -2182,8 +2187,19 @@ impl Session {
             }
             InitialHistory::Resumed(resumed_history) => {
                 let rollout_items = resumed_history.history;
+                let hydrated_rollout_items = if rollout_items
+                    .iter()
+                    .any(|item| matches!(item, RolloutItem::ForkReference(_)))
+                {
+                    self.materialize_rollout_items_for_replay(&rollout_items)
+                        .await
+                } else {
+                    rollout_items.clone()
+                };
+                let restored_connector_selection =
+                    Self::extract_connector_selection_from_rollout(&hydrated_rollout_items);
                 let previous_turn_settings = self
-                    .apply_rollout_reconstruction(&turn_context, &rollout_items)
+                    .apply_rollout_reconstruction(&turn_context, &hydrated_rollout_items)
                     .await;
 
                 // If resuming, warn when the last recorded model differs from the current one.
@@ -2208,10 +2224,13 @@ impl Session {
 
                 // Seed usage info from the recorded rollout so UIs can show token counts
                 // immediately on resume/fork.
-                if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
+                if let Some(info) = Self::last_token_info_from_rollout(&hydrated_rollout_items) {
                     let mut state = self.state.lock().await;
                     state.set_token_info(Some(info));
                 }
+                if let Some(selected_connectors) = restored_connector_selection {
+                    self.set_connector_selection(selected_connectors).await;
+                }
 
                 // Defer seeding the session's initial context until the first turn starts so
                 // turn/start overrides can be merged before we write to the rollout.
@@ -2220,18 +2239,40 @@ impl Session {
                 }
             }
             InitialHistory::Forked(rollout_items) => {
-                self.apply_rollout_reconstruction(&turn_context, &rollout_items)
+                let persisted_rollout_items = rollout_items
+                    .iter()
+                    .position(|item| matches!(item, RolloutItem::ForkReference(_)))
+                    .map(|index| rollout_items[index..].to_vec());
+                let hydrated_rollout_items = if rollout_items
+                    .iter()
+                    .any(|item| matches!(item, RolloutItem::ForkReference(_)))
+                {
+                    self.materialize_rollout_items_for_replay(&rollout_items)
+                        .await
+                } else {
+                    rollout_items.clone()
+                };
+                let restored_connector_selection =
+                    Self::extract_connector_selection_from_rollout(&hydrated_rollout_items);
+
+                self.apply_rollout_reconstruction(&turn_context, &hydrated_rollout_items)
                     .await;
 
                 // Seed usage info from the recorded rollout so UIs can show token counts
                 // immediately on resume/fork.
-                if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
+                if let Some(info) = Self::last_token_info_from_rollout(&hydrated_rollout_items) {
                     let mut state = self.state.lock().await;
                     state.set_token_info(Some(info));
                 }
+                if let Some(selected_connectors) = restored_connector_selection {
+                    self.set_connector_selection(selected_connectors).await;
+                }
 
-                // If persisting, persist all rollout items as-is (recorder filters)
-                if !rollout_items.is_empty() {
+                // Persist only the compact fork reference suffix so child rollouts do not
+                // duplicate the full parent history they inherited in memory.
+                if let Some(persisted_rollout_items) = persisted_rollout_items {
+                    self.persist_rollout_items(&persisted_rollout_items).await;
+                } else if !rollout_items.is_empty() {
                     self.persist_rollout_items(&rollout_items).await;
                 }
 
@@ -2272,6 +2313,41 @@ impl Session {
         })
     }
 
+    fn extract_connector_selection_from_rollout(
+        rollout_items: &[RolloutItem],
+    ) -> Option<HashSet<String>> {
+        let mut active_selected_connectors: Option<HashSet<String>> = None;
+
+        for item in rollout_items {
+            let RolloutItem::ResponseItem(response_item) = item else {
+                continue;
+            };
+            let ResponseItem::FunctionCallOutput { output, .. } = response_item else {
+                continue;
+            };
+            let Some(content) = output.body.to_text() else {
+                continue;
+            };
+            let Ok(payload) = serde_json::from_str::<Value>(&content) else {
+                continue;
+            };
+            let Some(selected_connectors) = payload
+                .get("active_selected_tools")
+                .and_then(Value::as_array)
+            else {
+                continue;
+            };
+            let connector_ids = selected_connectors
+                .iter()
+                .filter_map(Value::as_str)
+                .map(ToOwned::to_owned)
+                .collect::<HashSet<_>>();
+            active_selected_connectors = Some(connector_ids);
+        }
+
+        active_selected_connectors
+    }
+
     async fn previous_turn_settings(&self) -> Option<PreviousTurnSettings> {
         let state = self.state.lock().await;
         state.previous_turn_settings()

codex-rs/core/src/codex/rollout_reconstruction.rs

@@ -84,11 +84,34 @@ fn finalize_active_segment<'a>(
 }
 
 impl Session {
+    pub(super) async fn materialize_rollout_items_for_replay(
+        &self,
+        rollout_items: &[RolloutItem],
+    ) -> Vec<RolloutItem> {
+        let codex_home = {
+            self.state
+                .lock()
+                .await
+                .session_configuration
+                .codex_home
+                .clone()
+        };
+        crate::rollout::truncation::materialize_rollout_items_for_replay(
+            codex_home.as_path(),
+            rollout_items,
+        )
+        .await
+    }
+
     pub(super) async fn reconstruct_history_from_rollout(
         &self,
         turn_context: &TurnContext,
         rollout_items: &[RolloutItem],
     ) -> RolloutReconstruction {
+        let rollout_items = self
+            .materialize_rollout_items_for_replay(rollout_items)
+            .await;
+        let rollout_items = rollout_items.as_slice();
         // Replay metadata should already match the shape of the future lazy reverse loader, even
         // while history materialization still uses an eager bridge. Scan newest-to-oldest,
         // stopping once a surviving replacement-history checkpoint and the required resume metadata
@@ -207,7 +230,9 @@ impl Session {
                         active_segment.get_or_insert_with(ActiveReplaySegment::default);
                     active_segment.counts_as_user_turn |= is_user_turn_boundary(response_item);
                 }
-                RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => {}
+                RolloutItem::EventMsg(_)
+                | RolloutItem::ForkReference(_)
+                | RolloutItem::SessionMeta(_) => {}
             }
 
             if base_replacement_history.is_some()
@@ -275,6 +300,7 @@ impl Session {
                     history.drop_last_n_user_turns(rollback.num_turns);
                 }
                 RolloutItem::EventMsg(_)
+                | RolloutItem::ForkReference(_)
                 | RolloutItem::TurnContext(_)
                 | RolloutItem::SessionMeta(_) => {}
             }

codex-rs/core/src/codex/rollout_reconstruction_tests.rs

@@ -7,9 +7,17 @@ use codex_protocol::AgentPath;
 use codex_protocol::ThreadId;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::ForkReferenceItem;
 use codex_protocol::protocol::InterAgentCommunication;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::SessionMeta;
+use codex_protocol::protocol::SessionMetaLine;
+use codex_protocol::protocol::SessionSource;
 use pretty_assertions::assert_eq;
+use std::path::Path;
 use std::path::PathBuf;
+use tempfile::TempDir;
 
 fn user_message(text: &str) -> ResponseItem {
     ResponseItem::Message {
@@ -54,6 +62,52 @@ fn inter_agent_assistant_message(text: &str) -> ResponseItem {
     }
 }
 
+fn write_rollout_items(
+    root: &Path,
+    thread_id: ThreadId,
+    items: &[RolloutItem],
+) -> std::io::Result<PathBuf> {
+    let rollout_dir = root
+        .join(crate::SESSIONS_SUBDIR)
+        .join("2026")
+        .join("03")
+        .join("05");
+    std::fs::create_dir_all(&rollout_dir)?;
+    let rollout_path = rollout_dir.join(format!("rollout-2026-03-05T00-00-00-{thread_id}.jsonl"));
+    let session_meta_line = RolloutLine {
+        timestamp: "2026-03-05T00:00:00Z".to_string(),
+        item: RolloutItem::SessionMeta(SessionMetaLine {
+            meta: SessionMeta {
+                id: thread_id,
+                timestamp: "2026-03-05T00:00:00Z".to_string(),
+                cwd: root.to_path_buf(),
+                originator: "codex".to_string(),
+                cli_version: "test".to_string(),
+                source: SessionSource::Exec,
+                agent_nickname: None,
+                agent_role: None,
+                model_provider: Some("openai".to_string()),
+                base_instructions: None,
+                dynamic_tools: None,
+                memory_mode: None,
+                forked_from_id: None,
+            },
+            git: None,
+        }),
+    };
+    let mut text = format!("{}\n", serde_json::to_string(&session_meta_line).unwrap());
+    for item in items {
+        let line = RolloutLine {
+            timestamp: "2026-03-05T00:00:01Z".to_string(),
+            item: item.clone(),
+        };
+        text.push_str(&serde_json::to_string(&line).unwrap());
+        text.push('\n');
+    }
+    std::fs::write(&rollout_path, text)?;
+    Ok(rollout_path)
+}
+
 #[tokio::test]
 async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previous_turn_settings()
 {
@@ -93,6 +147,129 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previ
     assert!(session.reference_context_item().await.is_none());
 }
 
+#[tokio::test]
+async fn reconstruct_history_materializes_fork_reference_rollout_items() {
+    let (session, turn_context) = make_session_and_context().await;
+    let dir = TempDir::new().expect("create temp dir");
+    let parent_thread_id = ThreadId::new();
+    let parent_rollout_path = write_rollout_items(
+        dir.path(),
+        parent_thread_id,
+        &[
+            RolloutItem::ResponseItem(user_message("first user")),
+            RolloutItem::ResponseItem(assistant_message("first reply")),
+            RolloutItem::ResponseItem(user_message("second user")),
+            RolloutItem::ResponseItem(assistant_message("second reply")),
+        ],
+    )
+    .expect("write parent rollout");
+    let rollout_items = vec![RolloutItem::ForkReference(ForkReferenceItem {
+        rollout_path: parent_rollout_path,
+        nth_user_message: 1,
+    })];
+
+    let reconstructed = session
+        .reconstruct_history_from_rollout(&turn_context, &rollout_items)
+        .await;
+
+    assert_eq!(
+        reconstructed.history,
+        vec![user_message("first user"), assistant_message("first reply")]
+    );
+}
+
+#[tokio::test]
+async fn record_initial_history_forked_materializes_fork_reference_rollout_items() {
+    let (session, turn_context) = make_session_and_context().await;
+    let codex_home = turn_context.config.codex_home.clone();
+    let parent_thread_id = ThreadId::new();
+    let parent_rollout_path = write_rollout_items(
+        codex_home.as_path(),
+        parent_thread_id,
+        &[
+            RolloutItem::ResponseItem(user_message("first user")),
+            RolloutItem::ResponseItem(assistant_message("first reply")),
+            RolloutItem::ResponseItem(user_message("second user")),
+            RolloutItem::ResponseItem(assistant_message("second reply")),
+        ],
+    )
+    .expect("write parent rollout");
+    let rollout_items = vec![RolloutItem::ForkReference(ForkReferenceItem {
+        rollout_path: parent_rollout_path,
+        nth_user_message: 1,
+    })];
+
+    session
+        .record_initial_history(InitialHistory::Forked(rollout_items))
+        .await;
+
+    let reconstruction_turn = session.new_default_turn().await;
+    let mut expected = vec![user_message("first user"), assistant_message("first reply")];
+    expected.extend(
+        session
+            .build_initial_context(reconstruction_turn.as_ref())
+            .await,
+    );
+
+    let history = session.state.lock().await.clone_history();
+    assert_eq!(expected, history.raw_items());
+}
+
+#[tokio::test]
+async fn reconstruct_history_resolves_fork_reference_after_parent_archive_and_unarchive() {
+    let (session, turn_context) = make_session_and_context().await;
+    let codex_home = turn_context.config.codex_home.clone();
+    let parent_thread_id = ThreadId::new();
+    let parent_rollout_path = write_rollout_items(
+        codex_home.as_path(),
+        parent_thread_id,
+        &[
+            RolloutItem::ResponseItem(user_message("first user")),
+            RolloutItem::ResponseItem(assistant_message("first reply")),
+            RolloutItem::ResponseItem(user_message("second user")),
+            RolloutItem::ResponseItem(assistant_message("second reply")),
+        ],
+    )
+    .expect("write parent rollout");
+    let rollout_items = vec![RolloutItem::ForkReference(ForkReferenceItem {
+        rollout_path: parent_rollout_path.clone(),
+        nth_user_message: 1,
+    })];
+    let expected_history = vec![user_message("first user"), assistant_message("first reply")];
+
+    let archived_rollout_dir = codex_home
+        .join(crate::ARCHIVED_SESSIONS_SUBDIR)
+        .join("2026")
+        .join("03")
+        .join("05");
+    std::fs::create_dir_all(&archived_rollout_dir).expect("create archived rollout dir");
+    let archived_rollout_path = archived_rollout_dir.join(
+        parent_rollout_path
+            .file_name()
+            .expect("parent rollout file name"),
+    );
+    std::fs::rename(&parent_rollout_path, &archived_rollout_path).expect("archive parent rollout");
+
+    let reconstructed = session
+        .reconstruct_history_from_rollout(&turn_context, &rollout_items)
+        .await;
+    assert_eq!(reconstructed.history, expected_history);
+
+    let unarchived_rollout_dir = codex_home
+        .join(crate::SESSIONS_SUBDIR)
+        .join("2026")
+        .join("03")
+        .join("05");
+    std::fs::create_dir_all(&unarchived_rollout_dir).expect("create unarchived rollout dir");
+    std::fs::rename(&archived_rollout_path, &parent_rollout_path)
+        .expect("unarchive parent rollout");
+
+    let reconstructed = session
+        .reconstruct_history_from_rollout(&turn_context, &rollout_items)
+        .await;
+    assert_eq!(reconstructed.history, expected_history);
+}
+
 #[tokio::test]
 async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lifecycle_turn_with_missing_turn_context_id()
  {

codex-rs/core/src/lib.rs

@@ -179,6 +179,7 @@ pub use rollout::list::parse_cursor;
 pub use rollout::list::read_head_for_summary;
 pub use rollout::list::read_session_meta_line;
 pub use rollout::policy::EventPersistenceMode;
+pub use rollout::resolve_fork_reference_rollout_path;
 pub use rollout::rollout_date_parts;
 pub use rollout::session_index::find_thread_names_by_ids;
 mod function_tool;

codex-rs/core/src/thread_manager.rs

@@ -34,6 +34,7 @@ use codex_protocol::config_types::CollaborationModeMask;
 #[cfg(test)]
 use codex_protocol::models::ResponseItem;
 use codex_protocol::openai_models::ModelPreset;
+use codex_protocol::protocol::ForkReferenceItem;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::McpServerRefreshConfig;
 use codex_protocol::protocol::Op;
@@ -45,6 +46,7 @@ use codex_protocol::protocol::W3cTraceContext;
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use std::collections::HashMap;
+use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
@@ -607,18 +609,22 @@ impl ThreadManager {
         S: Into<ForkSnapshot>,
     {
         let snapshot = snapshot.into();
-        let history = RolloutRecorder::get_rollout_history(&path).await?;
+        // True forks must discard the source rollout's conversation id so the child gets a
+        // distinct thread id and preserves `forked_from_id` in its SessionMeta. Using the
+        // resume loader here silently turns a fork into an in-place resume.
+        let history = RolloutRecorder::get_fork_history(&path).await?;
         let snapshot_state = snapshot_turn_state(&history);
-        let history = match snapshot {
+        let mut history = match snapshot {
             ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => {
-                truncate_before_nth_user_message(history, nth_user_message, &snapshot_state)
+                truncate_before_nth_user_message(
+                    config.codex_home.as_path(),
+                    history,
+                    nth_user_message,
+                    &snapshot_state,
+                )
+                .await
             }
             ForkSnapshot::Interrupted => {
-                let history = match history {
-                    InitialHistory::New => InitialHistory::New,
-                    InitialHistory::Forked(history) => InitialHistory::Forked(history),
-                    InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
-                };
                 if snapshot_state.ends_mid_turn {
                     append_interrupted_boundary(history, snapshot_state.active_turn_id)
                 } else {
@@ -626,6 +632,33 @@ impl ThreadManager {
                 }
             }
         };
+        if let (
+            ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message),
+            InitialHistory::Forked(items),
+        ) = (snapshot, &mut history)
+        {
+            let source_session_meta = items.iter().find_map(|item| match item {
+                RolloutItem::SessionMeta(meta_line) => Some(meta_line.clone()),
+                RolloutItem::ForkReference(_)
+                | RolloutItem::ResponseItem(_)
+                | RolloutItem::Compacted(_)
+                | RolloutItem::TurnContext(_)
+                | RolloutItem::EventMsg(_) => None,
+            });
+            // Keep the source SessionMeta in-memory so startup can derive `forked_from_id`
+            // for SessionConfigured while still persisting only the compact ForkReference
+            // suffix to the child rollout on disk.
+            *items = source_session_meta
+                .into_iter()
+                .map(RolloutItem::SessionMeta)
+                .chain(std::iter::once(RolloutItem::ForkReference(
+                    ForkReferenceItem {
+                        rollout_path: path.clone(),
+                        nth_user_message,
+                    },
+                )))
+                .collect();
+        }
         Box::pin(self.state.spawn_thread(
             config,
             history,
@@ -918,12 +951,19 @@ impl ThreadManagerState {
 /// when the source thread is currently mid-turn they fall back to cutting
 /// before the active turn's opening boundary so the fork omits the unfinished
 /// suffix entirely.
-fn truncate_before_nth_user_message(
+async fn truncate_before_nth_user_message(
+    codex_home: &Path,
     history: InitialHistory,
     n: usize,
     snapshot_state: &SnapshotTurnState,
 ) -> InitialHistory {
-    let items: Vec<RolloutItem> = history.get_rollout_items();
+    let mut items: Vec<RolloutItem> = history.get_rollout_items();
+    if items
+        .iter()
+        .any(|item| matches!(item, RolloutItem::ForkReference(_)))
+    {
+        items = truncation::materialize_rollout_items_for_replay(codex_home, &items).await;
+    }
     let user_positions = truncation::user_message_positions_in_rollout(&items);
     let rolled = if snapshot_state.ends_mid_turn && n >= user_positions.len() {
         if let Some(cut_idx) = snapshot_state
@@ -1037,3 +1077,131 @@ fn append_interrupted_boundary(history: InitialHistory, turn_id: Option<String>)
 #[cfg(test)]
 #[path = "thread_manager_tests.rs"]
 mod tests;
+#[cfg(test)]
+mod fork_reference_tests {
+    use super::*;
+    use crate::codex::make_session_and_context;
+    use assert_matches::assert_matches;
+    use codex_protocol::models::ContentItem;
+    use codex_protocol::models::ReasoningItemReasoningSummary;
+    use codex_protocol::models::ResponseItem;
+    use pretty_assertions::assert_eq;
+
+    fn user_msg(text: &str) -> ResponseItem {
+        ResponseItem::Message {
+            id: None,
+            role: "user".to_string(),
+            content: vec![ContentItem::OutputText {
+                text: text.to_string(),
+            }],
+            end_turn: None,
+            phase: None,
+        }
+    }
+    fn assistant_msg(text: &str) -> ResponseItem {
+        ResponseItem::Message {
+            id: None,
+            role: "assistant".to_string(),
+            content: vec![ContentItem::OutputText {
+                text: text.to_string(),
+            }],
+            end_turn: None,
+            phase: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn drops_from_last_user_only() {
+        let items = [
+            user_msg("u1"),
+            assistant_msg("a1"),
+            assistant_msg("a2"),
+            user_msg("u2"),
+            assistant_msg("a3"),
+            ResponseItem::Reasoning {
+                id: "r1".to_string(),
+                summary: vec![ReasoningItemReasoningSummary::SummaryText {
+                    text: "s".to_string(),
+                }],
+                content: None,
+                encrypted_content: None,
+            },
+            ResponseItem::FunctionCall {
+                id: None,
+                call_id: "c1".to_string(),
+                name: "tool".to_string(),
+                namespace: None,
+                arguments: "{}".to_string(),
+            },
+            assistant_msg("a4"),
+        ];
+
+        let initial: Vec<RolloutItem> = items
+            .iter()
+            .cloned()
+            .map(RolloutItem::ResponseItem)
+            .collect();
+        let truncated =
+            truncate_before_nth_user_message(Path::new("/tmp"), InitialHistory::Forked(initial), 1)
+                .await;
+        let got_items = truncated.get_rollout_items();
+        let expected_items = vec![
+            RolloutItem::ResponseItem(items[0].clone()),
+            RolloutItem::ResponseItem(items[1].clone()),
+            RolloutItem::ResponseItem(items[2].clone()),
+        ];
+        assert_eq!(
+            serde_json::to_value(&got_items).unwrap(),
+            serde_json::to_value(&expected_items).unwrap()
+        );
+
+        let initial2: Vec<RolloutItem> = items
+            .iter()
+            .cloned()
+            .map(RolloutItem::ResponseItem)
+            .collect();
+        let truncated2 = truncate_before_nth_user_message(
+            Path::new("/tmp"),
+            InitialHistory::Forked(initial2),
+            2,
+        )
+        .await;
+        assert_matches!(truncated2, InitialHistory::New);
+    }
+
+    #[tokio::test]
+    async fn ignores_session_prefix_messages_when_truncating() {
+        let (session, turn_context) = make_session_and_context().await;
+        let mut items = session.build_initial_context(&turn_context).await;
+        items.push(user_msg("feature request"));
+        items.push(assistant_msg("ack"));
+        items.push(user_msg("second question"));
+        items.push(assistant_msg("answer"));
+
+        let rollout_items: Vec<RolloutItem> = items
+            .iter()
+            .cloned()
+            .map(RolloutItem::ResponseItem)
+            .collect();
+
+        let truncated = truncate_before_nth_user_message(
+            Path::new("/tmp"),
+            InitialHistory::Forked(rollout_items),
+            1,
+        )
+        .await;
+        let got_items = truncated.get_rollout_items();
+
+        let expected: Vec<RolloutItem> = vec![
+            RolloutItem::ResponseItem(items[0].clone()),
+            RolloutItem::ResponseItem(items[1].clone()),
+            RolloutItem::ResponseItem(items[2].clone()),
+            RolloutItem::ResponseItem(items[3].clone()),
+        ];
+
+        assert_eq!(
+            serde_json::to_value(&got_items).unwrap(),
+            serde_json::to_value(&expected).unwrap()
+        );
+    }
+}

codex-rs/core/src/thread_manager_tests.rs

@@ -3,8 +3,8 @@ use crate::codex::make_session_and_context;
 use crate::config::test_config;
 use crate::models_manager::collaboration_mode_presets::CollaborationModesConfig;
 use crate::models_manager::manager::RefreshStrategy;
-use crate::rollout::RolloutRecorder;
 use crate::tasks::interrupted_turn_history_marker;
+use assert_matches::assert_matches;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ReasoningItemReasoningSummary;
 use codex_protocol::models::ResponseItem;
@@ -15,6 +15,7 @@ use codex_protocol::protocol::UserMessageEvent;
 use core_test_support::PathExt;
 use core_test_support::responses::mount_models_once;
 use pretty_assertions::assert_eq;
+use std::path::Path;
 use std::time::Duration;
 use tempfile::tempdir;
 use wiremock::MockServer;
@@ -42,8 +43,8 @@ fn assistant_msg(text: &str) -> ResponseItem {
     }
 }
 
-#[test]
-fn truncates_before_requested_user_message() {
+#[tokio::test]
+async fn truncates_before_requested_user_message() {
     let items = [
         user_msg("u1"),
         assistant_msg("a1"),
@@ -74,6 +75,7 @@ fn truncates_before_requested_user_message() {
         .map(RolloutItem::ResponseItem)
         .collect();
     let truncated = truncate_before_nth_user_message(
+        Path::new("/tmp"),
         InitialHistory::Forked(initial),
         /*n*/ 1,
         &SnapshotTurnState {
@@ -81,7 +83,8 @@ fn truncates_before_requested_user_message() {
             active_turn_id: None,
             active_turn_start_index: None,
         },
-    );
+    )
+    .await;
     let got_items = truncated.get_rollout_items();
     let expected_items = vec![
         RolloutItem::ResponseItem(items[0].clone()),
@@ -99,6 +102,7 @@ fn truncates_before_requested_user_message() {
         .map(RolloutItem::ResponseItem)
         .collect();
     let truncated2 = truncate_before_nth_user_message(
+        Path::new("/tmp"),
         InitialHistory::Forked(initial2.clone()),
         /*n*/ 2,
         &SnapshotTurnState {
@@ -106,15 +110,16 @@ fn truncates_before_requested_user_message() {
             active_turn_id: None,
             active_turn_start_index: None,
         },
-    );
+    )
+    .await;
     assert_eq!(
         serde_json::to_value(truncated2.get_rollout_items()).unwrap(),
         serde_json::to_value(initial2).unwrap()
     );
 }
 
-#[test]
-fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
+#[tokio::test]
+async fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
     let items = vec![
         RolloutItem::ResponseItem(user_msg("u1")),
         RolloutItem::ResponseItem(assistant_msg("a1")),
@@ -123,6 +128,7 @@ fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
     ];
 
     let truncated = truncate_before_nth_user_message(
+        Path::new("/tmp"),
         InitialHistory::Forked(items.clone()),
         usize::MAX,
         &SnapshotTurnState {
@@ -130,7 +136,8 @@ fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
             active_turn_id: None,
             active_turn_start_index: None,
         },
-    );
+    )
+    .await;
 
     assert_eq!(
         serde_json::to_value(truncated.get_rollout_items()).unwrap(),
@@ -157,8 +164,8 @@ fn fork_thread_accepts_legacy_usize_snapshot_argument() {
     let _: fn(&ThreadManager, Config, std::path::PathBuf) = assert_legacy_snapshot_callsite;
 }
 
-#[test]
-fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
+#[tokio::test]
+async fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
     let items = vec![
         RolloutItem::ResponseItem(user_msg("u1")),
         RolloutItem::ResponseItem(assistant_msg("a1")),
@@ -182,10 +189,12 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
     );
 
     let truncated = truncate_before_nth_user_message(
+        Path::new("/tmp"),
         InitialHistory::Forked(items.clone()),
         usize::MAX,
         &snapshot_state,
-    );
+    )
+    .await;
 
     assert_eq!(
         serde_json::to_value(truncated.get_rollout_items()).unwrap(),
@@ -209,6 +218,7 @@ async fn ignores_session_prefix_messages_when_truncating() {
         .collect();
 
     let truncated = truncate_before_nth_user_message(
+        Path::new("/tmp"),
         InitialHistory::Forked(rollout_items),
         /*n*/ 1,
         &SnapshotTurnState {
@@ -216,7 +226,8 @@ async fn ignores_session_prefix_messages_when_truncating() {
             active_turn_id: None,
             active_turn_start_index: None,
         },
-    );
+    )
+    .await;
     let got_items = truncated.get_rollout_items();
 
     let expected: Vec<RolloutItem> = vec![

codex-rs/core/src/thread_rollout_truncation.rs

@@ -4,10 +4,14 @@
 //! interpreting them via `event_mapping::parse_turn_item(...)`.
 
 use crate::event_mapping;
+use crate::resolve_fork_reference_rollout_path;
+use crate::rollout::RolloutRecorder;
 use codex_protocol::items::TurnItem;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::RolloutItem;
+use std::path::Path;
+use tracing::warn;
 
 /// Return the indices of user message boundaries in a rollout.
 ///
@@ -68,6 +72,77 @@ pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
     items[..cut_idx].to_vec()
 }
 
+pub(crate) async fn materialize_rollout_items_for_replay(
+    codex_home: &Path,
+    rollout_items: &[RolloutItem],
+) -> Vec<RolloutItem> {
+    const MAX_FORK_REFERENCE_DEPTH: usize = 8;
+
+    let mut materialized = Vec::new();
+    let mut stack: Vec<(Vec<RolloutItem>, usize, usize)> = vec![(rollout_items.to_vec(), 0, 0)];
+
+    while let Some((items, mut idx, depth)) = stack.pop() {
+        while idx < items.len() {
+            match &items[idx] {
+                RolloutItem::ForkReference(reference) => {
+                    if depth >= MAX_FORK_REFERENCE_DEPTH {
+                        warn!(
+                            "skipping fork reference recursion at depth {} for {:?}",
+                            depth, reference.rollout_path
+                        );
+                        idx += 1;
+                        continue;
+                    }
+
+                    let resolved_rollout_path = match resolve_fork_reference_rollout_path(
+                        codex_home,
+                        &reference.rollout_path,
+                    )
+                    .await
+                    {
+                        Ok(path) => path,
+                        Err(err) => {
+                            warn!(
+                                "failed to resolve fork reference rollout {:?}: {err}",
+                                reference.rollout_path
+                            );
+                            idx += 1;
+                            continue;
+                        }
+                    };
+                    let parent_history = match RolloutRecorder::get_rollout_history(
+                        &resolved_rollout_path,
+                    )
+                    .await
+                    {
+                        Ok(history) => history,
+                        Err(err) => {
+                            warn!(
+                                "failed to load fork reference rollout {:?} (resolved from {:?}): {err}",
+                                resolved_rollout_path, reference.rollout_path
+                            );
+                            idx += 1;
+                            continue;
+                        }
+                    };
+                    let parent_items = truncate_rollout_before_nth_user_message_from_start(
+                        &parent_history.get_rollout_items(),
+                        reference.nth_user_message,
+                    );
+
+                    stack.push((items, idx + 1, depth));
+                    stack.push((parent_items, 0, depth + 1));
+                    break;
+                }
+                item => materialized.push(item.clone()),
+            }
+            idx += 1;
+        }
+    }
+
+    materialized
+}
+
 #[cfg(test)]
 #[path = "thread_rollout_truncation_tests.rs"]
 mod tests;