Load model context from a bounded rollout suffix (#32896)

## Why

Reconstructing the latest model-visible context does not require replaying an
entire paginated rollout when a usable compaction checkpoint and the associated
completed-turn metadata are available.

## What changed

- Add `ThreadStore::load_latest_model_context` and `StoredModelContext` for
  loading replay-ready model context independently of full thread history.
- Reverse-scan plain paginated JSONL rollouts until the newest safe bounded
  suffix is found, while preserving canonical session metadata and chronological
  replay order.
- Fall back to complete history for legacy or compressed rollouts and whenever
  compaction or rollback records make a bounded cutoff unsafe.

## Testing

- Cover checkpoint selection, turn-metadata boundaries, agent messages,
  contextual user fragments, and full-history fallbacks.

GitOrigin-RevId: 3572f4ecc7aa4099a6d9f0d3e72d0ef432cd9497

codex-rs/rollout/src/model_context.rs

@@ -0,0 +1,197 @@
+use codex_protocol::items::TurnItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::InterAgentCommunication;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::SessionMetaLine;
+
+/// Whether a reverse model-context scan needs more rollout items.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ModelContextScanProgress {
+    /// The reader should provide the next older rollout item.
+    Continue,
+    /// The scan has collected a safe bounded suffix.
+    Complete,
+}
+
+/// Accumulates newest-to-oldest rollout items until they are sufficient to reconstruct the latest
+/// model context.
+///
+/// Storage implementations own how they fetch older items. Local JSONL readers and future
+/// reverse-paged cloud readers can both feed their items through this scan to share the cutoff
+/// rules and chronological replay assembly.
+///
+/// The scan stops once it has both:
+///
+/// - `saw_compaction`: a `CompactedItem` with `replacement_history` and `window_number`;
+/// - `saw_completed_turn_context`: a completed user turn with a compatible `TurnContextItem`.
+///
+/// If the scan reaches the beginning before finding a bounded cutoff, it has already collected
+/// the complete replay and so we can return that directly.
+///
+/// `TurnContextItem` does not identify whether it came from a user turn, so one only counts after
+/// the same turn also proves a user-turn boundary: a paginated
+/// `ItemCompleted(UserMessage)` marker, agent message, or inter-agent message. Paginated writers
+/// persist that marker for real user turns; older rollouts without it conservatively scan to the
+/// beginning. A raw `role=user` response item is not sufficient because contextual user fragments
+/// use that role but do not count as turn boundaries during reconstruction. The compaction restores
+/// model-visible items; the turn context restores previous settings (`model`, `comp_hash`, and
+/// `realtime_active`) and the reference baseline.
+///
+/// These paginated shapes disable the bounded cutoff:
+///
+/// - compaction without `replacement_history` or `window_number`;
+/// - rollback markers;
+///
+/// When one appears, the scanner continues to the beginning and returns the complete replay.
+#[derive(Debug, Default)]
+pub struct ModelContextScan {
+    items_newest_first: Vec<RolloutItem>,
+    saw_compaction: bool,
+    saw_completed_turn_context: bool,
+    must_scan_to_start: bool,
+    active_segment: ActiveTurnSegment,
+}
+
+impl ModelContextScan {
+    /// Adds the next newest-to-oldest rollout item and reports whether the reader can stop.
+    pub fn push(&mut self, item: RolloutItem) -> ModelContextScanProgress {
+        let progress = self.observe(&item);
+        self.items_newest_first.push(item);
+        progress
+    }
+
+    /// Returns the collected items in chronological order with canonical head metadata.
+    ///
+    /// Call this after the reader reaches the beginning of its source or after [`Self::push`]
+    /// returns [`ModelContextScanProgress::Complete`].
+    pub fn finish(mut self, session_meta: SessionMetaLine) -> Vec<RolloutItem> {
+        self.items_newest_first.reverse();
+        if self.has_bounded_cutoff() {
+            // A bounded scan stops before reaching the head. Prepend the separately loaded head
+            // SessionMeta, which remains canonical when copied fork history contains later
+            // metadata.
+            self.items_newest_first
+                .insert(0, RolloutItem::SessionMeta(session_meta));
+        }
+        self.items_newest_first
+    }
+
+    fn observe(&mut self, item: &RolloutItem) -> ModelContextScanProgress {
+        if self.must_scan_to_start {
+            return ModelContextScanProgress::Continue;
+        }
+
+        match item {
+            RolloutItem::Compacted(compacted)
+                if compacted.replacement_history.is_none() || compacted.window_number.is_none() =>
+            {
+                self.must_scan_to_start = true;
+            }
+            RolloutItem::Compacted(_) => {
+                self.saw_compaction = true;
+            }
+            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => {
+                // Paginated threads reject rollback. Keep old rollouts correct rather than
+                // duplicating rollback survival semantics in this bounded selector.
+                self.must_scan_to_start = true;
+            }
+            RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => {
+                if self.active_segment.turn_id.is_none() {
+                    self.active_segment.turn_id = Some(event.turn_id.clone());
+                }
+                if turn_ids_are_compatible(
+                    self.active_segment.turn_id.as_deref(),
+                    Some(event.turn_id.as_str()),
+                ) {
+                    self.active_segment.has_user_turn |=
+                        matches!(&event.item, TurnItem::UserMessage(_));
+                }
+            }
+            RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => {
+                self.active_segment
+                    .turn_id
+                    .get_or_insert_with(|| event.turn_id.clone());
+            }
+            RolloutItem::EventMsg(EventMsg::TurnAborted(event)) => {
+                if let Some(turn_id) = &event.turn_id {
+                    self.active_segment
+                        .turn_id
+                        .get_or_insert_with(|| turn_id.clone());
+                }
+            }
+            RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => {
+                if turn_ids_are_compatible(
+                    self.active_segment.turn_id.as_deref(),
+                    Some(event.turn_id.as_str()),
+                ) {
+                    self.finalize_active_segment();
+                }
+            }
+            RolloutItem::TurnContext(context) => {
+                if self.active_segment.turn_id.is_none() {
+                    self.active_segment.turn_id = context.turn_id.clone();
+                }
+                if turn_ids_are_compatible(
+                    self.active_segment.turn_id.as_deref(),
+                    context.turn_id.as_deref(),
+                ) {
+                    self.active_segment.has_turn_context = true;
+                }
+            }
+            RolloutItem::ResponseItem(response_item) => {
+                self.active_segment.has_user_turn |=
+                    response_item_counts_as_user_turn(response_item);
+            }
+            RolloutItem::InterAgentCommunication(_) => {
+                self.active_segment.has_user_turn = true;
+            }
+            RolloutItem::EventMsg(EventMsg::UserMessage(_)) => {
+                self.active_segment.has_user_turn = true;
+            }
+            RolloutItem::EventMsg(_)
+            | RolloutItem::SessionMeta(_)
+            | RolloutItem::InterAgentCommunicationMetadata { .. }
+            | RolloutItem::WorldState(_) => {}
+        }
+
+        if self.has_bounded_cutoff() {
+            ModelContextScanProgress::Complete
+        } else {
+            ModelContextScanProgress::Continue
+        }
+    }
+
+    fn finalize_active_segment(&mut self) {
+        if self.active_segment.has_user_turn && self.active_segment.has_turn_context {
+            self.saw_completed_turn_context = true;
+        }
+        self.active_segment = ActiveTurnSegment::default();
+    }
+
+    fn has_bounded_cutoff(&self) -> bool {
+        !self.must_scan_to_start && self.saw_compaction && self.saw_completed_turn_context
+    }
+}
+
+#[derive(Debug, Default)]
+struct ActiveTurnSegment {
+    turn_id: Option<String>,
+    has_user_turn: bool,
+    has_turn_context: bool,
+}
+
+fn turn_ids_are_compatible(active_turn_id: Option<&str>, item_turn_id: Option<&str>) -> bool {
+    active_turn_id
+        .is_none_or(|turn_id| item_turn_id.is_none_or(|item_turn_id| item_turn_id == turn_id))
+}
+
+fn response_item_counts_as_user_turn(response_item: &ResponseItem) -> bool {
+    match response_item {
+        ResponseItem::AgentMessage { .. } => true,
+        ResponseItem::Message { role, content, .. } => {
+            role == "assistant" && InterAgentCommunication::is_message_content(content)
+        }
+        _ => false,
+    }
+}