Record rollout boundaries for materialized turns (#34562)

## What changed

- Store the starting byte offset and terminal ordinal and byte offset for each turn projected into SQLite.
- Advance offsets across blank and rejected physical lines while leaving incomplete trailing lines for the next materialization pass.
- Stop later rollout records from changing a turn after it reaches a terminal state.

## Testing

- Cover active and completed turn positions, trailing partial lines, skipped invalid lines, subagent history prefixes, and records received after terminal state.

GitOrigin-RevId: a7da2f0b00c0bf47b841db874c79a0d495c18acd

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

@@ -11,9 +11,16 @@ use tokio::io::AsyncSeekExt;
 use tracing::warn;
 
 use super::LocalThreadStore;
+use super::thread_history::ProjectedRolloutLine;
 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,
@@ -21,6 +28,7 @@ pub(super) async fn materialize_to_sqlite(
 ) -> ThreadStoreResult<()> {
     let start_offset = super::thread_history::next_rollout_byte_offset(store, thread_id).await?;
     let (lines, next_offset) = read_complete_rollout_lines(rollout_path, start_offset).await?;
+    // Empty valid records can still consume bytes through blank or rejected complete lines.
     if lines.is_empty() && start_offset == next_offset {
         return Ok(());
     }
@@ -32,18 +40,26 @@ pub(super) async fn materialize_to_sqlite(
 
     let projections = lines
         .iter()
-        .map(|line| {
+        .map(|record| {
+            let line = &record.line;
+            let ordinal = line.ordinal.ok_or_else(|| ThreadStoreError::Internal {
+                message: format!("paginated rollout line for {thread_id} is missing an ordinal"),
+            })?;
             let created_at_ms = DateTime::parse_from_rfc3339(line.timestamp.as_str())
                 .map(|timestamp| timestamp.timestamp_millis())
                 .map_err(thread_history_error)?;
-            let changes = if line.ordinal.is_some_and(|ordinal| {
-                subagent_history_start_ordinal.is_some_and(|start| ordinal < start)
-            }) {
+            let changes = if subagent_history_start_ordinal.is_some_and(|start| ordinal < start) {
                 ThreadHistoryChangeSet::default()
             } else {
                 project_rollout_line(line)
             };
-            Ok((line.ordinal, created_at_ms, changes))
+            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(
@@ -59,7 +75,7 @@ pub(super) async fn materialize_to_sqlite(
 async fn read_complete_rollout_lines(
     rollout_path: &Path,
     start_offset: u64,
-) -> ThreadStoreResult<(Vec<RolloutLine>, u64)> {
+) -> ThreadStoreResult<(Vec<CompleteRolloutLine>, u64)> {
     let next_offset = match tokio::fs::metadata(rollout_path).await {
         Ok(metadata) => metadata.len(),
         Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => {
@@ -86,6 +102,8 @@ async fn read_complete_rollout_lines(
     file.read_exact(bytes.as_mut_slice())
         .await
         .map_err(thread_store_io_error)?;
+    // Only project the newline-terminated prefix; leave a trailing partial record for the next
+    // pass.
     let complete_byte_count = bytes
         .iter()
         .rposition(|byte| *byte == b'\n')
@@ -99,18 +117,39 @@ async fn read_complete_rollout_lines(
         .ok_or_else(|| ThreadStoreError::Internal {
             message: "durable rollout byte offset overflow".to_string(),
         })?;
-    let text = std::str::from_utf8(&bytes[..complete_byte_count]).map_err(thread_history_error)?;
     let mut lines = Vec::new();
-    for line in text.lines().filter(|line| !line.is_empty()) {
-        match serde_json::from_str(line) {
-            Ok(line) => lines.push(line),
-            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}");
+    let mut line_start_offset = start_offset;
+    // Preserve each complete physical line's trailing newline so byte offsets advance through
+    // every durable byte, including blank or rejected lines that do not project a row.
+    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(|_| {
+                ThreadStoreError::Internal {
+                    message: "durable rollout byte offset overflow".to_string(),
+                }
+            })?)
+            .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) {
+            match serde_json::from_slice(line_bytes) {
+                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;
     }
     Ok((lines, next_offset))
 }