diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs
index 7dec79b486..72424cc19b 100644
--- a/codex-rs/core/src/codex.rs
+++ b/codex-rs/core/src/codex.rs
@@ -1948,44 +1948,26 @@ impl Session {
                 }
             }
             InitialHistory::Resumed(resumed_history) => {
+                let derive_from_rollout_source = |rollout_source: &InMemoryRolloutSource| {
+                    (
+                        self.reconstruct_history_from_rollout(&turn_context, rollout_source),
+                        Self::extract_mcp_tool_selection_from_rollout_source(rollout_source),
+                        Self::last_token_info_from_rollout_source(rollout_source),
+                    )
+                };
                 let (reconstructed_rollout, restored_tool_selection, token_info) =
                     if resumed_history.history.is_empty() {
                         let rollout = self.services.rollout.lock().await;
-                        match rollout.as_ref() {
-                            Some(rollout) => {
-                                let rollout_source = rollout.source.lock().await;
-                                (
-                                    self.reconstruct_history_from_rollout(
-                                        &turn_context,
-                                        &rollout_source,
-                                    ),
-                                    Self::extract_mcp_tool_selection_from_rollout_source(
-                                        &rollout_source,
-                                    ),
-                                    Self::last_token_info_from_rollout_source(&rollout_source),
-                                )
-                            }
-                            None => {
-                                let rollout_source = InMemoryRolloutSource::new(Vec::new());
-                                (
-                                    self.reconstruct_history_from_rollout(
-                                        &turn_context,
-                                        &rollout_source,
-                                    ),
-                                    Self::extract_mcp_tool_selection_from_rollout_source(
-                                        &rollout_source,
-                                    ),
-                                    Self::last_token_info_from_rollout_source(&rollout_source),
-                                )
-                            }
+                        if let Some(rollout) = rollout.as_ref() {
+                            let rollout_source = rollout.source.lock().await;
+                            derive_from_rollout_source(&rollout_source)
+                        } else {
+                            derive_from_rollout_source(&InMemoryRolloutSource::new(Vec::new()))
                         }
                     } else {
-                        let rollout_source = InMemoryRolloutSource::new(resumed_history.history);
-                        (
-                            self.reconstruct_history_from_rollout(&turn_context, &rollout_source),
-                            Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source),
-                            Self::last_token_info_from_rollout_source(&rollout_source),
-                        )
+                        derive_from_rollout_source(&InMemoryRolloutSource::new(
+                            resumed_history.history,
+                        ))
                     };
                 let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone();
                 self.set_previous_turn_settings(previous_turn_settings.clone())
diff --git a/codex-rs/core/src/rollout/metadata.rs b/codex-rs/core/src/rollout/metadata.rs
index 48268ebf6c..3a603f0da7 100644
--- a/codex-rs/core/src/rollout/metadata.rs
+++ b/codex-rs/core/src/rollout/metadata.rs
@@ -100,20 +100,18 @@ pub(crate) async fn extract_metadata_from_rollout(
     let parse_errors = loaded_rollout.parse_errors;
     let source = loaded_rollout.source;
     let rollout_start = source.inclusive_start_of_rollout_index();
-    let has_any_parsed_items = source.iter_forward_from(rollout_start).next().is_some();
+    let rollout_items = source
+        .iter_forward_from(rollout_start)
+        .map(|(_, item)| item)
+        .collect::<Vec<_>>();
+    let has_any_parsed_items = !rollout_items.is_empty();
     if parse_errors > 0 && !has_any_parsed_items {
         anyhow::bail!(
             "rollout contains parse errors and no readable items: {}",
             rollout_path.display()
         );
     }
-    let builder = builder_from_items(
-        source
-            .iter_forward_from(rollout_start)
-            .map(|(_, item)| item),
-        rollout_path,
-    )
-    .ok_or_else(|| {
+    let builder = builder_from_items(rollout_items, rollout_path).ok_or_else(|| {
         anyhow::anyhow!(
             "rollout missing metadata builder: {}",
             rollout_path.display()
diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs
index cb81ca4b3e..be7c4bc1cb 100644
--- a/codex-rs/core/src/rollout/recorder.rs
+++ b/codex-rs/core/src/rollout/recorder.rs
@@ -700,6 +700,8 @@ impl RolloutStore {
     /// Ensure all queued rollout writes have been persisted to disk.
     pub async fn persist(&self) -> std::io::Result<()> {
         let (tx, rx) = oneshot::channel();
+        // Control commands share the same ordering barrier as `record_items` so the writer sees
+        // them after any earlier in-memory source updates have been queued.
         let queue_order_guard = self.queue_order.lock().await;
         self.tx
             .send(RolloutCmd::Persist { ack: tx })
@@ -713,6 +715,8 @@ impl RolloutStore {
     /// Flush all queued writes and wait until they are committed by the writer task.
     pub async fn flush(&self) -> std::io::Result<()> {
         let (tx, rx) = oneshot::channel();
+        // `Flush` also participates in the ordering barrier so the writer drains any earlier
+        // `record_items` updates before acknowledging the flush.
         let queue_order_guard = self.queue_order.lock().await;
         self.tx
             .send(RolloutCmd::Flush { ack: tx })
@@ -810,6 +814,8 @@ impl RolloutStore {
     /// Shut down the background writer after draining all previously queued work.
     pub async fn shutdown(&self) -> std::io::Result<()> {
         let (tx_done, rx_done) = oneshot::channel();
+        // `Shutdown` uses the same barrier so the writer observes it after any earlier in-memory
+        // source updates and queued control commands.
         let queue_order_guard = self.queue_order.lock().await;
         match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await {
             Ok(_) => {}