codex: use ThreadStore history for core review forks (#20577)

- fork loaded parent threads from `ThreadStore` history in core agent
control paths
- migrate guardian review fork history to loaded session history instead
of rereading rollout files

## Verification
- `cargo test -p codex-core spawn_agent_fork`

codex-rs/core/src/agent/control.rs

@@ -354,14 +354,10 @@ impl AgentControl {
         let parent_thread_id = *parent_thread_id;
         let parent_thread = state.get_thread(parent_thread_id).await.ok();
         if let Some(parent_thread) = parent_thread.as_ref() {
-            // `record_conversation_items` only queues rollout writes asynchronously.
-            // Flush/materialize the live parent before snapshotting JSONL for a fork.
-            parent_thread
-                .codex
-                .session
-                .ensure_rollout_materialized()
-                .await;
-            parent_thread.codex.session.flush_rollout().await?;
+            // `record_conversation_items` only queues persistence writes asynchronously.
+            // Flush before snapshotting store history for a fork.
+            parent_thread.ensure_rollout_materialized().await;
+            parent_thread.flush_rollout().await?;
         }
 
         let parent_history = state

codex-rs/core/src/guardian/review_session.rs

@@ -35,7 +35,6 @@ use crate::config::NetworkProxySpec;
 use crate::config::Permissions;
 use crate::context::ContextualUserFragment;
 use crate::context::GuardianFollowupReviewReminder;
-use crate::rollout::recorder::RolloutRecorder;
 use crate::session::Codex;
 use crate::session::session::Session;
 use crate::session::turn_context::TurnContext;
@@ -774,12 +773,11 @@ async fn append_guardian_followup_reminder(review_session: &GuardianReviewSessio
 async fn load_rollout_items_for_fork(
     session: &Session,
 ) -> anyhow::Result<Option<Vec<RolloutItem>>> {
+    session.try_ensure_rollout_materialized().await?;
     session.flush_rollout().await?;
-    let Some(rollout_path) = session.current_rollout_path().await? else {
-        return Ok(None);
-    };
-    let history = RolloutRecorder::get_rollout_history(rollout_path.as_path()).await?;
-    Ok(Some(history.get_rollout_items()))
+    let live_thread = session.live_thread_for_persistence("guardian review fork")?;
+    let history = live_thread.load_history(/*include_archived*/ true).await?;
+    Ok(Some(history.items))
 }
 
 async fn wait_for_guardian_review(

codex-rs/core/src/rollout.rs

@@ -50,6 +50,7 @@ pub(crate) mod list {
     pub use codex_rollout::find_thread_path_by_id_str;
 }
 
+#[cfg(test)]
 pub(crate) mod recorder {
     pub use codex_rollout::RolloutRecorder;
 }