rollout: add ordinals to rollout JSONL lines

codex-rs/app-server/src/request_processors/thread_processor_tests.rs

@@ -1015,6 +1015,7 @@ mod thread_processor_behavior_tests {
 
         let line = RolloutLine {
             timestamp: timestamp.clone(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: session_meta.clone(),
                 git: None,
@@ -1082,6 +1083,7 @@ mod thread_processor_behavior_tests {
 
         let line = RolloutLine {
             timestamp,
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: session_meta,
                 git: None,
@@ -1124,6 +1126,7 @@ mod thread_processor_behavior_tests {
 
         let line = RolloutLine {
             timestamp,
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: session_meta,
                 git: None,

codex-rs/cli/src/doctor/thread_inventory.rs

@@ -819,6 +819,7 @@ mod tests {
             let parsed_thread_id = ThreadId::from_string(thread_id).expect("thread id");
             let rollout_line = RolloutLine {
                 timestamp: timestamp.to_string(),
+                ordinal: None,
                 item: RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine {
                     meta: codex_protocol::protocol::SessionMeta {
                         session_id: parsed_thread_id.into(),

codex-rs/core/src/personality_migration_tests.rs

@@ -69,10 +69,12 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R
     };
     let meta_line = RolloutLine {
         timestamp: TEST_TIMESTAMP.to_string(),
+        ordinal: None,
         item: RolloutItem::SessionMeta(session_meta),
     };
     let user_event = RolloutLine {
         timestamp: TEST_TIMESTAMP.to_string(),
+        ordinal: None,
         item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
             client_id: None,
             message: "hello".to_string(),

codex-rs/core/tests/suite/client.rs

@@ -370,6 +370,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<(
     let rollout = vec![
         RolloutLine {
             timestamp: "2024-01-01T00:00:00.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: SessionMeta {
                     session_id: thread_id.into(),
@@ -387,6 +388,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<(
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:01.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
                 id: Some("fc_existing".to_string()),
                 name: "do_it".to_string(),
@@ -895,6 +897,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
     let rollout = vec![
         RolloutLine {
             timestamp: "2024-01-01T00:00:00.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: SessionMeta {
                     session_id: thread_id.into(),
@@ -912,10 +915,12 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:01.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(legacy_custom_tool_call),
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:02.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
                 id: None,
                 call_id: "legacy-js-call".to_string(),
@@ -926,6 +931,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:03.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::Message {
                 id: None,
                 role: "user".to_string(),
@@ -1031,6 +1037,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
     let rollout = vec![
         RolloutLine {
             timestamp: "2024-01-01T00:00:00.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: SessionMeta {
                     session_id: thread_id.into(),
@@ -1048,6 +1055,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:01.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
                 id: None,
                 name: "view_image".to_string(),
@@ -1059,6 +1067,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:01.500Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::FunctionCallOutput {
                 id: None,
                 call_id: function_call_id.to_string(),
@@ -1073,6 +1082,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:02.000Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::CustomToolCall {
                 id: None,
                 status: Some("completed".to_string()),
@@ -1085,6 +1095,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
         },
         RolloutLine {
             timestamp: "2024-01-01T00:00:02.500Z".to_string(),
+            ordinal: None,
             item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
                 id: None,
                 call_id: custom_call_id.to_string(),

codex-rs/core/tests/suite/personality_migration.rs

@@ -85,10 +85,12 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R
     };
     let meta_line = RolloutLine {
         timestamp: TEST_TIMESTAMP.to_string(),
+        ordinal: None,
         item: RolloutItem::SessionMeta(session_meta),
     };
     let user_event = RolloutLine {
         timestamp: TEST_TIMESTAMP.to_string(),
+        ordinal: None,
         item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
             client_id: None,
             message: "hello".to_string(),
@@ -139,6 +141,7 @@ async fn write_rollout_with_meta_only(dir: &Path, thread_id: ThreadId) -> io::Re
     };
     let meta_line = RolloutLine {
         timestamp: TEST_TIMESTAMP.to_string(),
+        ordinal: None,
         item: RolloutItem::SessionMeta(session_meta),
     };
 

codex-rs/core/tests/suite/sqlite_state.rs

@@ -384,10 +384,12 @@ async fn backfill_scans_existing_rollouts() -> Result<()> {
             let lines = [
                 RolloutLine {
                     timestamp: "2026-01-27T12:00:00Z".to_string(),
+                    ordinal: None,
                     item: RolloutItem::SessionMeta(session_meta_line),
                 },
                 RolloutLine {
                     timestamp: "2026-01-27T12:00:01Z".to_string(),
+                    ordinal: None,
                     item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
                         client_id: None,
                         message: "hello from backfill".to_string(),

codex-rs/memories/write/src/startup_tests.rs

@@ -678,6 +678,7 @@ async fn seed_stage1_candidate(
     let rollout_path = codex_home.join(format!("rollout-{thread_id}.jsonl"));
     let line = RolloutLine {
         timestamp: updated_at.to_rfc3339(),
+        ordinal: None,
         item: RolloutItem::ResponseItem(ResponseItem::Message {
             id: None,
             role: "user".to_string(),

codex-rs/protocol/src/protocol.rs

@@ -3325,6 +3325,8 @@ impl Mul<f64> for TruncationPolicy {
 #[derive(Serialize, Deserialize, Clone, JsonSchema)]
 pub struct RolloutLine {
     pub timestamp: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ordinal: Option<u64>,
     #[serde(flatten)]
     pub item: RolloutItem,
 }

codex-rs/rollout/src/compression_tests.rs

@@ -483,10 +483,12 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) ->
     let lines = [
         RolloutLine {
             timestamp: "2025-01-03T12:00:00Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(session_meta_line),
         },
         RolloutLine {
             timestamp: "2025-01-03T12:00:01Z".to_string(),
+            ordinal: None,
             item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
                 message: message.to_string(),
                 ..Default::default()

codex-rs/rollout/src/lib.rs

@@ -8,6 +8,7 @@ pub(crate) mod compression;
 pub(crate) mod config;
 pub(crate) mod list;
 pub(crate) mod metadata;
+mod ordinal;
 mod persistence_metrics;
 pub(crate) mod policy;
 pub(crate) mod recorder;

codex-rs/rollout/src/metadata_tests.rs

@@ -62,6 +62,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() {
     };
     let rollout_line = RolloutLine {
         timestamp: "2026-01-27T12:34:56Z".to_string(),
+        ordinal: Some(0),
         item: RolloutItem::SessionMeta(session_meta_line.clone()),
     };
     let json = serde_json::to_string(&rollout_line).expect("rollout json");
@@ -93,6 +94,7 @@ async fn extract_metadata_from_rollout_rejects_unknown_history_mode() {
         .join(format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl"));
     let mut rollout_line = serde_json::to_value(RolloutLine {
         timestamp: "2026-01-27T12:34:56Z".to_string(),
+        ordinal: None,
         item: RolloutItem::SessionMeta(SessionMetaLine {
             meta: SessionMeta {
                 session_id: id.into(),
@@ -158,13 +160,15 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() {
     let lines = vec![
         RolloutLine {
             timestamp: "2026-01-27T12:34:56Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: session_meta,
                 git: None,
             }),
         },
         RolloutLine {
             timestamp: "2026-01-27T12:35:00Z".to_string(),
+            ordinal: None,
             item: RolloutItem::SessionMeta(SessionMetaLine {
                 meta: polluted_meta,
                 git: None,
@@ -427,6 +431,7 @@ fn write_rollout_in_sessions_with_cwd(
     };
     let rollout_line = RolloutLine {
         timestamp: event_ts.to_string(),
+        ordinal: None,
         item: RolloutItem::SessionMeta(session_meta_line),
     };
     let json = serde_json::to_string(&rollout_line).expect("serialize rollout");

codex-rs/rollout/src/ordinal.rs

@@ -0,0 +1,109 @@
+use std::fs::File;
+use std::io;
+use std::io::BufRead;
+use std::io::BufReader;
+use std::path::Path;
+
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::ThreadHistoryMode;
+
+use crate::reverse_jsonl_scanner::ReverseJsonlScanner;
+use crate::reverse_jsonl_scanner::ScanOutcome;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum RolloutOrdinalState {
+    Legacy,
+    Paginated { next: Option<u64> },
+}
+
+impl RolloutOrdinalState {
+    pub(crate) fn for_new_rollout(history_mode: ThreadHistoryMode) -> Self {
+        match history_mode {
+            ThreadHistoryMode::Legacy => Self::Legacy,
+            ThreadHistoryMode::Paginated => Self::Paginated { next: Some(0) },
+        }
+    }
+
+    pub(crate) fn current(&self) -> io::Result<Option<u64>> {
+        match self {
+            Self::Legacy => Ok(None),
+            Self::Paginated { next } => {
+                let ordinal = (*next)
+                    .ok_or_else(|| io::Error::other("paginated rollout record ordinal overflow"))?;
+                Ok(Some(ordinal))
+            }
+        }
+    }
+
+    pub(crate) fn advance(&mut self) {
+        if let Self::Paginated { next } = self
+            && let Some(ordinal) = *next
+        {
+            *next = ordinal.checked_add(1);
+        }
+    }
+}
+
+pub(crate) async fn ordinal_state_for_rollout(path: &Path) -> io::Result<RolloutOrdinalState> {
+    let path = path.to_path_buf();
+    tokio::task::spawn_blocking(move || ordinal_state_for_rollout_blocking(path.as_path()))
+        .await
+        .map_err(io::Error::other)?
+}
+
+fn ordinal_state_for_rollout_blocking(path: &Path) -> io::Result<RolloutOrdinalState> {
+    let Some(history_mode) = read_history_mode(path)? else {
+        return Ok(RolloutOrdinalState::Legacy);
+    };
+    if matches!(history_mode, ThreadHistoryMode::Legacy) {
+        return Ok(RolloutOrdinalState::Legacy);
+    }
+
+    let mut scanner = ReverseJsonlScanner::new(File::open(path)?)?;
+    let record = loop {
+        match scanner.scan_next::<RolloutLine>()? {
+            Some(ScanOutcome::Parsed(record)) => break record,
+            Some(ScanOutcome::Rejected(_)) => continue,
+            None => {
+                return Err(io::Error::other(format!(
+                    "rollout at {} contains no valid records",
+                    path.display()
+                )));
+            }
+        }
+    };
+    let ordinal = record.ordinal.ok_or_else(|| {
+        io::Error::other(format!(
+            "final paginated rollout record at {} is missing an ordinal",
+            path.display()
+        ))
+    })?;
+    Ok(RolloutOrdinalState::Paginated {
+        next: ordinal.checked_add(1),
+    })
+}
+
+fn read_history_mode(path: &Path) -> io::Result<Option<ThreadHistoryMode>> {
+    let reader = BufReader::new(File::open(path)?);
+    for line in reader.lines() {
+        let line = line?;
+        if line.trim().is_empty() {
+            continue;
+        }
+        let record: RolloutLine = serde_json::from_str(line.as_str()).map_err(|error| {
+            io::Error::other(format!(
+                "failed to parse first rollout record at {}: {error}",
+                path.display()
+            ))
+        })?;
+        let RolloutItem::SessionMeta(session_meta) = record.item else {
+            return Err(io::Error::other(format!(
+                "rollout at {} does not start with session metadata",
+                path.display()
+            )));
+        };
+        return Ok(Some(session_meta.meta.history_mode));
+    }
+    Ok(None)
+}