diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs
index 413adb9ecc9a..a6535a77bc46 100644
--- a/codex-rs/thread-store/src/local/live_writer.rs
+++ b/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,7 +303,6 @@ 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?;
@@ -313,7 +310,6 @@ async fn write_and_project(
             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)),
-    }
-}
diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs
index f421b1e82469..64e586725acf 100644
--- a/codex-rs/thread-store/src/local/thread_history.rs
+++ b/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,
diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs
index 380a962ea0b0..41e69f6daa3a 100644
--- a/codex-rs/thread-store/src/local/thread_history_materialization.rs
+++ b/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) {
diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs
index 6b43f60bf6be..e0c37ce3da13 100644
--- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs
+++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs
@@ -12,6 +12,7 @@ use codex_protocol::models::BaseInstructions;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::ItemCompletedEvent;
 use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::ThreadHistoryMode;
 use codex_protocol::protocol::ThreadMemoryMode;
@@ -294,6 +295,260 @@ async fn turn_creation_recovers_summary_ids_from_earlier_items() {
     );
 }
 
+#[tokio::test]
+async fn next_write_catches_up_unprojected_durable_suffix() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    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(home.path())
+        .await
+        .expect("open thread history db");
+    let checkpoint = projection_state(&pool, thread_id).await;
+    store
+        .append_items(AppendThreadItemsParams {
+            thread_id,
+            items: vec![turn_started("turn-1")],
+        })
+        .await
+        .expect("append turn start");
+
+    let thread_id_string = thread_id.to_string();
+    sqlx::query("DELETE FROM thread_turns WHERE thread_id = ?")
+        .bind(thread_id_string.as_str())
+        .execute(&pool)
+        .await
+        .expect("remove projected turn");
+    sqlx::query(
+        r#"
+UPDATE thread_history_projection_state
+SET next_rollout_byte_offset = ?, next_rollout_ordinal = ?
+WHERE thread_id = ?
+        "#,
+    )
+    .bind(checkpoint.0)
+    .bind(checkpoint.1)
+    .bind(thread_id_string.as_str())
+    .execute(&pool)
+    .await
+    .expect("rewind projection state");
+
+    store
+        .append_items(AppendThreadItemsParams {
+            thread_id,
+            items: vec![completed_item(
+                thread_id,
+                "turn-1",
+                TurnItem::UserMessage(UserMessageItem {
+                    id: "user-1".to_string(),
+                    client_id: None,
+                    content: Vec::new(),
+                }),
+            )],
+        })
+        .await
+        .expect("append after simulated projection failure");
+
+    let rows = sqlx::query_as::<_, (String, String)>(
+        r#"
+SELECT
+    (SELECT status FROM thread_turns WHERE thread_id = ? AND turn_id = 'turn-1'),
+    (SELECT item_id FROM thread_items WHERE thread_id = ? AND turn_id = 'turn-1')
+        "#,
+    )
+    .bind(thread_id_string.as_str())
+    .bind(thread_id_string.as_str())
+    .fetch_one(&pool)
+    .await
+    .expect("read recovered rows");
+    assert_eq!(rows, ("inProgress".to_string(), "user-1".to_string()));
+
+    let rollout_path = store
+        .live_rollout_path(thread_id)
+        .await
+        .expect("rollout path");
+    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 synchronized_catch_up_does_not_replay_old_rows() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let thread_id = ThreadId::default();
+    create_paginated_thread(&store, thread_id).await;
+    store
+        .append_items(AppendThreadItemsParams {
+            thread_id,
+            items: vec![turn_started("turn-1")],
+        })
+        .await
+        .expect("append turn start");
+
+    let pool = codex_state::open_thread_history_db(home.path())
+        .await
+        .expect("open thread history db");
+    let before = projection_state(&pool, thread_id).await;
+    sqlx::query("UPDATE thread_turns SET status = 'sentinel' WHERE thread_id = ?")
+        .bind(thread_id.to_string())
+        .execute(&pool)
+        .await
+        .expect("mark projected turn");
+    let rollout_path = store
+        .live_rollout_path(thread_id)
+        .await
+        .expect("rollout path");
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect("catch up synchronized rollout");
+
+    assert_eq!(projection_state(&pool, thread_id).await, before);
+    let status =
+        sqlx::query_scalar::<_, String>("SELECT status FROM thread_turns WHERE thread_id = ?")
+            .bind(thread_id.to_string())
+            .fetch_one(&pool)
+            .await
+            .expect("read projected turn");
+    assert_eq!(status, "sentinel");
+}
+
+#[tokio::test]
+async fn catch_up_leaves_trailing_partial_line_unprojected() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    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(home.path())
+        .await
+        .expect("open thread history db");
+    let before = projection_state(&pool, thread_id).await;
+    let complete_line = rollout_line(Some(1), turn_started("turn-1"));
+    let partial_line = rollout_line(
+        Some(2),
+        completed_item(
+            thread_id,
+            "turn-1",
+            TurnItem::UserMessage(UserMessageItem {
+                id: "user-1".to_string(),
+                client_id: None,
+                content: Vec::new(),
+            }),
+        ),
+    );
+    let complete_suffix = format!("{complete_line}\n");
+    let rollout_path = store
+        .live_rollout_path(thread_id)
+        .await
+        .expect("rollout path");
+    append_suffix(
+        rollout_path.as_path(),
+        format!("{complete_suffix}{partial_line}").as_str(),
+    );
+
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect("catch up complete suffix");
+
+    let expected_offset =
+        before.0 + i64::try_from(complete_suffix.len()).expect("complete suffix byte count");
+    assert_eq!(
+        projection_state(&pool, thread_id).await,
+        (expected_offset, 2)
+    );
+    let counts = sqlx::query_as::<_, (i64, i64)>(
+        r#"
+SELECT
+    (SELECT COUNT(*) FROM thread_turns WHERE thread_id = ?),
+    (SELECT COUNT(*) FROM thread_items WHERE thread_id = ?)
+        "#,
+    )
+    .bind(thread_id.to_string())
+    .bind(thread_id.to_string())
+    .fetch_one(&pool)
+    .await
+    .expect("read projected row counts");
+    assert_eq!(counts, (1, 0));
+}
+
+#[tokio::test]
+async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() {
+    let cases = [
+        (
+            "missing ordinal",
+            format!(
+                "{}\n",
+                rollout_line(/*ordinal*/ None, turn_started("turn-1"))
+            ),
+        ),
+        (
+            "duplicate ordinal",
+            format!(
+                "{}\n{}\n",
+                rollout_line(Some(1), turn_started("turn-1")),
+                rollout_line(Some(1), turn_started("turn-2")),
+            ),
+        ),
+        (
+            "out of order ordinal",
+            format!("{}\n", rollout_line(Some(2), turn_started("turn-1"))),
+        ),
+    ];
+    for (name, suffix) in cases {
+        let home = TempDir::new().expect("temp dir");
+        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        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(home.path())
+            .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(), suffix.as_str());
+
+        super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+            .await
+            .expect_err(name);
+
+        assert_eq!(
+            projection_state(&pool, thread_id).await,
+            before,
+            "{name} should not advance projection state"
+        );
+        let counts = sqlx::query_as::<_, (i64, i64)>(
+            r#"
+SELECT
+    (SELECT COUNT(*) FROM thread_turns WHERE thread_id = ?),
+    (SELECT COUNT(*) FROM thread_items WHERE thread_id = ?)
+            "#,
+        )
+        .bind(thread_id.to_string())
+        .bind(thread_id.to_string())
+        .fetch_one(&pool)
+        .await
+        .expect("read projected row counts");
+        assert_eq!(counts, (0, 0), "{name} should not project rows");
+    }
+}
+
 #[tokio::test]
 async fn jsonl_failure_does_not_create_projection_database() {
     let home = TempDir::new().expect("temp dir");
@@ -313,6 +568,31 @@ async fn jsonl_failure_does_not_create_projection_database() {
     assert!(!codex_state::thread_history_db_path(home.path()).exists());
 }
 
+#[tokio::test]
+async fn catch_up_rejects_missing_rollout_after_projection() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    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");
+    store
+        .shutdown_thread(thread_id)
+        .await
+        .expect("close rollout");
+    fs::remove_file(rollout_path.as_path()).expect("remove rollout");
+
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
+        .await
+        .expect_err("missing projected rollout should fail");
+}
+
 #[tokio::test]
 async fn sqlite_failure_does_not_fail_durable_jsonl_write() {
     let home = TempDir::new().expect("temp dir");
@@ -363,9 +643,6 @@ async fn rejected_rollout_line_does_not_poison_projection() {
         .live_rollout_path(thread_id)
         .await
         .expect("rollout path");
-    let start_offset = fs::metadata(rollout_path.as_path())
-        .expect("rollout metadata")
-        .len();
     let mut file = fs::OpenOptions::new()
         .append(true)
         .open(rollout_path.as_path())
@@ -387,7 +664,7 @@ async fn rejected_rollout_line_does_not_poison_projection() {
         .expect("queue valid retry");
     recorder.flush().await.expect("flush valid retry");
 
-    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path(), start_offset)
+    super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path())
         .await
         .expect("project valid retry after rejected line");
 
@@ -558,3 +835,35 @@ fn completed_item(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> Rollout
         completed_at_ms: 1,
     }))
 }
+
+async fn projection_state(pool: &sqlx::SqlitePool, thread_id: ThreadId) -> (i64, i64) {
+    sqlx::query_as::<_, (i64, i64)>(
+        r#"
+SELECT next_rollout_byte_offset, next_rollout_ordinal
+FROM thread_history_projection_state
+WHERE thread_id = ?
+        "#,
+    )
+    .bind(thread_id.to_string())
+    .fetch_one(pool)
+    .await
+    .expect("read projection state")
+}
+
+fn rollout_line(ordinal: Option<u64>, item: RolloutItem) -> String {
+    serde_json::to_string(&RolloutLine {
+        timestamp: "2025-01-01T00:00:00.000Z".to_string(),
+        ordinal,
+        item,
+    })
+    .expect("serialize rollout line")
+}
+
+fn append_suffix(rollout_path: &std::path::Path, suffix: &str) {
+    let mut file = fs::OpenOptions::new()
+        .append(true)
+        .open(rollout_path)
+        .expect("open rollout suffix");
+    file.write_all(suffix.as_bytes()).expect("append suffix");
+    file.flush().expect("flush suffix");
+}