Gate paginated thread history on the state database (#35787)

## Why

Local thread stores without an initialized state database should not implicitly
create SQLite files or partially delete threads that have materialized history.

## What changed

- Report paginated history listing as unsupported when no state database is
  available, and skip history projection and materialization in that mode.
- Reject paginated forks without a state database.
- Validate access to materialized history before deleting rollout files, so a
  failed deletion preserves both the rollout and its history rows.

## Testing

Added coverage that a store without a state database creates no SQLite files
and preserves materialized history when deletion is unsupported. Existing
projection tests now initialize the state runtime explicitly.

GitOrigin-RevId: 2eaa0f5f0de8d5e4d84375ec50b310746a0ee68e

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

@@ -154,6 +154,8 @@ async fn delete_thread_after_reference_check(
             });
         }
     }
+    super::thread_history::delete_thread(store, thread_id).await?;
+
     // Drop the recorder before removing files, but retain its writer lock until cleanup finishes.
     if let Some(writer_lock) = store
         .live_recorders
@@ -173,9 +175,6 @@ async fn delete_thread_after_reference_check(
         .map_err(|err| ThreadStoreError::Internal {
             message: format!("failed to delete thread name index entries for {thread_id}: {err}"),
         })?;
-    // Keep this before ThreadNotFound so a retry can finish cleanup after an earlier attempt
-    // already removed the rollout file.
-    super::thread_history::delete_thread(store, thread_id).await?;
 
     if !found_rollout_path {
         return Err(ThreadStoreError::ThreadNotFound { thread_id });
@@ -504,10 +503,86 @@ mod tests {
         assert!(!delete_rollout_file(&store, path.as_path(), thread_id).expect("delete rollout"));
     }
 
+    #[tokio::test]
+    async fn delete_thread_without_state_db_preserves_materialized_thread_history() {
+        let home = TempDir::new().expect("temp dir");
+        let config = test_config(home.path());
+        let store = LocalThreadStore::new(config.clone(), /*state_db*/ None);
+        let uuid = Uuid::from_u128(312);
+        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
+        let rollout_path = write_session_file_with_history_mode(
+            home.path(),
+            "2025-01-03T12-00-00",
+            uuid,
+            ThreadHistoryMode::Paginated,
+        )
+        .expect("session file");
+        let pool = codex_state::open_thread_history_db(&config.sqlite)
+            .await
+            .expect("open existing thread history database");
+        let thread_id_string = thread_id.to_string();
+        sqlx::query(
+            "INSERT INTO thread_turns (thread_id, turn_id, rollout_ordinal, status) VALUES (?, 'turn-1', 1, 'completed')",
+        )
+        .bind(thread_id_string.as_str())
+        .execute(&pool)
+        .await
+        .expect("insert turn");
+        sqlx::query(
+            "INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_json) VALUES (?, 'turn-1', 'item-1', 2, 1, '{}')",
+        )
+        .bind(thread_id_string.as_str())
+        .execute(&pool)
+        .await
+        .expect("insert item");
+        sqlx::query(
+            "INSERT INTO thread_history_projection_state (thread_id, next_rollout_byte_offset, next_rollout_ordinal) VALUES (?, 3, 3)",
+        )
+        .bind(thread_id_string.as_str())
+        .execute(&pool)
+        .await
+        .expect("insert projection state");
+
+        let error = store
+            .delete_thread(DeleteThreadParams { thread_id })
+            .await
+            .expect_err("projected history without a state database should prevent deletion");
+
+        assert!(matches!(
+            error,
+            ThreadStoreError::Unsupported {
+                operation: "paginated_history"
+            }
+        ));
+        assert!(rollout_path.exists());
+        let counts = sqlx::query_as::<_, (i64, i64, i64)>(
+            r#"
+SELECT
+    (SELECT COUNT(*) FROM thread_turns WHERE thread_id = ?),
+    (SELECT COUNT(*) FROM thread_items WHERE thread_id = ?),
+    (SELECT COUNT(*) FROM thread_history_projection_state WHERE thread_id = ?)
+            "#,
+        )
+        .bind(thread_id_string.as_str())
+        .bind(thread_id_string.as_str())
+        .bind(thread_id_string.as_str())
+        .fetch_one(&pool)
+        .await
+        .expect("read preserved history rows");
+        assert_eq!(counts, (1, 1, 1));
+    }
+
     #[tokio::test]
     async fn delete_thread_removes_materialized_thread_history() {
         let home = TempDir::new().expect("temp dir");
-        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        let config = test_config(home.path());
+        let state_db = codex_state::StateRuntime::init(
+            config.sqlite.clone(),
+            config.default_model_provider_id.clone(),
+        )
+        .await
+        .expect("initialize state database for materialized history");
+        let store = LocalThreadStore::new(config, Some(state_db));
         let uuid = Uuid::from_u128(306);
         let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
         let rollout_path = write_session_file_with_history_mode(

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

@@ -207,6 +207,11 @@ impl LocalThreadStore {
     }
 
     async fn thread_history_db(&self) -> ThreadStoreResult<&sqlx::SqlitePool> {
+        if self.state_db.is_none() {
+            return Err(ThreadStoreError::Unsupported {
+                operation: "paginated_history",
+            });
+        }
         self.thread_history_db
             .get_or_try_init(|| async {
                 codex_state::open_thread_history_db(&self.config.sqlite).await
@@ -452,7 +457,7 @@ impl ThreadStore for LocalThreadStore {
     }
 
     fn supports_paginated_history_lists(&self) -> bool {
-        true
+        self.state_db.is_some()
     }
 
     fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreFuture<'_, TurnPage> {

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

@@ -44,6 +44,11 @@ pub(super) async fn prepare(
         .ok_or_else(|| ThreadStoreError::Internal {
             message: "fork lineage has no source segment".to_string(),
         })?;
+    if store.state_db.is_none() {
+        return Err(ThreadStoreError::Unsupported {
+            operation: "prepare_fork",
+        });
+    }
     if !matches!(boundary, ForkBoundary::Latest) {
         for segment in lineage
             .segments()

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

@@ -40,6 +40,9 @@ pub(super) async fn projection_state(
     store: &LocalThreadStore,
     thread_id: ThreadId,
 ) -> ThreadStoreResult<Option<RolloutProjectionState>> {
+    if store.state_db.is_none() {
+        return Ok(None);
+    }
     let db_path = store.config.sqlite.thread_history_db_path();
     if !tokio::fs::try_exists(db_path.as_path())
         .await

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

@@ -26,6 +26,9 @@ pub(super) async fn materialize_to_sqlite(
     thread_id: ThreadId,
     rollout_path: &Path,
 ) -> ThreadStoreResult<()> {
+    if store.state_db.is_none() {
+        return Ok(());
+    }
     let start_offset = super::thread_history::projection_state(store, thread_id)
         .await?
         .map_or(0, |state| state.next_byte_offset);

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

@@ -51,6 +51,45 @@ use crate::ThreadPersistenceMetadata;
 use crate::ThreadSortKey;
 use crate::ThreadStore;
 
+#[tokio::test]
+async fn paginated_history_without_state_db_does_not_initialize_sqlite() {
+    let home = TempDir::new().expect("temp dir");
+    let config = test_config(home.path());
+    let sqlite = config.sqlite.clone();
+    let store = LocalThreadStore::new(config, /*state_db*/ None);
+    let thread_id = ThreadId::default();
+
+    assert!(!store.supports_paginated_history_lists());
+    create_paginated_thread(&store, thread_id).await;
+    store
+        .append_items(AppendThreadItemsParams {
+            thread_id,
+            items: vec![turn_started("turn-1")],
+        })
+        .await
+        .expect("append paginated rollout");
+    store
+        .persist_thread(thread_id)
+        .await
+        .expect("persist paginated rollout");
+    store
+        .flush_thread(thread_id)
+        .await
+        .expect("flush paginated rollout");
+    store
+        .shutdown_thread(thread_id)
+        .await
+        .expect("shutdown paginated rollout");
+
+    for runtime_db in sqlite.runtime_db_paths() {
+        assert!(
+            !runtime_db.path.exists(),
+            "expected no SQLite initialization for {}",
+            runtime_db.path.display()
+        );
+    }
+}
+
 /// Separate Codex and SQLite homes must work together across startup backfill,
 /// thread listing, and projection-backed paginated history reads.
 #[tokio::test]
@@ -242,7 +281,7 @@ async fn split_homes_support_backfill_listing_and_paginated_history() {
 #[tokio::test]
 async fn paginated_live_append_materializes_turn_items_and_state() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -383,7 +422,7 @@ WHERE thread_id = ?
 #[tokio::test]
 async fn referenced_paginated_rollout_projects_inherited_ordinal_range() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let source_id = ThreadId::default();
     create_paginated_thread(&store, source_id).await;
     store
@@ -498,7 +537,7 @@ async fn referenced_paginated_rollout_projects_inherited_ordinal_range() {
 #[tokio::test]
 async fn named_fork_boundaries_reject_invisible_and_noncanonical_turns() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let source_id = ThreadId::default();
     create_paginated_thread(&store, source_id).await;
     store
@@ -601,7 +640,7 @@ async fn named_fork_boundaries_reject_invisible_and_noncanonical_turns() {
 #[tokio::test]
 async fn active_turn_stores_only_its_start_position() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
 
@@ -662,7 +701,7 @@ async fn active_turn_stores_only_its_start_position() {
 #[tokio::test]
 async fn paginated_fork_persists_empty_source() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     let rollout_path = store
@@ -684,7 +723,7 @@ async fn paginated_fork_persists_empty_source() {
 #[tokio::test]
 async fn paginated_fork_materializes_compressed_source_and_ancestor() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let ancestor_thread_id = ThreadId::default();
     create_paginated_thread(&store, ancestor_thread_id).await;
     store
@@ -794,7 +833,7 @@ async fn paginated_fork_materializes_compressed_source_and_ancestor() {
 #[tokio::test]
 async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_finishes() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let ancestor_thread_id = ThreadId::default();
     create_paginated_thread(&store, ancestor_thread_id).await;
     store
@@ -878,7 +917,7 @@ async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_fini
 #[tokio::test]
 async fn prepared_fork_reserves_source_until_child_reference_is_durable() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let source_thread_id = ThreadId::default();
     create_paginated_thread(&store, source_thread_id).await;
     store
@@ -958,7 +997,7 @@ async fn prepared_fork_reserves_source_until_child_reference_is_durable() {
 #[tokio::test]
 async fn subagent_prefix_advances_projection_without_materializing_history() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_subagent_thread(
         &store,
@@ -1039,7 +1078,7 @@ async fn subagent_prefix_advances_projection_without_materializing_history() {
 #[tokio::test]
 async fn unexpected_duplicate_item_completion_does_not_poison_projection() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
 
@@ -1118,7 +1157,7 @@ WHERE thread_id = ? AND turn_id = ? AND item_id = ?
 #[tokio::test]
 async fn terminal_turn_does_not_change_after_later_records() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
 
@@ -1325,7 +1364,7 @@ async fn summary_items_use_final_answers_and_ignore_commentary() {
 #[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 store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1409,7 +1448,7 @@ SELECT
 #[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 store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1452,7 +1491,7 @@ async fn synchronized_catch_up_does_not_replay_old_rows() {
 #[tokio::test]
 async fn catch_up_preserves_trailing_partial_line_boundaries() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1531,7 +1570,7 @@ async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() {
     ];
     for (name, suffix) in cases {
         let home = TempDir::new().expect("temp dir");
-        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        let store = projection_store(home.path()).await;
         let thread_id = ThreadId::default();
         create_paginated_thread(&store, thread_id).await;
         store
@@ -1602,7 +1641,7 @@ async fn jsonl_failure_does_not_create_projection_database() {
 #[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 store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1627,11 +1666,9 @@ async fn catch_up_rejects_missing_rollout_after_projection() {
 #[tokio::test]
 async fn sqlite_failure_does_not_fail_durable_jsonl_write() {
     let home = TempDir::new().expect("temp dir");
-    let sqlite_home = home.path().join("not-a-directory");
-    fs::write(sqlite_home.as_path(), "not a directory").expect("block sqlite home");
-    let mut config = test_config(home.path());
-    config.sqlite = codex_state::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
-    let store = LocalThreadStore::new(config, /*state_db*/ None);
+    let store = projection_store(home.path()).await;
+    fs::create_dir(store.config.sqlite.thread_history_db_path())
+        .expect("block thread history database");
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
 
@@ -1662,7 +1699,7 @@ async fn sqlite_failure_does_not_fail_durable_jsonl_write() {
 #[tokio::test]
 async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1720,7 +1757,7 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() {
 #[tokio::test]
 async fn shutdown_materializes_items_queued_without_a_flush() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     let recorder = store
@@ -1760,7 +1797,7 @@ async fn shutdown_materializes_items_queued_without_a_flush() {
 #[tokio::test]
 async fn delete_waits_for_in_flight_projection_before_removing_rows() {
     let home = TempDir::new().expect("temp dir");
-    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let store = projection_store(home.path()).await;
     let thread_id = ThreadId::default();
     create_paginated_thread(&store, thread_id).await;
     store
@@ -1817,6 +1854,17 @@ SELECT
     assert_eq!(counts, (0, 0, 0));
 }
 
+async fn projection_store(codex_home: &Path) -> LocalThreadStore {
+    let config = test_config(codex_home);
+    let state_db = codex_state::StateRuntime::init(
+        config.sqlite.clone(),
+        config.default_model_provider_id.clone(),
+    )
+    .await
+    .expect("initialize state database for paginated history");
+    LocalThreadStore::new(config, Some(state_db))
+}
+
 async fn create_paginated_thread(store: &LocalThreadStore, thread_id: ThreadId) {
     create_paginated_subagent_thread(
         store, thread_id, /*history_base*/ None, /*subagent_history_start_ordinal*/ None,