Resume thread history projection from its SQLite checkpoint

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

@@ -151,13 +151,11 @@ pub(super) async fn shutdown_thread(
     if matches!(history_mode, ThreadHistoryMode::Legacy) {
         recorder.shutdown().await.map_err(thread_store_io_error)?;
     } else {
-        let start_offset = rollout_len_or_zero(rollout_path.as_path()).await?;
         recorder.shutdown().await.map_err(thread_store_io_error)?;
         if let Err(err) = super::thread_history_materialization::materialize_to_sqlite(
             store,
             thread_id,
             rollout_path.as_path(),
-            start_offset,
         )
         .await
         {
@@ -305,15 +303,13 @@ async fn write_and_project(
         durable_write(&recorder, write_op).await?;
     } else {
         let rollout_path = recorder.rollout_path();
-        let start_offset = rollout_len_or_zero(rollout_path).await?;
         // SQLite is a rebuildable view. The flush barrier must win before projection starts so it
         // can lag JSONL after failure, but can never get ahead of canonical history.
         durable_write(&recorder, write_op).await?;
         if let Err(err) = super::thread_history_materialization::materialize_to_sqlite(
             store,
             thread_id,
             rollout_path,
-            start_offset,
         )
         .await
         {
@@ -339,11 +335,3 @@ async fn durable_write(recorder: &RolloutRecorder, write: RolloutWriteOp) -> Thr
         RolloutWriteOp::Flush => recorder.flush().await.map_err(thread_store_io_error),
     }
 }
-
-async fn rollout_len_or_zero(rollout_path: &std::path::Path) -> ThreadStoreResult<u64> {
-    match tokio::fs::metadata(rollout_path).await {
-        Ok(metadata) => Ok(metadata.len()),
-        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(0),
-        Err(err) => Err(thread_store_io_error(err)),
-    }
-}

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

@@ -12,6 +12,32 @@ mod read;
 pub(super) use read::list_items;
 pub(super) use read::list_turns;
 
+pub(super) async fn next_rollout_byte_offset(
+    store: &LocalThreadStore,
+    thread_id: ThreadId,
+) -> ThreadStoreResult<u64> {
+    let db_path = codex_state::thread_history_db_path(store.config.sqlite_home.as_path());
+    if !tokio::fs::try_exists(db_path.as_path())
+        .await
+        .map_err(thread_history_error)?
+    {
+        return Ok(0);
+    }
+
+    let pool = store.thread_history_db().await?;
+    let offset = sqlx::query_scalar::<_, i64>(
+        "SELECT next_rollout_byte_offset FROM thread_history_projection_state WHERE thread_id = ?",
+    )
+    .bind(thread_id.to_string())
+    .fetch_optional(pool)
+    .await
+    .map_err(thread_history_error)?
+    .unwrap_or(0);
+    u64::try_from(offset).map_err(|_| ThreadStoreError::Internal {
+        message: format!("thread history projection for {thread_id} has a negative byte offset"),
+    })
+}
+
 pub(super) async fn apply_projection(
     store: &LocalThreadStore,
     thread_id: ThreadId,

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

@@ -17,9 +17,9 @@ pub(super) async fn materialize_to_sqlite(
     store: &LocalThreadStore,
     thread_id: ThreadId,
     rollout_path: &Path,
-    start_offset: u64,
 ) -> ThreadStoreResult<()> {
-    let (lines, next_offset) = read_durable_rollout_lines(rollout_path, start_offset).await?;
+    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?;
     if lines.is_empty() && start_offset == next_offset {
         return Ok(());
     }
@@ -43,14 +43,17 @@ pub(super) async fn materialize_to_sqlite(
     .await
 }
 
-async fn read_durable_rollout_lines(
+async fn read_complete_rollout_lines(
     rollout_path: &Path,
     start_offset: u64,
 ) -> ThreadStoreResult<(Vec<RolloutLine>, u64)> {
-    let next_offset = tokio::fs::metadata(rollout_path)
-        .await
-        .map_err(thread_store_io_error)?
-        .len();
+    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 => {
+            return Ok((Vec::new(), 0));
+        }
+        Err(err) => return Err(thread_store_io_error(err)),
+    };
     let byte_count =
         next_offset
             .checked_sub(start_offset)
@@ -70,12 +73,20 @@ async fn read_durable_rollout_lines(
     file.read_exact(bytes.as_mut_slice())
         .await
         .map_err(thread_store_io_error)?;
-    if bytes.last().is_some_and(|byte| *byte != b'\n') {
-        return Err(ThreadStoreError::Internal {
-            message: "durable rollout append is not newline terminated".to_string(),
-        });
-    }
-    let text = std::str::from_utf8(bytes.as_slice()).map_err(thread_history_error)?;
+    let complete_byte_count = bytes
+        .iter()
+        .rposition(|byte| *byte == b'\n')
+        .map_or(0, |index| index + 1);
+    let next_offset = start_offset
+        .checked_add(u64::try_from(complete_byte_count).map_err(|_| {
+            ThreadStoreError::Internal {
+                message: "durable rollout append exceeds addressable memory".to_string(),
+            }
+        })?)
+        .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) {