openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Materialize paginated thread history in SQLite (#32923)

Browse Source 
## What changed

- Project durable paginated rollout records into rebuildable SQLite tables for turns, items, and projection progress while keeping JSONL as the source of truth.
- Add cursor-based local reads for turn summaries and thread or per-turn items, including archived-thread and history-mode validation.
- Serialize writes, shutdown, and deletion per thread so projection updates cannot race cleanup, and remove projected rows when a thread is deleted.
- Preserve stable item ordering while updating repeated snapshots and expose structured turn error details.

## Testing

- Cover projection, snapshot updates, summary recovery, pagination, failure handling, shutdown flushing, legacy-mode rejection, and deletion races.

GitOrigin-RevId: 34533185ece7f1f5690004c7f12b0f1f34dc2640
This commit is contained in:
Owen Lin
2026-07-14 02:00:39 +00:00
committed by copyberry
parent f96cf4d1c3
commit 0ef9fa4d65
15 changed files with 2104 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -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",
codex-rs/app-server/src/request_processors/thread_processor.rs
+11 -12
View File
@@ -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::<ThreadItem>(&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::<ThreadItem>(&item.item_json).map_err(|err| {
internal_error(format!(
"failed to deserialize stored thread item {}: {err}",
item.item_id
))
})
.collect::<Result<Vec<_>, _>>()?;
})
.collect::<Result<Vec<_>, _>>()?;
Ok(ThreadItemsListResponse {
data,
codex-rs/state/src/lib.rs
+1
View File
@@ -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;
codex-rs/state/src/runtime.rs
+13
View File
@@ -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<SqlitePool> {
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,
codex-rs/state/thread_history_migrations/0001_thread_history.sql
+1
View File
@@ -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)
);
codex-rs/thread-store/Cargo.toml
+2
View File
@@ -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 }
codex-rs/thread-store/src/local/delete_thread.rs
+72 -7
View File
@@ -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");
codex-rs/thread-store/src/local/live_writer.rs
+130 -43
View File
@@ -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(&params.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<RolloutItem>),
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<u64> {
match tokio::fs::metadata(rollout_path).await {
Ok(metadata) => Ok(metadata.len()),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(err) => Err(thread_store_io_error(err)),
}
}
codex-rs/thread-store/src/local/mod.rs
+53 -12
View File
@@ -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<Mutex<HashMap<ThreadId, LiveRecorderEntry>>>,
live_writer_locks: Arc<LiveWriterLocks>,
state_db: Option<StateDbHandle>,
thread_history_db: Arc<OnceCell<sqlx::SqlitePool>>,
}
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<HashMap<ThreadId, Arc<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<RolloutRecorder> {
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<TurnPage> {
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<ItemPage> {
thread_history::list_items(self, params).await
}
}
impl ThreadStore for LocalThreadStore {
codex-rs/thread-store/src/local/thread_history.rs
+347
View File
@@ -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<u64>, 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> {
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<sqlx::Error> 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}"),
}
}
codex-rs/thread-store/src/local/thread_history/read.rs
+406
View File
@@ -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<String> },
}
struct StoredTurnRow {
turn_id: String,
rollout_ordinal: i64,
status: StoredTurnStatus,
error: Option<StoredTurnError>,
started_at: Option<i64>,
completed_at: Option<i64>,
duration_ms: Option<i64>,
first_user_item_id: Option<String>,
final_agent_item_id: Option<String>,
}
struct StoredThreadItemRow {
item: StoredThreadItem,
rollout_ordinal: i64,
}
pub(in crate::local) async fn list_turns(
store: &LocalThreadStore,
params: ListTurnsParams,
) -> ThreadStoreResult<TurnPage> {
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::<Sqlite>::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::<ThreadStoreResult<Vec<_>>>()?;
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<ItemPage> {
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::<Sqlite>::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::<ThreadStoreResult<Vec<_>>>()?;
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<i64> {
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<Option<HistoryCursor>> {
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<Sqlite>,
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<i64>,
last_ordinal: Option<i64>,
has_more: bool,
) -> ThreadStoreResult<(Option<String>, Option<String>)> {
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<String> {
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<StoredTurnRow> {
let status = match row.try_get::<String, _>("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::<Option<String>, _>("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<Vec<StoredThreadItem>> {
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<StoredThreadItemRow> {
let rollout_ordinal = row.try_get::<i64, _>("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::<String, _>("item_json")?.into_bytes(),
},
rollout_ordinal,
})
}
codex-rs/thread-store/src/local/thread_history/read_tests.rs
+384
View File
@@ -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<String>,
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<String>,
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()
}
codex-rs/thread-store/src/local/thread_history_materialization.rs
+108
View File
@@ -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::<ThreadStoreResult<Vec<_>>>()?;
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<RolloutLine>, 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;
codex-rs/thread-store/src/local/thread_history_materialization_tests.rs
+560
View File
@@ -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<i64>,
Option<i64>,
Option<i64>,
Option<String>,
Option<String>,
),
>(
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::<ThreadItem>(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<String>, Option<String>)>(
"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,
}))
}
codex-rs/thread-store/src/types.rs
+14 -14
View File
@@ -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<CodexErrorInfo>,
/// Optional additional detail for clients that expose expanded error context.
pub additional_details: Option<String>,
}
@@ -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<RolloutItem>,
/// Opaque serialized turn metadata supplied by a projected durable store.
pub metadata_json: Option<Vec<u8>>,
/// Semantic turn creation timestamp in milliseconds, when supplied by a projected durable
/// store.
pub turn_created_at_ms: Option<i64>,
/// Projected app-server item snapshots associated with this turn, according to `items_view`.
pub items: Vec<StoredThreadItem>,
/// 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<String>,
pub item_key: String,
pub item_ordinal: u64,
pub item_created_at_ms: i64,
pub materialized_thread_item_json: Vec<u8>,
/// 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<u8>,
}
/// A page of persisted items within a thread, optionally filtered to a turn.