diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index 358736d610..3347c1e44d 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/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( diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 0249417eea..9f3968fa5f 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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> { diff --git a/codex-rs/thread-store/src/local/paginated_fork.rs b/codex-rs/thread-store/src/local/paginated_fork.rs index d3317b9c83..810443a7a2 100644 --- a/codex-rs/thread-store/src/local/paginated_fork.rs +++ b/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() diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index d1fedb3bf6..b20cc17989 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/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> { + 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 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 823068a78c..989715493b 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/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); 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 5fa69246d1..ae4f4c46f8 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 @@ -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,