diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index c17a31915566..350fe41fa4ae 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4054,6 +4054,7 @@ name = "codex-thread-store" version = "0.0.0" dependencies = [ "chrono", + "codex-app-server-protocol", "codex-git-utils", "codex-install-context", "codex-otel", @@ -4064,6 +4065,7 @@ dependencies = [ "pretty_assertions", "serde", "serde_json", + "sqlx", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index a2ec69ed9de4..ac347a8985f5 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2515,19 +2515,18 @@ impl ThreadRequestProcessor { } err => internal_error(format!("failed to list thread items: {err}")), })?; - let data = - page.items - .into_iter() - .map(|item| { - serde_json::from_slice::(&item.materialized_thread_item_json) - .map_err(|err| { - internal_error(format!( - "failed to deserialize stored thread item {}: {err}", - item.item_key - )) - }) + let data = page + .items + .into_iter() + .map(|item| { + serde_json::from_slice::(&item.item_json).map_err(|err| { + internal_error(format!( + "failed to deserialize stored thread item {}: {err}", + item.item_id + )) }) - .collect::, _>>()?; + }) + .collect::, _>>()?; Ok(ThreadItemsListResponse { data, diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index cdac70a4eb31..4ab6349f7c63 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -78,6 +78,7 @@ pub use runtime::logs_db_filename; pub use runtime::logs_db_path; pub use runtime::memories_db_filename; pub use runtime::memories_db_path; +pub use runtime::open_thread_history_db; pub use runtime::runtime_db_path_for_corruption_error; pub use runtime::runtime_db_paths; pub use runtime::sqlite_error_detail_is_corruption; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index ad659ede5270..0055d262da62 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -23,6 +23,7 @@ use crate::migrations::runtime_goals_migrator; use crate::migrations::runtime_logs_migrator; use crate::migrations::runtime_memories_migrator; use crate::migrations::runtime_state_migrator; +use crate::migrations::runtime_thread_history_migrator; use crate::model::AgentJobRow; use crate::model::ThreadRow; use crate::model::anchor_from_item; @@ -414,6 +415,18 @@ async fn open_memories_sqlite( open_sqlite(path, migrator, MEMORIES_DB, telemetry_override).await } +/// Open and migrate the rebuildable paginated thread-history database. +pub async fn open_thread_history_db(sqlite_home: &Path) -> anyhow::Result { + let migrator = runtime_thread_history_migrator(); + open_sqlite( + thread_history_db_path(sqlite_home).as_path(), + &migrator, + THREAD_HISTORY_DB, + /*telemetry_override*/ None, + ) + .await +} + async fn open_sqlite( path: &Path, migrator: &Migrator, diff --git a/codex-rs/state/thread_history_migrations/0001_thread_history.sql b/codex-rs/state/thread_history_migrations/0001_thread_history.sql index 99ec54e052a8..8de55dde1981 100644 --- a/codex-rs/state/thread_history_migrations/0001_thread_history.sql +++ b/codex-rs/state/thread_history_migrations/0001_thread_history.sql @@ -20,6 +20,7 @@ CREATE TABLE thread_items ( turn_id TEXT NOT NULL, item_id TEXT NOT NULL, rollout_ordinal INTEGER NOT NULL, + created_at_ms INTEGER NOT NULL, item_json TEXT NOT NULL, PRIMARY KEY (thread_id, turn_id, item_id) ); diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index 35b3919c605e..8260c250a2f8 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -14,6 +14,7 @@ workspace = true [dependencies] chrono = { workspace = true, features = ["serde"] } +codex-app-server-protocol = { workspace = true } codex-git-utils = { workspace = true } codex-install-context = { workspace = true } codex-otel = { workspace = true } @@ -23,6 +24,7 @@ codex-state = { workspace = true } codex-utils-path = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sqlx = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index ecd962f64721..d7a44b805008 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -1,8 +1,8 @@ //! Local hard-delete support for persisted threads. //! //! Existing rollout files are deleted before this operation reports success. A rollout file that -//! vanishes after discovery counts as already deleted. SQLite cleanup happens at the app-server -//! layer after every associated rollout has been removed so failed deletes can be retried. +//! vanishes after discovery counts as already deleted. The app-server deletes main state DB rows +//! after every associated rollout is removed; this module deletes local history projection rows. use std::io::ErrorKind; use std::path::Path; @@ -25,10 +25,10 @@ pub(super) async fn delete_thread( params: DeleteThreadParams, ) -> ThreadStoreResult<()> { let thread_id = params.thread_id; + let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; let thread_id_str = thread_id.to_string(); let state_db_ctx = store.state_db().await; let mut rollout_paths = Vec::new(); - match find_thread_path_by_id_str( store.config.codex_home.as_path(), thread_id_str.as_str(), @@ -44,7 +44,6 @@ pub(super) async fn delete_thread( }); } } - match find_archived_thread_path_by_id_str( store.config.codex_home.as_path(), thread_id_str.as_str(), @@ -64,7 +63,9 @@ pub(super) async fn delete_thread( }); } } - + // Stop the live writer before removing files. The per-thread lock keeps new writes and + // replacements out while we find paths and clean up rollout files and history rows. + store.live_recorders.lock().await.remove(&thread_id); let found_rollout_path = !rollout_paths.is_empty(); for rollout_path in rollout_paths { delete_rollout_file(store, rollout_path.as_path(), thread_id)?; @@ -74,13 +75,14 @@ pub(super) async fn delete_thread( .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 }); } - store.live_recorders.lock().await.remove(&thread_id); - Ok(()) } @@ -133,6 +135,7 @@ fn delete_rollout_path( #[cfg(test)] mod tests { use codex_protocol::ThreadId; + use codex_protocol::protocol::ThreadHistoryMode; use pretty_assertions::assert_eq; use tempfile::TempDir; use uuid::Uuid; @@ -143,6 +146,7 @@ mod tests { use crate::local::test_support::test_config; use crate::local::test_support::write_archived_session_file; use crate::local::test_support::write_session_file; + use crate::local::test_support::write_session_file_with_history_mode; #[tokio::test] async fn delete_thread_removes_active_and_archived_rollouts() { @@ -191,6 +195,67 @@ mod tests { assert!(!delete_rollout_file(&store, path.as_path(), thread_id).expect("delete rollout")); } + #[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 uuid = Uuid::from_u128(306); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + 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(home.path()) + .await + .expect("open thread history db"); + 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"); + + store + .delete_thread(DeleteThreadParams { thread_id }) + .await + .expect("delete thread"); + + 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 remaining history rows"); + assert_eq!(counts, (0, 0, 0)); + } + #[tokio::test] async fn delete_thread_reports_missing_thread() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index d11589c6e838..413adb9ecc9a 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -1,6 +1,8 @@ use std::path::PathBuf; use codex_protocol::ThreadId; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::RolloutConfig; use codex_rollout::RolloutRecorder; @@ -26,6 +28,7 @@ pub(super) async fn create_thread( params: CreateThreadParams, ) -> ThreadStoreResult<()> { let thread_id = params.thread_id; + let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; let history_mode = params.history_mode; store.ensure_live_recorder_absent(thread_id).await?; let recorder = create_thread::create_thread(store, params).await?; @@ -38,6 +41,7 @@ pub(super) async fn resume_thread( store: &LocalThreadStore, params: ResumeThreadParams, ) -> ThreadStoreResult<()> { + let _live_writer_guard = store.live_writer_locks.lock(params.thread_id).await; store.ensure_live_recorder_absent(params.thread_id).await?; let history_mode = if let Some(history) = params.history.as_deref() { canonical_history_mode_from_rollout_items(history) @@ -115,65 +119,52 @@ pub(super) async fn append_items( store: &LocalThreadStore, params: AppendThreadItemsParams, ) -> ThreadStoreResult<()> { - // A live append should always have a recorder: create/resume installs one, while - // shutdown/discard/delete removes it. Keep the lookup defensive so late appends fail after - // teardown. - let (recorder, history_mode) = store - .live_recorders - .lock() - .await - .get(¶ms.thread_id) - .map(|entry| (entry.recorder.clone(), entry.history_mode)) - .ok_or(ThreadStoreError::ThreadNotFound { - thread_id: params.thread_id, - })?; - let persisted_items = persisted_rollout_items(params.items.as_slice(), history_mode); - if persisted_items.is_empty() { - return Ok(()); - } - recorder - .record_canonical_items(persisted_items.as_slice()) - .await - .map_err(thread_store_io_error)?; - // LiveThread applies metadata immediately after append_items returns. Wait for the local - // writer so SQLite never gets ahead of JSONL for accepted live appends. - recorder.flush().await.map_err(thread_store_io_error) + write_and_project( + store, + params.thread_id, + RolloutWriteOp::AppendItems(params.items), + ) + .await } pub(super) async fn persist_thread( store: &LocalThreadStore, thread_id: ThreadId, ) -> ThreadStoreResult<()> { - store - .live_recorder(thread_id) - .await? - .persist() - .await - .map_err(thread_store_io_error)?; - sync_materialized_rollout_path(store, thread_id).await + write_and_project(store, thread_id, RolloutWriteOp::Persist).await } pub(super) async fn flush_thread( store: &LocalThreadStore, thread_id: ThreadId, ) -> ThreadStoreResult<()> { - store - .live_recorder(thread_id) - .await? - .flush() - .await - .map_err(thread_store_io_error)?; - sync_materialized_rollout_path(store, thread_id).await + write_and_project(store, thread_id, RolloutWriteOp::Flush).await } pub(super) async fn shutdown_thread( store: &LocalThreadStore, thread_id: ThreadId, ) -> ThreadStoreResult<()> { - let recorder = store.live_recorder(thread_id).await?; + let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; + let (recorder, history_mode) = live_writer_parts(store, thread_id).await?; let rollout_path = recorder.rollout_path().to_path_buf(); - recorder.shutdown().await.map_err(thread_store_io_error)?; - sync_materialized_rollout_path(store, thread_id).await?; + 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 + { + warn!("failed to project durable rollout during shutdown for {thread_id}: {err}"); + } + } + sync_materialized_rollout_path(store, thread_id, rollout_path.as_path()).await?; if let Some(metrics) = codex_otel::global() && let Ok(metadata) = tokio::fs::metadata(rollout_path).await { @@ -188,6 +179,7 @@ pub(super) async fn discard_thread( store: &LocalThreadStore, thread_id: ThreadId, ) -> ThreadStoreResult<()> { + let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; store .live_recorders .lock() @@ -215,9 +207,9 @@ pub(super) async fn rollout_path( async fn sync_materialized_rollout_path( store: &LocalThreadStore, thread_id: ThreadId, + rollout_path: &std::path::Path, ) -> ThreadStoreResult<()> { - let rollout_path = rollout_path(store, thread_id).await?; - if codex_rollout::existing_rollout_path(rollout_path.as_path()) + if codex_rollout::existing_rollout_path(rollout_path) .await .is_none() { @@ -238,7 +230,7 @@ async fn sync_materialized_rollout_path( return Ok(()); }; if metadata.rollout_path != rollout_path { - metadata.rollout_path = rollout_path; + metadata.rollout_path = rollout_path.to_path_buf(); state_db .upsert_thread(&metadata) .await @@ -260,3 +252,98 @@ fn thread_store_io_error(err: std::io::Error) -> ThreadStoreError { message: err.to_string(), } } + +/// The rollout writer has three distinct lifecycle moments: +/// - `AppendItems` is normal turn/event persistence and adds new rollout records. +/// - `Persist` makes the thread durable before any turn items exist; locally this can write the +/// initial `SessionMeta`. +/// - `Flush` writes any rollout records already queued in the recorder and ensures they are +/// durably persisted. +/// +/// Each can advance the rollout JSONL file on disk, so we need to make sure we materialize the +/// new data into the SQLite history tables (turns and items) as necessary. +enum RolloutWriteOp { + AppendItems(Vec), + Persist, + Flush, +} + +async fn live_writer_parts( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> ThreadStoreResult<(RolloutRecorder, ThreadHistoryMode)> { + let live_recorders = store.live_recorders.lock().await; + let entry = live_recorders + .get(&thread_id) + .ok_or(ThreadStoreError::ThreadNotFound { thread_id })?; + Ok((entry.recorder.clone(), entry.history_mode)) +} + +async fn write_and_project( + store: &LocalThreadStore, + thread_id: ThreadId, + write_op: RolloutWriteOp, +) -> ThreadStoreResult<()> { + // Every live write should have a recorder: create/resume installs one, while + // shutdown/discard/delete removes it. Keep the lookup defensive so late writes fail after + // teardown. + let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; + let (recorder, history_mode) = live_writer_parts(store, thread_id).await?; + let sync_rollout_path = matches!(&write_op, RolloutWriteOp::Persist | RolloutWriteOp::Flush); + let write_op = match write_op { + RolloutWriteOp::AppendItems(items) => { + let items = persisted_rollout_items(items.as_slice(), history_mode); + if items.is_empty() { + return Ok(()); + } + RolloutWriteOp::AppendItems(items) + } + RolloutWriteOp::Persist => RolloutWriteOp::Persist, + RolloutWriteOp::Flush => RolloutWriteOp::Flush, + }; + if matches!(history_mode, ThreadHistoryMode::Legacy) { + 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 + { + warn!("failed to project durable rollout for {thread_id}: {err}"); + } + } + if sync_rollout_path { + sync_materialized_rollout_path(store, thread_id, recorder.rollout_path()).await?; + } + Ok(()) +} + +async fn durable_write(recorder: &RolloutRecorder, write: RolloutWriteOp) -> ThreadStoreResult<()> { + match write { + RolloutWriteOp::AppendItems(items) => { + recorder + .record_canonical_items(items.as_slice()) + .await + .map_err(thread_store_io_error)?; + recorder.flush().await.map_err(thread_store_io_error) + } + RolloutWriteOp::Persist => recorder.persist().await.map_err(thread_store_io_error), + RolloutWriteOp::Flush => recorder.flush().await.map_err(thread_store_io_error), + } +} + +async fn rollout_len_or_zero(rollout_path: &std::path::Path) -> ThreadStoreResult { + 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/mod.rs b/codex-rs/thread-store/src/local/mod.rs index e7e3f1e5d491..922a09c5ac3d 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -7,6 +7,8 @@ mod live_writer; mod model_context; mod read_thread; mod search_threads; +mod thread_history; +mod thread_history_materialization; mod unarchive_thread; mod update_thread_metadata; @@ -22,12 +24,17 @@ use std::collections::hash_map::Entry; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::OnceCell; +use tokio::sync::OwnedMutexGuard; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::ItemPage; +use crate::ListItemsParams; use crate::ListThreadsParams; +use crate::ListTurnsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; @@ -42,6 +49,7 @@ use crate::ThreadStore; use crate::ThreadStoreError; use crate::ThreadStoreFuture; use crate::ThreadStoreResult; +use crate::TurnPage; use crate::UpdateThreadMetadataParams; /// Local filesystem/SQLite-backed implementation of [`ThreadStore`]. @@ -61,7 +69,9 @@ use crate::UpdateThreadMetadataParams; pub struct LocalThreadStore { pub(super) config: LocalThreadStoreConfig, live_recorders: Arc>>, + live_writer_locks: Arc, state_db: Option, + thread_history_db: Arc>, } struct LiveRecorderEntry { @@ -72,6 +82,26 @@ struct LiveRecorderEntry { history_mode: ThreadHistoryMode, } +#[derive(Default)] +struct LiveWriterLocks { + // Keep per-thread locks after a writer goes idle. Removing one while another caller is about + // to acquire it could let two operations for the same thread run at once. + by_thread: Mutex>>>, +} + +impl LiveWriterLocks { + async fn lock(&self, thread_id: ThreadId) -> OwnedMutexGuard<()> { + let lock = self + .by_thread + .lock() + .await + .entry(thread_id) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone(); + lock.lock_owned().await + } +} + /// Process-scoped configuration for local thread storage. /// /// This describes where local storage lives. New-thread rollout metadata such @@ -108,7 +138,9 @@ impl LocalThreadStore { Self { config, live_recorders: Arc::new(Mutex::new(HashMap::new())), + live_writer_locks: Arc::new(LiveWriterLocks::default()), state_db, + thread_history_db: Arc::new(OnceCell::new()), } } @@ -117,6 +149,17 @@ impl LocalThreadStore { self.state_db.clone() } + async fn thread_history_db(&self) -> ThreadStoreResult<&sqlx::SqlitePool> { + self.thread_history_db + .get_or_try_init(|| async { + codex_state::open_thread_history_db(self.config.sqlite_home.as_path()).await + }) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to open thread history database: {err}"), + }) + } + /// Read a local rollout-backed thread by path. pub async fn read_thread_by_rollout_path( &self, @@ -138,18 +181,6 @@ impl LocalThreadStore { live_writer::rollout_path(self, thread_id).await } - pub(super) async fn live_recorder( - &self, - thread_id: ThreadId, - ) -> ThreadStoreResult { - self.live_recorders - .lock() - .await - .get(&thread_id) - .map(|entry| entry.recorder.clone()) - .ok_or(ThreadStoreError::ThreadNotFound { thread_id }) - } - pub(super) async fn ensure_live_recorder_absent( &self, thread_id: ThreadId, @@ -237,6 +268,16 @@ impl LocalThreadStore { ) .await } + + /// Lists projection-backed turns without enabling app-server routing yet. + pub async fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreResult { + thread_history::list_turns(self, params).await + } + + /// Lists projection-backed items without enabling app-server routing yet. + pub async fn list_items(&self, params: ListItemsParams) -> ThreadStoreResult { + thread_history::list_items(self, params).await + } } impl ThreadStore for LocalThreadStore { diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs new file mode 100644 index 000000000000..f421b1e82469 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -0,0 +1,347 @@ +use codex_app_server_protocol::ThreadHistoryChangeSet; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::TurnStatus; +use codex_protocol::ThreadId; + +use super::LocalThreadStore; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +mod read; + +pub(super) use read::list_items; +pub(super) use read::list_turns; + +pub(super) async fn apply_projection( + store: &LocalThreadStore, + thread_id: ThreadId, + start_offset: u64, + next_offset: u64, + projections: Vec<(Option, i64, ThreadHistoryChangeSet)>, +) -> ThreadStoreResult<()> { + let pool = store.thread_history_db().await?; + // Write the projected rows and advance the JSONL offset and ordinal in one transaction. If + // SQLite fails, it stays behind the durable rollout instead of claiming data it did not + // materialize. + let mut transaction = pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(thread_history_error)?; + let thread_id = thread_id.to_string(); + let projection_state = 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.as_str()) + .fetch_optional(&mut *transaction) + .await + .map_err(thread_history_error)?; + let (expected_offset, mut next_ordinal) = projection_state.unwrap_or((0, 0)); + let start_offset = sqlite_integer(start_offset, "rollout byte offset")?; + if expected_offset != start_offset { + return Err(ThreadStoreError::Internal { + message: format!("thread history projection for {thread_id} is behind durable rollout"), + }); + } + + for (ordinal, created_at_ms, changes) in projections { + let ordinal = ordinal + .ok_or_else(|| ThreadStoreError::Internal { + message: format!("paginated rollout line for {thread_id} is missing an ordinal"), + }) + .and_then(|ordinal| sqlite_integer(ordinal, "rollout ordinal"))?; + if ordinal != next_ordinal { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}" + ), + }); + } + apply_change_set( + &mut transaction, + thread_id.as_str(), + ordinal, + created_at_ms, + changes, + ) + .await?; + next_ordinal = next_ordinal + .checked_add(1) + .ok_or_else(|| ThreadStoreError::Internal { + message: "rollout ordinal exceeds SQLite integer range".to_string(), + })?; + } + + sqlx::query( + r#" +INSERT INTO thread_history_projection_state ( + thread_id, + next_rollout_byte_offset, + next_rollout_ordinal +) VALUES (?, ?, ?) +ON CONFLICT(thread_id) DO UPDATE SET + next_rollout_byte_offset = excluded.next_rollout_byte_offset, + next_rollout_ordinal = excluded.next_rollout_ordinal + "#, + ) + .bind(thread_id.as_str()) + .bind(sqlite_integer(next_offset, "rollout byte offset")?) + .bind(next_ordinal) + .execute(&mut *transaction) + .await + .map_err(thread_history_error)?; + transaction.commit().await.map_err(thread_history_error) +} + +pub(super) async fn delete_thread( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> ThreadStoreResult<()> { + 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_delete_error)? + { + return Ok(()); + } + + let pool = store.thread_history_db().await?; + let mut transaction = pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(thread_history_delete_error)?; + let thread_id = thread_id.to_string(); + sqlx::query("DELETE FROM thread_items WHERE thread_id = ?") + .bind(thread_id.as_str()) + .execute(&mut *transaction) + .await + .map_err(thread_history_delete_error)?; + sqlx::query("DELETE FROM thread_turns WHERE thread_id = ?") + .bind(thread_id.as_str()) + .execute(&mut *transaction) + .await + .map_err(thread_history_delete_error)?; + sqlx::query("DELETE FROM thread_history_projection_state WHERE thread_id = ?") + .bind(thread_id.as_str()) + .execute(&mut *transaction) + .await + .map_err(thread_history_delete_error)?; + transaction + .commit() + .await + .map_err(thread_history_delete_error) +} + +async fn apply_change_set( + transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + thread_id: &str, + rollout_ordinal: i64, + created_at_ms: i64, + changes: ThreadHistoryChangeSet, +) -> ThreadStoreResult<()> { + for turn in changes.changed_turns { + let turn_id = turn.turn_id; + let error_json = turn + .error + .as_ref() + .map(serde_json::to_string) + .transpose() + .map_err(thread_history_error)?; + // The same turn can appear again as it moves from started to completed. Update its latest + // status, error, and timestamps, but keep the rollout ordinal from the first record that + // created it. + sqlx::query( + r#" +INSERT INTO thread_turns ( + thread_id, + turn_id, + rollout_ordinal, + status, + error_json, + started_at, + completed_at, + duration_ms +) VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(thread_id, turn_id) DO UPDATE SET + status = excluded.status, + error_json = excluded.error_json, + started_at = excluded.started_at, + completed_at = excluded.completed_at, + duration_ms = excluded.duration_ms + "#, + ) + .bind(thread_id) + .bind(turn_id.as_str()) + .bind(rollout_ordinal) + .bind(turn_status(&turn.status)) + .bind(error_json) + .bind(turn.started_at) + .bind(turn.completed_at) + .bind(turn.duration_ms) + .execute(&mut **transaction) + .await + .map_err(thread_history_error)?; + + // Review turns can persist completed items before their turn lifecycle record. Fill the + // summary IDs from those older item rows when the turn row finally arrives. + sqlx::query( + r#" +UPDATE thread_turns +SET + first_user_item_id = COALESCE( + first_user_item_id, + ( + SELECT item_id + FROM thread_items + WHERE thread_id = ? + AND turn_id = ? + AND json_extract(item_json, '$.type') = 'userMessage' + ORDER BY rollout_ordinal + LIMIT 1 + ) + ), + final_agent_item_id = COALESCE( + ( + SELECT item_id + FROM thread_items + WHERE thread_id = ? + AND turn_id = ? + AND json_extract(item_json, '$.type') = 'agentMessage' + ORDER BY rollout_ordinal DESC + LIMIT 1 + ), + final_agent_item_id + ) +WHERE thread_id = ? AND turn_id = ? + "#, + ) + .bind(thread_id) + .bind(turn_id.as_str()) + .bind(thread_id) + .bind(turn_id.as_str()) + .bind(thread_id) + .bind(turn_id.as_str()) + .execute(&mut **transaction) + .await + .map_err(thread_history_error)?; + } + + for item in changes.changed_items { + let item_id = item.item.id().to_string(); + let item_json = serde_json::to_string(&item.item).map_err(thread_history_error)?; + // The same item can appear again with a newer snapshot. Replace its JSON, but keep the + // ordinal and creation timestamp from the first record so item ordering and age stay + // stable. + sqlx::query( + r#" +INSERT INTO thread_items ( + thread_id, + turn_id, + item_id, + rollout_ordinal, + created_at_ms, + item_json +) VALUES (?, ?, ?, ?, ?, ?) +ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET + item_json = excluded.item_json + "#, + ) + .bind(thread_id) + .bind(item.turn_id.as_str()) + .bind(item_id.as_str()) + .bind(rollout_ordinal) + .bind(created_at_ms) + .bind(item_json) + .execute(&mut **transaction) + .await + .map_err(thread_history_error)?; + + // Keep the first user item and latest agent item on the turn row so reads do not need to + // scan every item in the turn. + match item.item { + ThreadItem::UserMessage { .. } => { + sqlx::query( + r#" +UPDATE thread_turns +SET first_user_item_id = COALESCE(first_user_item_id, ?) +WHERE thread_id = ? AND turn_id = ? + "#, + ) + .bind(item_id.as_str()) + .bind(thread_id) + .bind(item.turn_id.as_str()) + .execute(&mut **transaction) + .await + .map_err(thread_history_error)?; + } + ThreadItem::AgentMessage { .. } => { + sqlx::query( + r#" +UPDATE thread_turns +SET final_agent_item_id = ? +WHERE thread_id = ? AND turn_id = ? + "#, + ) + .bind(item_id.as_str()) + .bind(thread_id) + .bind(item.turn_id.as_str()) + .execute(&mut **transaction) + .await + .map_err(thread_history_error)?; + } + ThreadItem::HookPrompt { .. } + | ThreadItem::Plan { .. } + | ThreadItem::Reasoning { .. } + | ThreadItem::CommandExecution { .. } + | ThreadItem::FileChange { .. } + | ThreadItem::McpToolCall { .. } + | ThreadItem::DynamicToolCall { .. } + | ThreadItem::CollabAgentToolCall { .. } + | ThreadItem::SubAgentActivity { .. } + | ThreadItem::WebSearch(_) + | ThreadItem::ImageView { .. } + | ThreadItem::Sleep { .. } + | ThreadItem::ImageGeneration(_) + | ThreadItem::EnteredReviewMode { .. } + | ThreadItem::ExitedReviewMode { .. } + | ThreadItem::ContextCompaction { .. } => {} + } + } + Ok(()) +} + +fn turn_status(status: &TurnStatus) -> &'static str { + match status { + TurnStatus::Completed => "completed", + TurnStatus::Interrupted => "interrupted", + TurnStatus::Failed => "failed", + TurnStatus::InProgress => "inProgress", + } +} + +fn sqlite_integer(value: u64, field: &str) -> ThreadStoreResult { + i64::try_from(value).map_err(|_| ThreadStoreError::Internal { + message: format!("{field} exceeds SQLite integer range"), + }) +} + +fn thread_history_error(err: impl std::fmt::Display) -> ThreadStoreError { + ThreadStoreError::Internal { + message: format!("failed to access thread history: {err}"), + } +} + +impl From for ThreadStoreError { + fn from(err: sqlx::Error) -> Self { + thread_history_error(err) + } +} + +fn thread_history_delete_error(err: impl std::fmt::Display) -> ThreadStoreError { + ThreadStoreError::Internal { + message: format!("failed to delete thread history: {err}"), + } +} diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs new file mode 100644 index 000000000000..de36e3ed4c27 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -0,0 +1,406 @@ +use codex_protocol::ThreadId; +use codex_protocol::protocol::ThreadHistoryMode; +use serde::Deserialize; +use serde::Serialize; +use sqlx::QueryBuilder; +use sqlx::Row; +use sqlx::Sqlite; + +use super::super::LocalThreadStore; +use super::thread_history_error; +use crate::ItemPage; +use crate::ListItemsParams; +use crate::ListTurnsParams; +use crate::SortDirection; +use crate::StoredThreadItem; +use crate::StoredTurn; +use crate::StoredTurnError; +use crate::StoredTurnItemsView; +use crate::StoredTurnStatus; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; +use crate::TurnPage; + +#[cfg(test)] +#[path = "read_tests.rs"] +mod tests; + +#[derive(Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +struct HistoryCursor { + thread_id: ThreadId, + scope: CursorScope, + rollout_ordinal: i64, + include_anchor: bool, +} + +#[derive(Clone, Deserialize, PartialEq, Eq, Serialize)] +#[serde(tag = "kind", rename_all = "camelCase")] +enum CursorScope { + Turns, + Items { turn_id: Option }, +} + +struct StoredTurnRow { + turn_id: String, + rollout_ordinal: i64, + status: StoredTurnStatus, + error: Option, + started_at: Option, + completed_at: Option, + duration_ms: Option, + first_user_item_id: Option, + final_agent_item_id: Option, +} + +struct StoredThreadItemRow { + item: StoredThreadItem, + rollout_ordinal: i64, +} + +pub(in crate::local) async fn list_turns( + store: &LocalThreadStore, + params: ListTurnsParams, +) -> ThreadStoreResult { + validate_thread_for_paginated_reads( + store, + params.thread_id, + params.include_archived, + "list_turns", + ) + .await?; + let scope = CursorScope::Turns; + let cursor = parse_cursor(params.cursor.as_deref(), params.thread_id, &scope)?; + let pool = store.thread_history_db().await?; + let limit = page_limit(params.page_size)?; + let mut query = QueryBuilder::::new( + r#" +SELECT + turn_id, + rollout_ordinal, + status, + error_json, + started_at, + completed_at, + duration_ms, + first_user_item_id, + final_agent_item_id +FROM thread_turns +WHERE thread_id = + "#, + ); + query.push_bind(params.thread_id.to_string()); + push_pagination_clause(&mut query, params.sort_direction, cursor.as_ref(), limit); + let rows = query + .build() + .fetch_all(pool) + .await + .map_err(thread_history_error)?; + let mut turns = rows + .into_iter() + .map(stored_turn_row) + .collect::>>()?; + let has_more = turns.len() > params.page_size; + turns.truncate(params.page_size); + + let (next_cursor, backwards_cursor) = page_cursors( + params.thread_id, + &scope, + turns.first().map(|turn| turn.rollout_ordinal), + turns.last().map(|turn| turn.rollout_ordinal), + has_more, + )?; + let mut stored_turns = Vec::with_capacity(turns.len()); + for turn in turns { + let items = match params.items_view { + StoredTurnItemsView::NotLoaded => Vec::new(), + StoredTurnItemsView::Summary => { + load_summary_items(pool, params.thread_id, &turn).await? + } + }; + stored_turns.push(StoredTurn { + turn_id: turn.turn_id, + items, + items_view: params.items_view, + status: turn.status, + error: turn.error, + started_at: turn.started_at, + completed_at: turn.completed_at, + duration_ms: turn.duration_ms, + }); + } + + Ok(TurnPage { + turns: stored_turns, + next_cursor, + backwards_cursor, + }) +} + +pub(in crate::local) async fn list_items( + store: &LocalThreadStore, + params: ListItemsParams, +) -> ThreadStoreResult { + validate_thread_for_paginated_reads( + store, + params.thread_id, + params.include_archived, + "list_items", + ) + .await?; + let scope = CursorScope::Items { + turn_id: params.turn_id.clone(), + }; + let cursor = parse_cursor(params.cursor.as_deref(), params.thread_id, &scope)?; + let pool = store.thread_history_db().await?; + let limit = page_limit(params.page_size)?; + let mut query = QueryBuilder::::new( + r#" +SELECT turn_id, item_id, rollout_ordinal, created_at_ms, item_json +FROM thread_items +WHERE thread_id = + "#, + ); + query.push_bind(params.thread_id.to_string()); + if let Some(turn_id) = params.turn_id.as_deref() { + query.push(" AND turn_id = ").push_bind(turn_id); + } + push_pagination_clause(&mut query, params.sort_direction, cursor.as_ref(), limit); + let rows = query + .build() + .fetch_all(pool) + .await + .map_err(thread_history_error)?; + let mut item_rows = rows + .into_iter() + .map(stored_thread_item_row) + .collect::>>()?; + let has_more = item_rows.len() > params.page_size; + item_rows.truncate(params.page_size); + let (next_cursor, backwards_cursor) = page_cursors( + params.thread_id, + &scope, + item_rows.first().map(|row| row.rollout_ordinal), + item_rows.last().map(|row| row.rollout_ordinal), + has_more, + )?; + let items = item_rows.into_iter().map(|row| row.item).collect(); + + Ok(ItemPage { + items, + next_cursor, + backwards_cursor, + }) +} + +async fn validate_thread_for_paginated_reads( + store: &LocalThreadStore, + thread_id: ThreadId, + include_archived: bool, + operation: &'static str, +) -> ThreadStoreResult<()> { + let Some(state_db) = store.state_db().await else { + return Err(ThreadStoreError::Unsupported { operation }); + }; + let Some(metadata) = + state_db + .get_thread(thread_id) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to read thread metadata: {err}"), + })? + else { + return Err(ThreadStoreError::Unsupported { operation }); + }; + if metadata.archived_at.is_some() && !include_archived { + return Err(ThreadStoreError::InvalidRequest { + message: format!("thread {thread_id} is archived"), + }); + } + match metadata.history_mode { + ThreadHistoryMode::Legacy => Err(ThreadStoreError::Unsupported { operation }), + ThreadHistoryMode::Paginated => Ok(()), + } +} + +fn page_limit(page_size: usize) -> ThreadStoreResult { + if page_size == 0 { + return Err(ThreadStoreError::InvalidRequest { + message: "page size must be positive".to_string(), + }); + } + let limit = page_size + .checked_add(1) + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: "page size is too large".to_string(), + })?; + i64::try_from(limit).map_err(|_| ThreadStoreError::InvalidRequest { + message: "page size is too large".to_string(), + }) +} + +fn parse_cursor( + cursor: Option<&str>, + thread_id: ThreadId, + scope: &CursorScope, +) -> ThreadStoreResult> { + let Some(cursor) = cursor else { + return Ok(None); + }; + let cursor_value: HistoryCursor = + serde_json::from_str(cursor).map_err(|_| invalid_cursor(cursor))?; + if cursor_value.thread_id != thread_id || &cursor_value.scope != scope { + return Err(invalid_cursor(cursor)); + } + Ok(Some(cursor_value)) +} + +fn push_pagination_clause( + query: &mut QueryBuilder, + direction: SortDirection, + cursor: Option<&HistoryCursor>, + limit: i64, +) { + if let Some(cursor) = cursor { + let comparator = match (direction, cursor.include_anchor) { + (SortDirection::Asc, true) => ">=", + (SortDirection::Asc, false) => ">", + (SortDirection::Desc, true) => "<=", + (SortDirection::Desc, false) => "<", + }; + query + .push(" AND rollout_ordinal ") + .push(comparator) + .push(" ") + .push_bind(cursor.rollout_ordinal); + } + let order = match direction { + SortDirection::Asc => "ASC", + SortDirection::Desc => "DESC", + }; + query + .push(" ORDER BY rollout_ordinal ") + .push(order) + .push(" LIMIT ") + .push_bind(limit); +} + +fn page_cursors( + thread_id: ThreadId, + scope: &CursorScope, + first_ordinal: Option, + last_ordinal: Option, + has_more: bool, +) -> ThreadStoreResult<(Option, Option)> { + let cursor = |rollout_ordinal, include_anchor| { + serialize_cursor(thread_id, scope, rollout_ordinal, include_anchor) + }; + let backwards_cursor = first_ordinal + .map(|rollout_ordinal| cursor(rollout_ordinal, /*include_anchor*/ true)) + .transpose()?; + let next_cursor = if has_more { + last_ordinal + .map(|rollout_ordinal| cursor(rollout_ordinal, /*include_anchor*/ false)) + .transpose()? + } else { + None + }; + Ok((next_cursor, backwards_cursor)) +} + +fn serialize_cursor( + thread_id: ThreadId, + scope: &CursorScope, + rollout_ordinal: i64, + include_anchor: bool, +) -> ThreadStoreResult { + serde_json::to_string(&HistoryCursor { + thread_id, + scope: scope.clone(), + rollout_ordinal, + include_anchor, + }) + .map_err(thread_history_error) +} + +fn invalid_cursor(cursor: &str) -> ThreadStoreError { + ThreadStoreError::InvalidRequest { + message: format!("invalid cursor: {cursor}"), + } +} + +fn stored_turn_row(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult { + let status = match row.try_get::("status")?.as_str() { + "completed" => StoredTurnStatus::Completed, + "interrupted" => StoredTurnStatus::Interrupted, + "failed" => StoredTurnStatus::Failed, + "inProgress" => StoredTurnStatus::InProgress, + status => { + return Err(ThreadStoreError::Internal { + message: format!("unknown stored turn status: {status}"), + }); + } + }; + let error_json = row.try_get::, _>("error_json")?; + let error = error_json + .as_deref() + .map(serde_json::from_str) + .transpose() + .map_err(thread_history_error)?; + Ok(StoredTurnRow { + turn_id: row.try_get("turn_id")?, + rollout_ordinal: row.try_get("rollout_ordinal")?, + status, + error, + started_at: row.try_get("started_at")?, + completed_at: row.try_get("completed_at")?, + duration_ms: row.try_get("duration_ms")?, + first_user_item_id: row.try_get("first_user_item_id")?, + final_agent_item_id: row.try_get("final_agent_item_id")?, + }) +} + +async fn load_summary_items( + pool: &sqlx::SqlitePool, + thread_id: ThreadId, + turn: &StoredTurnRow, +) -> ThreadStoreResult> { + let rows = sqlx::query( + r#" +SELECT turn_id, item_id, rollout_ordinal, created_at_ms, item_json +FROM thread_items +WHERE thread_id = ? + AND turn_id = ? + AND (item_id = ? OR item_id = ?) +ORDER BY rollout_ordinal ASC + "#, + ) + .bind(thread_id.to_string()) + .bind(turn.turn_id.as_str()) + .bind(turn.first_user_item_id.as_deref()) + .bind(turn.final_agent_item_id.as_deref()) + .fetch_all(pool) + .await + .map_err(thread_history_error)?; + rows.into_iter() + .map(|row| stored_thread_item_row(row).map(|row| row.item)) + .collect() +} + +fn stored_thread_item_row(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult { + let rollout_ordinal = row.try_get::("rollout_ordinal")?; + if rollout_ordinal < 0 { + return Err(ThreadStoreError::Internal { + message: format!("invalid stored item rollout ordinal: {rollout_ordinal}"), + }); + } + Ok(StoredThreadItemRow { + item: StoredThreadItem { + turn_id: row.try_get("turn_id")?, + item_id: row.try_get("item_id")?, + created_at_ms: row.try_get("created_at_ms")?, + item_json: row.try_get::("item_json")?.into_bytes(), + }, + rollout_ordinal, + }) +} diff --git a/codex-rs/thread-store/src/local/thread_history/read_tests.rs b/codex-rs/thread-store/src/local/thread_history/read_tests.rs new file mode 100644 index 000000000000..cfef95313d05 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -0,0 +1,384 @@ +use chrono::Utc; +use codex_app_server_protocol::CodexErrorInfo; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +use super::*; +use crate::local::test_support::test_config; + +#[tokio::test] +async fn list_turns_pages_projected_rows_and_applies_item_views() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let db = history_db(&store).await; + for (turn_id, ordinal, status, error, first_user, final_agent) in [ + ( + "turn-1", + 10, + "completed", + None, + Some("user-1"), + Some("agent-1"), + ), + ( + "turn-2", + 20, + "failed", + Some( + r#"{"message":"turn failed","codexErrorInfo":"serverOverloaded","additionalDetails":"retry later"}"#, + ), + None, + None, + ), + ("turn-3", 30, "inProgress", None, None, None), + ] { + insert_turn( + db, + thread_id, + turn_id, + ordinal, + status, + error, + first_user, + final_agent, + ) + .await; + } + for (turn_id, item_id, ordinal) in [ + ("turn-1", "user-1", 11), + ("turn-1", "middle-1", 12), + ("turn-1", "agent-1", 13), + ] { + insert_item(db, thread_id, turn_id, item_id, ordinal).await; + } + + let first_page = store + .list_turns(turn_params( + thread_id, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + StoredTurnItemsView::Summary, + )) + .await + .expect("first turns page"); + assert_eq!(turn_ids(&first_page), vec!["turn-1", "turn-2"]); + assert_eq!( + first_page.turns[0].items, + vec![ + expected_item("turn-1", "user-1", /*rollout_ordinal*/ 11), + expected_item("turn-1", "agent-1", /*rollout_ordinal*/ 13), + ] + ); + assert_eq!( + first_page.turns[1].error, + Some(StoredTurnError { + message: "turn failed".to_string(), + codex_error_info: Some(CodexErrorInfo::ServerOverloaded), + additional_details: Some("retry later".to_string()), + }) + ); + let second_page = store + .list_turns(turn_params( + thread_id, + first_page.next_cursor, + /*page_size*/ 2, + SortDirection::Asc, + StoredTurnItemsView::NotLoaded, + )) + .await + .expect("second turns page"); + assert_eq!(turn_ids(&second_page), vec!["turn-3"]); + assert_eq!(second_page.turns[0].items, Vec::new()); + assert_eq!(second_page.turns[0].status, StoredTurnStatus::InProgress); + let backwards_page = store + .list_turns(turn_params( + thread_id, + second_page.backwards_cursor, + /*page_size*/ 2, + SortDirection::Desc, + StoredTurnItemsView::NotLoaded, + )) + .await + .expect("backwards turns page"); + assert_eq!(turn_ids(&backwards_page), vec!["turn-3", "turn-2"]); +} + +#[tokio::test] +async fn list_items_pages_whole_thread_and_per_turn_rows() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let db = history_db(&store).await; + for (turn_id, ordinal) in [("turn-1", 10), ("turn-2", 20)] { + insert_turn( + db, + thread_id, + turn_id, + ordinal, + "completed", + /*error_json*/ None, + /*first_user_item_id*/ None, + /*final_agent_item_id*/ None, + ) + .await; + } + for (turn_id, item_id, ordinal) in [ + ("turn-1", "item-1", 11), + ("turn-1", "item-2", 12), + ("turn-2", "item-3", 21), + ("turn-2", "item-4", 22), + ("turn-2", "item-5", 23), + ] { + insert_item(db, thread_id, turn_id, item_id, ordinal).await; + } + + let first_page = store + .list_items(item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + )) + .await + .expect("first item page"); + assert_eq!( + first_page.items, + vec![ + expected_item("turn-1", "item-1", /*rollout_ordinal*/ 11), + expected_item("turn-1", "item-2", /*rollout_ordinal*/ 12), + ] + ); + let second_page = store + .list_items(item_params( + thread_id, + /*turn_id*/ None, + first_page.next_cursor, + /*page_size*/ 2, + SortDirection::Asc, + )) + .await + .expect("second item page"); + assert_eq!(item_ids(&second_page), vec!["item-3", "item-4"]); + let backwards_page = store + .list_items(item_params( + thread_id, + /*turn_id*/ None, + second_page.backwards_cursor, + /*page_size*/ 2, + SortDirection::Desc, + )) + .await + .expect("backwards item page"); + assert_eq!(item_ids(&backwards_page), vec!["item-3", "item-2"]); + + let turn_page = store + .list_items(item_params( + thread_id, + Some("turn-2"), + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Desc, + )) + .await + .expect("turn item page"); + assert_eq!(item_ids(&turn_page), vec!["item-5", "item-4"]); + let next_turn_page = store + .list_items(item_params( + thread_id, + Some("turn-2"), + turn_page.next_cursor, + /*page_size*/ 2, + SortDirection::Desc, + )) + .await + .expect("next turn item page"); + assert_eq!(item_ids(&next_turn_page), vec!["item-3"]); +} + +#[tokio::test] +async fn list_history_keeps_legacy_threads_unsupported() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Legacy).await; + + let error = store + .list_turns(turn_params( + thread_id, + /*cursor*/ None, + /*page_size*/ 1, + SortDirection::Asc, + StoredTurnItemsView::Summary, + )) + .await + .expect_err("legacy turns remain unsupported"); + assert!(matches!( + error, + ThreadStoreError::Unsupported { + operation: "list_turns" + } + )); + + let error = store + .list_turns(turn_params( + ThreadId::default(), + /*cursor*/ None, + /*page_size*/ 1, + SortDirection::Asc, + StoredTurnItemsView::Summary, + )) + .await + .expect_err("unindexed threads remain unsupported"); + assert!(matches!( + error, + ThreadStoreError::Unsupported { + operation: "list_turns" + } + )); +} + +async fn store_with_mode(history_mode: ThreadHistoryMode) -> (TempDir, LocalThreadStore, ThreadId) { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let thread_id = ThreadId::default(); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state runtime"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + home.path().join("missing-rollout.jsonl"), + Utc::now(), + SessionSource::Cli, + ); + builder.history_mode = history_mode; + runtime + .upsert_thread(&builder.build(config.default_model_provider_id.as_str())) + .await + .expect("seed thread metadata"); + let store = LocalThreadStore::new(config, Some(runtime)); + (home, store, thread_id) +} + +async fn history_db(store: &LocalThreadStore) -> &sqlx::SqlitePool { + store + .thread_history_db() + .await + .expect("open history fixture database") +} + +#[allow(clippy::too_many_arguments)] +async fn insert_turn( + db: &sqlx::SqlitePool, + thread_id: ThreadId, + turn_id: &str, + rollout_ordinal: i64, + status: &str, + error_json: Option<&str>, + first_user_item_id: Option<&str>, + final_agent_item_id: Option<&str>, +) { + sqlx::query( + r#" +INSERT INTO thread_turns ( + thread_id, + turn_id, + rollout_ordinal, + status, + error_json, + first_user_item_id, + final_agent_item_id +) VALUES (?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(thread_id.to_string()) + .bind(turn_id) + .bind(rollout_ordinal) + .bind(status) + .bind(error_json) + .bind(first_user_item_id) + .bind(final_agent_item_id) + .execute(db) + .await + .expect("insert turn fixture"); +} + +async fn insert_item( + db: &sqlx::SqlitePool, + thread_id: ThreadId, + turn_id: &str, + item_id: &str, + rollout_ordinal: i64, +) { + sqlx::query( + "INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_json) VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(thread_id.to_string()) + .bind(turn_id) + .bind(item_id) + .bind(rollout_ordinal) + .bind(rollout_ordinal * 1_000) + .bind(format!(r#"{{"type":"userMessage","id":"{item_id}","content":[]}}"#)) + .execute(db) + .await + .expect("insert item fixture"); +} + +fn turn_params( + thread_id: ThreadId, + cursor: Option, + page_size: usize, + sort_direction: SortDirection, + items_view: StoredTurnItemsView, +) -> ListTurnsParams { + ListTurnsParams { + thread_id, + include_archived: false, + cursor, + page_size, + sort_direction, + items_view, + } +} + +fn item_params( + thread_id: ThreadId, + turn_id: Option<&str>, + cursor: Option, + page_size: usize, + sort_direction: SortDirection, +) -> ListItemsParams { + ListItemsParams { + thread_id, + turn_id: turn_id.map(str::to_owned), + include_archived: false, + cursor, + page_size, + sort_direction, + } +} + +fn expected_item(turn_id: &str, item_id: &str, rollout_ordinal: u64) -> StoredThreadItem { + StoredThreadItem { + turn_id: turn_id.to_string(), + item_id: item_id.to_string(), + created_at_ms: i64::try_from(rollout_ordinal).expect("fixture ordinal fits i64") * 1_000, + item_json: format!(r#"{{"type":"userMessage","id":"{item_id}","content":[]}}"#) + .into_bytes(), + } +} + +fn turn_ids(page: &TurnPage) -> Vec<&str> { + page.turns + .iter() + .map(|turn| turn.turn_id.as_str()) + .collect() +} + +fn item_ids(page: &ItemPage) -> Vec<&str> { + page.items + .iter() + .map(|item| item.item_id.as_str()) + .collect() +} diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs new file mode 100644 index 000000000000..380a962ea0b0 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -0,0 +1,108 @@ +use std::io::SeekFrom; +use std::path::Path; + +use chrono::DateTime; +use codex_app_server_protocol::project_rollout_line; +use codex_protocol::ThreadId; +use codex_protocol::protocol::RolloutLine; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncSeekExt; +use tracing::warn; + +use super::LocalThreadStore; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +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?; + if lines.is_empty() && start_offset == next_offset { + return Ok(()); + } + + let projections = lines + .iter() + .map(|line| { + let created_at_ms = DateTime::parse_from_rfc3339(line.timestamp.as_str()) + .map(|timestamp| timestamp.timestamp_millis()) + .map_err(thread_history_error)?; + Ok((line.ordinal, created_at_ms, project_rollout_line(line))) + }) + .collect::>>()?; + super::thread_history::apply_projection( + store, + thread_id, + start_offset, + next_offset, + projections, + ) + .await +} + +async fn read_durable_rollout_lines( + rollout_path: &Path, + start_offset: u64, +) -> ThreadStoreResult<(Vec, u64)> { + let next_offset = tokio::fs::metadata(rollout_path) + .await + .map_err(thread_store_io_error)? + .len(); + let byte_count = + next_offset + .checked_sub(start_offset) + .ok_or_else(|| ThreadStoreError::Internal { + message: "durable rollout shrank before projection".to_string(), + })?; + let byte_count = usize::try_from(byte_count).map_err(|_| ThreadStoreError::Internal { + message: "durable rollout append exceeds addressable memory".to_string(), + })?; + let mut bytes = vec![0; byte_count]; + let mut file = tokio::fs::File::open(rollout_path) + .await + .map_err(thread_store_io_error)?; + file.seek(SeekFrom::Start(start_offset)) + .await + .map_err(thread_store_io_error)?; + 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 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}"); + } + } + } + Ok((lines, next_offset)) +} + +fn thread_history_error(err: impl std::fmt::Display) -> ThreadStoreError { + ThreadStoreError::Internal { + message: format!("failed to project thread history: {err}"), + } +} + +fn thread_store_io_error(err: std::io::Error) -> ThreadStoreError { + ThreadStoreError::Internal { + message: err.to_string(), + } +} + +#[cfg(test)] +#[path = "thread_history_materialization_tests.rs"] +mod tests; 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 new file mode 100644 index 000000000000..6b43f60bf6be --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -0,0 +1,560 @@ +use std::fs; +use std::io::Write; +use std::time::Duration; + +use codex_app_server_protocol::ThreadItem; +use codex_protocol::ThreadId; +use codex_protocol::items::AgentMessageContent; +use codex_protocol::items::AgentMessageItem; +use codex_protocol::items::TurnItem; +use codex_protocol::items::UserMessageItem; +use codex_protocol::models::BaseInstructions; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ItemCompletedEvent; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; +use codex_protocol::protocol::ThreadMemoryMode; +use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnStartedEvent; +use codex_rollout::RolloutRecorder; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +use super::super::LocalThreadStore; +use super::super::test_support::test_config; +use crate::AppendThreadItemsParams; +use crate::CreateThreadParams; +use crate::DeleteThreadParams; +use crate::ThreadPersistenceMetadata; +use crate::ThreadStore; + +#[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 thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + turn_started("turn-1"), + completed_item( + thread_id, + "turn-1", + TurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + completed_item( + thread_id, + "turn-1", + TurnItem::AgentMessage(AgentMessageItem { + id: "agent-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "done".to_string(), + }], + phase: None, + memory_citation: None, + }), + ), + turn_completed("turn-1"), + ], + }) + .await + .expect("append paginated items"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let turn = sqlx::query_as::< + _, + ( + i64, + String, + Option, + Option, + Option, + Option, + Option, + ), + >( + r#" +SELECT + rollout_ordinal, + status, + started_at, + completed_at, + duration_ms, + first_user_item_id, + final_agent_item_id +FROM thread_turns +WHERE thread_id = ? AND turn_id = ? + "#, + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .fetch_one(&pool) + .await + .expect("read projected turn"); + assert_eq!( + turn, + ( + 1, + "completed".to_string(), + Some(10), + Some(20), + Some(10_000), + Some("user-1".to_string()), + Some("agent-1".to_string()), + ) + ); + + let items = sqlx::query_as::<_, (String, i64)>( + r#" +SELECT item_id, rollout_ordinal +FROM thread_items +WHERE thread_id = ? +ORDER BY rollout_ordinal + "#, + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected items"); + assert_eq!( + items, + vec![("user-1".to_string(), 2), ("agent-1".to_string(), 3)] + ); + + 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"); + let projection_state = 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"); + assert_eq!(projection_state, (rollout_len, 5)); +} + +#[tokio::test] +async fn replayed_item_snapshot_updates_content_without_reordering() { + 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"), + completed_item( + thread_id, + "turn-1", + TurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + ], + }) + .await + .expect("append first item snapshot"); + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let first_created_at_ms = sqlx::query_scalar::<_, i64>( + "SELECT created_at_ms FROM thread_items WHERE thread_id = ? AND turn_id = ? AND item_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .bind("user-1") + .fetch_one(&pool) + .await + .expect("read first item timestamp"); + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![completed_item( + thread_id, + "turn-1", + TurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: Some("updated".to_string()), + content: Vec::new(), + }), + )], + }) + .await + .expect("append replayed item snapshot"); + + let item = sqlx::query_as::<_, (i64, i64, String)>( + r#" +SELECT rollout_ordinal, created_at_ms, item_json +FROM thread_items +WHERE thread_id = ? AND turn_id = ? AND item_id = ? + "#, + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .bind("user-1") + .fetch_one(&pool) + .await + .expect("read projected item"); + assert_eq!(item.0, 2); + assert_eq!(item.1, first_created_at_ms); + assert_eq!( + serde_json::from_str::(item.2.as_str()).expect("parse projected item"), + ThreadItem::UserMessage { + id: "user-1".to_string(), + client_id: Some("updated".to_string()), + content: Vec::new(), + } + ); +} + +#[tokio::test] +async fn turn_creation_recovers_summary_ids_from_earlier_items() { + 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![ + completed_item( + thread_id, + "turn-1", + TurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + completed_item( + thread_id, + "turn-1", + TurnItem::AgentMessage(AgentMessageItem { + id: "agent-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "done".to_string(), + }], + phase: None, + memory_citation: None, + }), + ), + ], + }) + .await + .expect("append items before turn"); + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![turn_started("turn-1"), turn_completed("turn-1")], + }) + .await + .expect("append turn lifecycle"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let summary_ids = sqlx::query_as::<_, (Option, Option)>( + "SELECT first_user_item_id, final_agent_item_id FROM thread_turns WHERE thread_id = ? AND turn_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .fetch_one(&pool) + .await + .expect("read turn summary ids"); + assert_eq!( + summary_ids, + (Some("user-1".to_string()), Some("agent-1".to_string())) + ); +} + +#[tokio::test] +async fn jsonl_failure_does_not_create_projection_database() { + let home = TempDir::new().expect("temp dir"); + fs::write(home.path().join("sessions"), "not a directory").expect("block sessions 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_err("JSONL append should fail"); + + assert!(!codex_state::thread_history_db_path(home.path()).exists()); +} + +#[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_home = sqlite_home; + let store = LocalThreadStore::new(config, /*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("durable JSONL append should succeed"); + + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("rollout path"); + let (items, _, _) = RolloutRecorder::load_rollout_items(rollout_path.as_path()) + .await + .expect("load durable rollout"); + assert!(items.iter().any(|item| { + matches!( + item, + RolloutItem::EventMsg(EventMsg::TurnStarted(event)) + if event.turn_id == "turn-1" + ) + })); +} + +#[tokio::test] +async fn rejected_rollout_line_does_not_poison_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"); + 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()) + .expect("open rollout for rejected line"); + file.write_all(b"{not json}\n") + .expect("append rejected line"); + file.flush().expect("flush rejected line"); + let recorder = store + .live_recorders + .lock() + .await + .get(&thread_id) + .expect("live recorder") + .recorder + .clone(); + recorder + .record_canonical_items(&[turn_started("turn-1")]) + .await + .expect("queue valid retry"); + recorder.flush().await.expect("flush valid retry"); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path(), start_offset) + .await + .expect("project valid retry after rejected line"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let projected_turns = sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM thread_turns WHERE thread_id = ? AND turn_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .fetch_one(&pool) + .await + .expect("read projected turns"); + assert_eq!(projected_turns, 1); +} + +#[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 thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + let recorder = store + .live_recorders + .lock() + .await + .get(&thread_id) + .expect("live recorder") + .recorder + .clone(); + recorder + .record_canonical_items(&[turn_started("turn-1")]) + .await + .expect("queue rollout item"); + + store + .shutdown_thread(thread_id) + .await + .expect("shutdown live thread"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let projected_turns = sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM thread_turns WHERE thread_id = ? AND turn_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .fetch_one(&pool) + .await + .expect("read projected turns"); + assert_eq!(projected_turns, 1); +} + +#[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 thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + let write_permit = store.live_writer_locks.lock(thread_id).await; + + let append_store = store.clone(); + let append = tokio::spawn(async move { + append_store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![turn_started("turn-1")], + }) + .await + }); + tokio::time::sleep(Duration::from_millis(/*millis*/ 10)).await; + let delete_store = store.clone(); + let delete = tokio::spawn(async move { + delete_store + .delete_thread(DeleteThreadParams { thread_id }) + .await + }); + tokio::time::sleep(Duration::from_millis(/*millis*/ 10)).await; + assert!(!delete.is_finished()); + + drop(write_permit); + append + .await + .expect("join append") + .expect("finish in-flight append"); + delete.await.expect("join delete").expect("delete thread"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + 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.to_string()) + .bind(thread_id.to_string()) + .bind(thread_id.to_string()) + .fetch_one(&pool) + .await + .expect("read history row counts"); + assert_eq!(counts, (0, 0, 0)); +} + +async fn create_paginated_thread(store: &LocalThreadStore, thread_id: ThreadId) { + store + .create_thread(CreateThreadParams { + session_id: thread_id.into(), + thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: SessionSource::Exec, + thread_source: None, + originator: "test_originator".to_string(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + selected_capability_roots: Vec::new(), + multi_agent_version: None, + history_mode: ThreadHistoryMode::Paginated, + initial_window_id: "window-1".to_string(), + metadata: ThreadPersistenceMetadata { + cwd: Some(std::env::current_dir().expect("cwd")), + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await + .expect("create paginated thread"); +} + +fn turn_started(turn_id: &str) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn_id.to_string(), + trace_id: None, + started_at: Some(10), + model_context_window: None, + collaboration_mode_kind: Default::default(), + })) +} + +fn turn_completed(turn_id: &str) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn_id.to_string(), + last_agent_message: None, + error: None, + started_at: Some(10), + completed_at: Some(20), + duration_ms: Some(10_000), + time_to_first_token_ms: None, + })) +} + +fn completed_item(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id, + turn_id: turn_id.to_string(), + item, + completed_at_ms: 1, + })) +} diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 74f21872eff2..7483703bae20 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use chrono::DateTime; use chrono::Utc; +use codex_app_server_protocol::CodexErrorInfo; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -305,8 +306,6 @@ pub enum StoredTurnItemsView { /// Return display summary items for each turn. #[default] Summary, - /// Return every persisted item available for each turn. - Full, } /// Store-owned status for a persisted turn. @@ -324,9 +323,12 @@ pub enum StoredTurnStatus { /// Store-owned error details for a failed persisted turn. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub struct StoredTurnError { /// User-visible error message. pub message: String, + /// Structured Codex error classification, when available. + pub codex_error_info: Option, /// Optional additional detail for clients that expose expanded error context. pub additional_details: Option, } @@ -353,13 +355,8 @@ pub struct ListTurnsParams { pub struct StoredTurn { /// Turn id. pub turn_id: String, - /// Persisted rollout items associated with this turn, according to `items_view`. - pub items: Vec, - /// Opaque serialized turn metadata supplied by a projected durable store. - pub metadata_json: Option>, - /// Semantic turn creation timestamp in milliseconds, when supplied by a projected durable - /// store. - pub turn_created_at_ms: Option, + /// Projected app-server item snapshots associated with this turn, according to `items_view`. + pub items: Vec, /// Amount of item detail included in `items`. pub items_view: StoredTurnItemsView, /// Store-owned status for API layer projection. @@ -405,11 +402,14 @@ pub struct ListItemsParams { /// A projected app-server `ThreadItem` snapshot within a turn. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct StoredThreadItem { - pub turn_id: Option, - pub item_key: String, - pub item_ordinal: u64, - pub item_created_at_ms: i64, - pub materialized_thread_item_json: Vec, + /// Turn containing this item. + pub turn_id: String, + /// Stable item identifier within the turn. + pub item_id: String, + /// Unix timestamp (milliseconds) when this logical item was first projected. + pub created_at_ms: i64, + /// Serialized app-server ThreadItem snapshot. + pub item_json: Vec, } /// A page of persisted items within a thread, optionally filtered to a turn.