Make thread history projection resilient to malformed rollouts (#36188)

## Why

A failed rollout append can leave a rejected line before a valid same-ordinal
retry. Advancing the byte checkpoint past that line without also advancing the
ordinal checkpoint can prevent subsequent history from being projected.

## What changed

- Defer rejected and unknown rollout lines until a later valid ordinal shows
  whether they were replaced by a retry or consumed an ordinal.
- Advance skipped ordinal ranges together with the durable byte checkpoint,
  while rejecting gaps that cannot be explained by rejected lines.
- Prefer item event timestamps and parse the rollout timestamp only when an
  item needs it as a fallback.

## Testing

Added coverage for same-ordinal retries, inferred ordinal gaps, unknown and
malformed records, invalid fallback timestamps, and unexplained gaps.

GitOrigin-RevId: 374beda316c605d562bd9526abed19916598d69f

codex-rs/thread-store/src/local/thread_history.rs

@@ -21,16 +21,27 @@ pub(super) use turn_lookup::find_visible_turn;
 
 /// A valid complete rollout line with its absolute byte span in durable JSONL.
 ///
-/// `start_byte_offset..end_byte_offset` includes the terminating newline. Blank and rejected
+/// `start_byte_offset..end_byte_offset` includes the terminating newline.
-/// lines do not produce a value here, but still advance later spans.
 pub(super) struct ProjectedRolloutLine {
     pub ordinal: u64,
     pub start_byte_offset: u64,
     pub end_byte_offset: u64,
-    pub created_at_ms: i64,
+    pub fallback_created_at_ms: Option<i64>,
     pub changes: ThreadHistoryChangeSet,
 }
 
+/// One ordered update to apply while advancing a rollout projection checkpoint.
+///
+/// Skipped ordinal ranges keep the byte and ordinal checkpoints describing the same durable
+/// prefix even when a complete rollout line cannot be projected.
+pub(super) enum RolloutProjectionStep {
+    Line(ProjectedRolloutLine),
+    SkippedOrdinalRange {
+        start_ordinal: u64,
+        end_ordinal_exclusive: u64,
+    },
+}
+
 pub(super) struct RolloutProjectionState {
     pub next_byte_offset: u64,
     pub next_ordinal: u64,
@@ -91,7 +102,7 @@ pub(super) async fn apply_projection(
     start_offset: u64,
     next_offset: u64,
     initial_ordinal: u64,
-    projections: Vec<ProjectedRolloutLine>,
+    projections: Vec<RolloutProjectionStep>,
 ) -> ThreadStoreResult<()> {
     let pool = store.thread_history_db().await?;
     // Write the projected rows and advance the JSONL offset and ordinal in one transaction. If
@@ -123,6 +134,8 @@ WHERE thread_id = ?
     }
 
     for projection in projections {
+        match projection {
+            RolloutProjectionStep::Line(projection) => {
                 let ordinal = sqlite_integer(projection.ordinal, "rollout ordinal")?;
                 if ordinal != next_ordinal {
                     return Err(ThreadStoreError::Internal {
@@ -137,16 +150,42 @@ WHERE thread_id = ?
                     ordinal,
                     sqlite_integer(projection.start_byte_offset, "rollout byte offset")?,
                     sqlite_integer(projection.end_byte_offset, "rollout byte offset")?,
-            projection.created_at_ms,
+                    projection.fallback_created_at_ms,
                     projection.changes,
                 )
                 .await?;
-        next_ordinal = next_ordinal
+                next_ordinal =
+                    next_ordinal
                         .checked_add(1)
                         .ok_or_else(|| ThreadStoreError::Internal {
                             message: "rollout ordinal exceeds SQLite integer range".to_string(),
                         })?;
             }
+            RolloutProjectionStep::SkippedOrdinalRange {
+                start_ordinal,
+                end_ordinal_exclusive,
+            } => {
+                let start_ordinal = sqlite_integer(start_ordinal, "rollout ordinal")?;
+                if start_ordinal != next_ordinal {
+                    return Err(ThreadStoreError::Internal {
+                        message: format!(
+                            "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {start_ordinal}"
+                        ),
+                    });
+                }
+                let end_ordinal_exclusive =
+                    sqlite_integer(end_ordinal_exclusive, "rollout ordinal")?;
+                if end_ordinal_exclusive <= start_ordinal {
+                    return Err(ThreadStoreError::Internal {
+                        message: format!(
+                            "thread history projection for {thread_id} has an empty skipped ordinal range"
+                        ),
+                    });
+                }
+                next_ordinal = end_ordinal_exclusive;
+            }
+        }
+    }
 
     sqlx::query(
         r#"
@@ -214,7 +253,7 @@ async fn apply_change_set(
     rollout_ordinal: i64,
     rollout_byte_offset: i64,
     rollout_end_byte_offset: i64,
-    created_at_ms: i64,
+    fallback_created_at_ms: Option<i64>,
     changes: ThreadHistoryChangeSet,
 ) -> ThreadStoreResult<()> {
     for turn in changes.changed_turns {
@@ -342,6 +381,14 @@ WHERE thread_id = ?
     }
 
     for item in changes.changed_items {
+        let created_at_ms =
+            item.started_at_ms
+                .or(fallback_created_at_ms)
+                .ok_or_else(|| ThreadStoreError::Internal {
+                    message: format!(
+                        "thread history projection for {thread_id} is missing an item creation timestamp"
+                    ),
+                })?;
         let item_id = item.item.id().to_string();
         let item_json = serde_json::to_string(&item.item).map_err(thread_history_error)?;
         // Completed items are immutable: local producers emit ItemCompleted exactly once per

codex-rs/thread-store/src/local/thread_history_materialization.rs

@@ -12,15 +12,10 @@ use tracing::warn;
 
 use super::LocalThreadStore;
 use super::thread_history::ProjectedRolloutLine;
+use super::thread_history::RolloutProjectionStep;
 use crate::ThreadStoreError;
 use crate::ThreadStoreResult;
 
-struct CompleteRolloutLine {
-    line: RolloutLine,
-    start_byte_offset: u64,
-    end_byte_offset: u64,
-}
-
 pub(super) async fn materialize_to_sqlite(
     store: &LocalThreadStore,
     thread_id: ThreadId,
@@ -29,12 +24,15 @@ pub(super) async fn materialize_to_sqlite(
     if store.state_db.is_none() {
         return Ok(());
     }
-    let start_offset = super::thread_history::projection_state(store, thread_id)
+    let projection_state = super::thread_history::projection_state(store, thread_id).await?;
-        .await?
+    let start_offset = projection_state
+        .as_ref()
         .map_or(0, |state| state.next_byte_offset);
-    let (lines, next_offset) = read_complete_rollout_lines(rollout_path, start_offset).await?;
+    if projection_state.is_none()
-    // Empty valid records can still consume bytes through blank or rejected complete lines.
+        && !tokio::fs::try_exists(rollout_path)
-    if lines.is_empty() && start_offset == next_offset {
+            .await
+            .map_err(thread_store_io_error)?
+    {
         return Ok(());
     }
     let session_meta = codex_rollout::read_session_meta_line(rollout_path)
@@ -45,31 +43,21 @@ pub(super) async fn materialize_to_sqlite(
         .history_base
         .map_or(0, |base| base.end_ordinal_exclusive);
     let subagent_history_start_ordinal = session_meta.subagent_history_start_ordinal;
-
+    let expected_ordinal = projection_state
-    let projections = lines
+        .as_ref()
-        .iter()
+        .map_or(initial_ordinal, |state| state.next_ordinal);
-        .map(|record| {
+    let (projections, next_offset) = read_projection_steps(
-            let line = &record.line;
+        rollout_path,
-            let ordinal = line.ordinal.ok_or_else(|| ThreadStoreError::Internal {
+        start_offset,
-                message: format!("paginated rollout line for {thread_id} is missing an ordinal"),
+        expected_ordinal,
-            })?;
+        thread_id,
-            let created_at_ms = DateTime::parse_from_rfc3339(line.timestamp.as_str())
+        subagent_history_start_ordinal,
-                .map(|timestamp| timestamp.timestamp_millis())
+    )
-                .map_err(thread_history_error)?;
+    .await?;
-            let changes = if subagent_history_start_ordinal.is_some_and(|start| ordinal < start) {
+    // Empty valid records can still consume bytes through blank complete lines.
-                ThreadHistoryChangeSet::default()
+    if projections.is_empty() && start_offset == next_offset {
-            } else {
+        return Ok(());
-                project_rollout_line(line)
+    }
-            };
-            Ok(ProjectedRolloutLine {
-                ordinal,
-                start_byte_offset: record.start_byte_offset,
-                end_byte_offset: record.end_byte_offset,
-                created_at_ms,
-                changes,
-            })
-        })
-        .collect::<ThreadStoreResult<Vec<_>>>()?;
     super::thread_history::apply_projection(
         store,
         thread_id,
@@ -81,11 +69,14 @@ pub(super) async fn materialize_to_sqlite(
     .await
 }
 
-async fn read_complete_rollout_lines(
+async fn read_projection_steps(
     rollout_path: &Path,
     start_offset: u64,
-) -> ThreadStoreResult<(Vec<CompleteRolloutLine>, u64)> {
+    expected_ordinal: u64,
-    let next_offset = match tokio::fs::metadata(rollout_path).await {
+    thread_id: ThreadId,
+    subagent_history_start_ordinal: Option<u64>,
+) -> ThreadStoreResult<(Vec<RolloutProjectionStep>, u64)> {
+    let file_end_offset = match tokio::fs::metadata(rollout_path).await {
         Ok(metadata) => metadata.len(),
         Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => {
             return Ok((Vec::new(), 0));
@@ -93,7 +84,7 @@ async fn read_complete_rollout_lines(
         Err(err) => return Err(thread_store_io_error(err)),
     };
     let byte_count =
-        next_offset
+        file_end_offset
             .checked_sub(start_offset)
             .ok_or_else(|| ThreadStoreError::Internal {
                 message: "durable rollout shrank before projection".to_string(),
@@ -117,19 +108,13 @@ async fn read_complete_rollout_lines(
         .iter()
         .rposition(|byte| *byte == b'\n')
         .map_or(0, |index| index + 1);
-    let next_offset = start_offset
+    let mut projections = Vec::new();
-        .checked_add(u64::try_from(complete_byte_count).map_err(|_| {
+    let mut next_ordinal = expected_ordinal;
-            ThreadStoreError::Internal {
+    let mut next_offset = start_offset;
-                message: "durable rollout append exceeds addressable memory".to_string(),
+    let mut pending_rejected_line_count = 0;
-            }
-        })?)
-        .ok_or_else(|| ThreadStoreError::Internal {
-            message: "durable rollout byte offset overflow".to_string(),
-        })?;
-    let mut lines = Vec::new();
     let mut line_start_offset = start_offset;
-    // Preserve each complete physical line's trailing newline so byte offsets advance through
+    // Keep rejected lines pending until a later valid ordinal proves whether they consumed history.
-    // every durable byte, including blank or rejected lines that do not project a row.
+    // This lets a same-ordinal retry replace a failed write without advancing only one checkpoint.
     for line_bytes in bytes[..complete_byte_count].split_inclusive(|byte| *byte == b'\n') {
         let line_end_offset = line_start_offset
             .checked_add(u64::try_from(line_bytes.len()).map_err(|_| {
@@ -140,35 +125,153 @@ async fn read_complete_rollout_lines(
             .ok_or_else(|| ThreadStoreError::Internal {
                 message: "durable rollout byte offset overflow".to_string(),
             })?;
-        // Blank physical lines consume bytes but are not rollout records.
+        if line_bytes.iter().all(u8::is_ascii_whitespace) {
-        if !line_bytes.iter().all(u8::is_ascii_whitespace) {
+            if pending_rejected_line_count == 0 {
-            match serde_json::from_slice::<serde_json::Value>(line_bytes)
+                next_offset = line_end_offset;
-                .and_then(serde_json::from_value::<RolloutLine>)
-            {
-                Ok(line) => lines.push(CompleteRolloutLine {
-                    line,
-                    start_byte_offset: line_start_offset,
-                    end_byte_offset: line_end_offset,
-                }),
-                Err(err) => {
-                    // A failed append can leave a partial record behind. The rollout writer
-                    // repairs its newline before retrying, so skip rejected lines just like the
-                    // canonical rollout loader and keep projecting the valid retry that follows.
-                    warn!(
-                        "skipping rejected rollout line while projecting {rollout_path:?}: {err}"
-                    );
-                }
-            }
             }
             line_start_offset = line_end_offset;
+            continue;
         }
-    Ok((lines, next_offset))
+        let value = match serde_json::from_slice::<serde_json::Value>(line_bytes) {
+            Ok(value) => value,
+            Err(err) => {
+                warn!(
+                    thread_id = %thread_id,
+                    rollout_path = %rollout_path.display(),
+                    line_start_byte_offset = line_start_offset,
+                    line_end_byte_offset = line_end_offset,
+                    expected_ordinal = next_ordinal,
+                    error = %err,
+                    "deferring rejected rollout line until a later ordinal resolves it"
+                );
+                pending_rejected_line_count += 1;
+                line_start_offset = line_end_offset;
+                continue;
             }
-
+        };
-fn thread_history_error(err: impl std::fmt::Display) -> ThreadStoreError {
+        let value_ordinal = value.get("ordinal").and_then(serde_json::Value::as_u64);
-    ThreadStoreError::Internal {
+        let line = match serde_json::from_value::<RolloutLine>(value) {
-        message: format!("failed to project thread history: {err}"),
+            Ok(line) => Some(line),
+            Err(err) => {
+                warn!(
+                    thread_id = %thread_id,
+                    rollout_path = %rollout_path.display(),
+                    line_start_byte_offset = line_start_offset,
+                    line_end_byte_offset = line_end_offset,
+                    expected_ordinal = next_ordinal,
+                    line_ordinal = ?value_ordinal,
+                    error = %err,
+                    "deferring unknown rollout line until a later ordinal resolves it"
+                );
+                None
             }
+        };
+        let ordinal = match line
+            .as_ref()
+            .and_then(|line| line.ordinal)
+            .or(value_ordinal)
+        {
+            Some(ordinal) => ordinal,
+            None if line.is_none() => {
+                pending_rejected_line_count += 1;
+                line_start_offset = line_end_offset;
+                continue;
+            }
+            None => {
+                return Err(ThreadStoreError::Internal {
+                    message: format!(
+                        "paginated rollout line for {thread_id} is missing an ordinal"
+                    ),
+                });
+            }
+        };
+        if ordinal < next_ordinal {
+            return Err(ThreadStoreError::Internal {
+                message: format!(
+                    "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}"
+                ),
+            });
+        }
+        let Some(line) = line else {
+            pending_rejected_line_count += 1;
+            line_start_offset = line_end_offset;
+            continue;
+        };
+        let skipped_ordinal_count = ordinal - next_ordinal;
+        if skipped_ordinal_count > pending_rejected_line_count {
+            return Err(ThreadStoreError::Internal {
+                message: format!(
+                    "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}; {pending_rejected_line_count} rejected rollout lines cannot cover that gap"
+                ),
+            });
+        }
+        let changes = if subagent_history_start_ordinal.is_some_and(|start| ordinal < start) {
+            ThreadHistoryChangeSet::default()
+        } else {
+            project_rollout_line(&line)
+        };
+        let fallback_created_at_ms = if changes
+            .changed_items
+            .iter()
+            .any(|item| item.started_at_ms.is_none())
+        {
+            match DateTime::parse_from_rfc3339(line.timestamp.as_str()) {
+                Ok(timestamp) => Some(timestamp.timestamp_millis()),
+                Err(err) => {
+                    warn!(
+                        thread_id = %thread_id,
+                        rollout_path = %rollout_path.display(),
+                        line_start_byte_offset = line_start_offset,
+                        line_end_byte_offset = line_end_offset,
+                        expected_ordinal = next_ordinal,
+                        line_ordinal = ordinal,
+                        error = %err,
+                        "deferring rollout line with invalid timestamp until a later ordinal resolves it"
+                    );
+                    pending_rejected_line_count += 1;
+                    line_start_offset = line_end_offset;
+                    continue;
+                }
+            }
+        } else {
+            None
+        };
+        if skipped_ordinal_count > 0 {
+            warn!(
+                thread_id = %thread_id,
+                rollout_path = %rollout_path.display(),
+                line_start_byte_offset = line_start_offset,
+                line_end_byte_offset = line_end_offset,
+                expected_ordinal = next_ordinal,
+                line_ordinal = ordinal,
+                skipped_ordinal_start = next_ordinal,
+                skipped_ordinal_end_exclusive = ordinal,
+                "skipping rollout ordinal range after rejected lines"
+            );
+            projections.push(RolloutProjectionStep::SkippedOrdinalRange {
+                start_ordinal: next_ordinal,
+                end_ordinal_exclusive: ordinal,
+            });
+        }
+        pending_rejected_line_count = 0;
+        let next_line_ordinal =
+            ordinal
+                .checked_add(1)
+                .ok_or_else(|| ThreadStoreError::Internal {
+                    message: "rollout ordinal exceeds SQLite integer range".to_string(),
+                })?;
+        projections.push(RolloutProjectionStep::Line(ProjectedRolloutLine {
+            ordinal,
+            start_byte_offset: line_start_offset,
+            end_byte_offset: line_end_offset,
+            fallback_created_at_ms,
+            changes,
+        }));
+        next_ordinal = next_line_ordinal;
+        next_offset = line_end_offset;
+        line_start_offset = line_end_offset;
+    }
+    Ok((projections, next_offset))
 }
 
 fn thread_store_io_error(err: std::io::Error) -> ThreadStoreError {

codex-rs/thread-store/src/local/thread_history_materialization_tests.rs

@@ -1440,7 +1440,6 @@ async fn paginated_projection_accepts_float_rate_limits_and_later_final_answers(
         .expect("queue unprojected history");
     recorder.flush().await.expect("flush unprojected history");
     assert_eq!(projection_state(&pool, thread_id).await, checkpoint);
-
     super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
         .await
         .expect("project floating-point rate limits");
@@ -1712,6 +1711,13 @@ async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() {
             "out of order ordinal",
             format!("{}\n", rollout_line(Some(2), turn_started("turn-1"))),
         ),
+        (
+            "gap larger than rejected prefix",
+            format!(
+                "{{not json}}\n{}\n",
+                rollout_line(Some(3), turn_started("turn-1")),
+            ),
+        ),
     ];
     for (name, suffix) in cases {
         let home = TempDir::new().expect("temp dir");
@@ -1852,6 +1858,12 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
         .await
         .expect("persist session metadata");
 
+    let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing(
+        home.path().abs(),
+    ))
+    .await
+    .expect("open thread history db");
+    let before = projection_state(&pool, thread_id).await;
     let rollout_path = store
         .live_rollout_path(thread_id)
         .await
@@ -1863,6 +1875,14 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
     file.write_all(b"\n \t\r\n{not json}\n\xff\n")
         .expect("append blank and rejected lines");
     file.flush().expect("flush rejected line");
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect("leave rejected tail pending");
+    assert_eq!(
+        projection_state(&pool, thread_id).await,
+        (before.0 + 5, before.1)
+    );
+
     let recorder = store
         .live_recorders
         .lock()
@@ -1881,11 +1901,6 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
         .await
         .expect("project valid retry after rejected line");
 
-    let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing(
-        home.path().abs(),
-    ))
-    .await
-    .expect("open thread history db");
     let (expected_start_byte_offset, _) =
         rollout_line_byte_offsets(rollout_path.as_path(), /*ordinal*/ 1);
     let start_byte_offset = sqlx::query_scalar::<_, Option<i64>>(
@@ -1897,6 +1912,198 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
     .await
     .expect("read projected turn byte offset");
     assert_eq!(start_byte_offset, Some(expected_start_byte_offset));
+    let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len())
+        .expect("rollout length");
+    assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 2));
+}
+
+#[tokio::test]
+async fn unprojectable_rollout_lines_wait_for_later_ordinals() {
+    let unknown_line = |ordinal| {
+        format!(
+            concat!(
+                "{{\"timestamp\":\"2025-01-01T00:00:00.000Z\",\"ordinal\":{ordinal},",
+                "\"type\":\"future_item\",\"payload\":{{}}}}"
+            ),
+            ordinal = ordinal
+        )
+    };
+    let cases = [
+        ("unknown payload", unknown_line(1), unknown_line(2)),
+        (
+            "structurally invalid JSON",
+            "{}".to_string(),
+            "null".to_string(),
+        ),
+    ];
+    for (name, pending_line, skipped_line) in cases {
+        let home = TempDir::new().expect("temp dir");
+        let store = projection_store(home.path()).await;
+        let thread_id = ThreadId::default();
+        create_paginated_thread(&store, thread_id).await;
+        store
+            .persist_thread(thread_id)
+            .await
+            .expect("persist session metadata");
+
+        let pool = codex_state::open_thread_history_db(
+            &codex_state::SqliteConfig::new_for_testing(home.path().abs()),
+        )
+        .await
+        .expect("open thread history db");
+        let before = projection_state(&pool, thread_id).await;
+        let rollout_path = store
+            .live_rollout_path(thread_id)
+            .await
+            .expect("rollout path");
+        append_suffix(rollout_path.as_path(), format!("{pending_line}\n").as_str());
+
+        super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+            .await
+            .expect("leave unprojectable tail pending");
+        assert_eq!(projection_state(&pool, thread_id).await, before, "{name}");
+
+        append_suffix(
+            rollout_path.as_path(),
+            format!(
+                "{}\n{skipped_line}\n{}\n",
+                rollout_line(Some(1), turn_started("retry-turn")),
+                rollout_line(Some(3), turn_started("turn-1")),
+            )
+            .as_str(),
+        );
+
+        super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+            .await
+            .expect(name);
+
+        let rollout_len =
+            i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len())
+                .expect("rollout length");
+        assert_eq!(
+            projection_state(&pool, thread_id).await,
+            (rollout_len, 4),
+            "{name}"
+        );
+        let turn_ordinals = sqlx::query_as::<_, (String, i64)>(
+            "SELECT turn_id, rollout_ordinal FROM thread_turns WHERE thread_id = ? ORDER BY rollout_ordinal",
+        )
+        .bind(thread_id.to_string())
+        .fetch_all(&pool)
+        .await
+        .expect("read projected turns");
+        assert_eq!(
+            turn_ordinals,
+            vec![("retry-turn".to_string(), 1), ("turn-1".to_string(), 3)],
+            "{name}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn event_timestamps_allow_invalid_rollout_timestamps() {
+    let home = TempDir::new().expect("temp dir");
+    let store = projection_store(home.path()).await;
+    let thread_id = ThreadId::default();
+    create_paginated_thread(&store, thread_id).await;
+    store
+        .persist_thread(thread_id)
+        .await
+        .expect("persist session metadata");
+
+    let invalid_timestamp_line = |ordinal, item| {
+        rollout_line(Some(ordinal), item).replace("2025-01-01T00:00:00.000Z", "not-a-timestamp")
+    };
+    let rollout_path = store
+        .live_rollout_path(thread_id)
+        .await
+        .expect("rollout path");
+    append_suffix(
+        rollout_path.as_path(),
+        format!(
+            "{}\n{}\n",
+            invalid_timestamp_line(1, turn_started("turn-1")),
+            invalid_timestamp_line(
+                2,
+                completed_item(
+                    thread_id,
+                    "turn-1",
+                    agent_message("agent-1", MessagePhase::FinalAnswer),
+                ),
+            ),
+        )
+        .as_str(),
+    );
+
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect("project records with event timestamps");
+
+    let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing(
+        home.path().abs(),
+    ))
+    .await
+    .expect("open thread history db");
+    let created_at_ms = sqlx::query_scalar::<_, i64>(
+        "SELECT created_at_ms FROM thread_items WHERE thread_id = ? AND turn_id = ? AND item_id = ?",
+    )
+    .bind(thread_id.to_string())
+    .bind("turn-1")
+    .bind("agent-1")
+    .fetch_one(&pool)
+    .await
+    .expect("read projected item timestamp");
+    assert_eq!(created_at_ms, 0);
+    let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len())
+        .expect("rollout length");
+    assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 3));
+}
+
+#[tokio::test]
+async fn malformed_rollout_lines_skip_inferred_ordinal_gaps() {
+    let home = TempDir::new().expect("temp dir");
+    let store = projection_store(home.path()).await;
+    let thread_id = ThreadId::default();
+    create_paginated_thread(&store, thread_id).await;
+    store
+        .persist_thread(thread_id)
+        .await
+        .expect("persist session metadata");
+
+    let rollout_path = store
+        .live_rollout_path(thread_id)
+        .await
+        .expect("rollout path");
+    append_suffix(
+        rollout_path.as_path(),
+        format!(
+            "{{not json}}\n{}\n",
+            rollout_line(Some(2), turn_started("turn-1"))
+        )
+        .as_str(),
+    );
+
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect("skip malformed gap and project later history");
+
+    let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing(
+        home.path().abs(),
+    ))
+    .await
+    .expect("open thread history db");
+    let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len())
+        .expect("rollout length");
+    assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 3));
+    let turn_ordinal = sqlx::query_scalar::<_, i64>(
+        "SELECT rollout_ordinal FROM thread_turns WHERE thread_id = ? AND turn_id = ?",
+    )
+    .bind(thread_id.to_string())
+    .bind("turn-1")
+    .fetch_one(&pool)
+    .await
+    .expect("read projected turn");
+    assert_eq!(turn_ordinal, 2);
 }
 
 #[tokio::test]