From b834702b27b3403403f5d1cd8992960a1a6211a5 Mon Sep 17 00:00:00 2001 From: richardopenai Date: Thu, 23 Jul 2026 21:13:11 +0000 Subject: [PATCH] Support incremental replay of updated thread items (#35013) ## Why Persisted item snapshots can be updated after their initial projection while retaining their creation order. Callers need a way to read those updates incrementally without replaying every item. ## What changed - Track the rollout ordinal of each item's latest snapshot separately from its creation ordinal. - Let thread-store callers filter items after an exclusive update watermark and sort pages by either creation or update ordinal. - Keep existing app-server reads creation-ordered, reject incremental reads across forked lineages, and preserve compatibility with older writers. ## Testing - Cover update-watermark filtering, pagination, sort directions, per-turn reads, cursor validation, forked lineages, migration compatibility, and refreshed item snapshots. GitOrigin-RevId: d57de849abec27f788e9c384dbff073a0aca1053 --- codex-rs/app-server/src/request_processors.rs | 1 + .../request_processors/thread_processor.rs | 6 + .../app-server/tests/suite/v2/thread_read.rs | 11 +- codex-rs/state/src/migrations_tests.rs | 69 +++++++ .../0004_thread_items_updated_at_ordinal.sql | 15 ++ codex-rs/thread-store/src/in_memory.rs | 3 + codex-rs/thread-store/src/lib.rs | 1 + .../thread-store/src/local/thread_history.rs | 9 +- .../src/local/thread_history/read.rs | 22 +-- .../src/local/thread_history/read_tests.rs | 185 +++++++++++++++++- .../local/thread_history/segment_paging.rs | 123 ++++++++++-- .../thread_history_materialization_tests.rs | 11 +- codex-rs/thread-store/src/types.rs | 17 +- 13 files changed, 436 insertions(+), 37 deletions(-) create mode 100644 codex-rs/state/thread_history_migrations/0004_thread_items_updated_at_ordinal.sql diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 25a422799b..ab48f4daab 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -467,6 +467,7 @@ use codex_state::log_db::LogDbLayer; use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; use codex_thread_store::DeleteThreadsParams as StoreDeleteThreadsParams; use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; +use codex_thread_store::ItemSortKey as StoreItemSortKey; use codex_thread_store::ListItemsParams as StoreListItemsParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::ListTurnsParams as StoreListTurnsParams; 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 8861a9f049..b77278fb75 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2715,6 +2715,8 @@ impl ThreadRequestProcessor { cursor: cursor.clone(), page_size: THREAD_ITEMS_MAX_LIMIT, sort_direction: StoreSortDirection::Asc, + sort_key: StoreItemSortKey::CreatedAtOrdinal, + after_updated_at_ordinal: None, }) .await .map_err(paginated_history_list_error)?; @@ -2829,6 +2831,8 @@ impl ThreadRequestProcessor { cursor: None, page_size: 1, sort_direction: StoreSortDirection::Desc, + sort_key: StoreItemSortKey::CreatedAtOrdinal, + after_updated_at_ordinal: None, }) .await .map_err(paginated_history_list_error)?; @@ -2864,6 +2868,8 @@ impl ThreadRequestProcessor { SortDirection::Asc => StoreSortDirection::Asc, SortDirection::Desc => StoreSortDirection::Desc, }, + sort_key: StoreItemSortKey::CreatedAtOrdinal, + after_updated_at_ordinal: None, }) .await .map_err(|err| match err { diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index b3af8ccb3b..395fb46f0f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1518,6 +1518,15 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { memory_citation: None, }), ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::UserMessage(UserMessageItem { + id: "steer-1".to_string(), + client_id: Some("updated-steer".to_string()), + content: Vec::new(), + }), + ), paginated_turn_completed("turn-1"), paginated_turn_started("turn-2"), paginated_completed_item( @@ -1550,7 +1559,7 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { }, ThreadItem::UserMessage { id: "steer-1".to_string(), - client_id: None, + client_id: Some("updated-steer".to_string()), content: Vec::new(), }, ThreadItem::AgentMessage { diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 8ec2102d22..d5d836fdcd 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -6,6 +6,7 @@ use sqlx::migrate::Migrator; use std::borrow::Cow; use super::STATE_MIGRATOR; +use super::THREAD_HISTORY_MIGRATOR; use super::repair_legacy_recency_migration_version; fn migrator_through(version: i64) -> Migrator { @@ -100,6 +101,74 @@ INSERT INTO threads ( pool.close().await; } +#[tokio::test] +async fn thread_item_update_ordinals_allow_older_writers() { + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home) + .await + .expect("sqlite home should be created"); + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); + let pre_update_ordinal_migrator = Migrator { + migrations: Cow::Owned( + THREAD_HISTORY_MIGRATOR + .migrations + .iter() + .filter(|migration| migration.version < 4) + .cloned() + .collect(), + ), + ignore_missing: THREAD_HISTORY_MIGRATOR.ignore_missing, + locking: THREAD_HISTORY_MIGRATOR.locking, + table_name: THREAD_HISTORY_MIGRATOR.table_name.clone(), + create_schemas: THREAD_HISTORY_MIGRATOR.create_schemas.clone(), + no_tx: THREAD_HISTORY_MIGRATOR.no_tx, + }; + let pool = sqlite + .open_thread_history_db( + &pre_update_ordinal_migrator, + /*telemetry_override*/ None, + ) + .await + .expect("pre-update-ordinal migrations should apply"); + sqlx::query( + r#" +INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_type, item_json) VALUES + ('thread-1', 'turn-1', 'existing-item-1', 11, 1_100, 'userMessage', '{}'), + ('thread-1', 'turn-1', 'existing-item-2', 12, 1_200, 'userMessage', '{}') + "#, + ) + .execute(&pool) + .await + .expect("pre-migration items should be inserted"); + THREAD_HISTORY_MIGRATOR + .run(&pool) + .await + .expect("update-ordinal migration should apply"); + sqlx::query( + r#" +INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_type, item_json) VALUES + ('thread-1', 'turn-1', 'old-writer-item-1', 13, 1_300, 'userMessage', '{}'), + ('thread-1', 'turn-1', 'old-writer-item-2', 14, 1_400, 'userMessage', '{}') + "#, + ) + .execute(&pool) + .await + .expect("older writers should be able to append multiple items after migration"); + let ordinals = sqlx::query_as::<_, (i64, i64)>( + "SELECT rollout_ordinal, updated_at_ordinal FROM thread_items WHERE thread_id = ? ORDER BY rollout_ordinal", + ) + .bind("thread-1") + .fetch_all(&pool) + .await + .expect("old-writer items should load"); + assert_eq!(ordinals, vec![(11, 11), (12, 12), (13, 0), (14, 0)]); + + pool.close().await; +} + #[tokio::test] async fn agent_job_tables_are_dropped_when_upgrading() { let sqlite_home = crate::runtime::test_support::unique_temp_dir(); diff --git a/codex-rs/state/thread_history_migrations/0004_thread_items_updated_at_ordinal.sql b/codex-rs/state/thread_history_migrations/0004_thread_items_updated_at_ordinal.sql new file mode 100644 index 0000000000..30629325f8 --- /dev/null +++ b/codex-rs/state/thread_history_migrations/0004_thread_items_updated_at_ordinal.sql @@ -0,0 +1,15 @@ +ALTER TABLE thread_items ADD COLUMN updated_at_ordinal INTEGER NOT NULL DEFAULT 0; + +-- As of this migration, existing projected items originate from exactly one ItemCompleted rollout +-- event, so their creation ordinal is also their update ordinal. +UPDATE thread_items +SET updated_at_ordinal = rollout_ordinal; + +-- Older writers can still append items with the zero default after this migration. Incremental +-- replay excludes those items until a newer writer projects an update for them. +-- Keep this index non-unique so mixed-version writers can continue to persist history. +CREATE INDEX idx_thread_items_updated_page + ON thread_items(thread_id, updated_at_ordinal); + +CREATE INDEX idx_thread_items_by_turn_updated_page + ON thread_items(thread_id, turn_id, updated_at_ordinal); diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 83b9e6a194..b1da2b85d4 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -50,6 +50,7 @@ fn stores() -> &'static Mutex>> { #[cfg(test)] mod tests { use super::*; + use crate::ItemSortKey; use crate::ListItemsParams; use crate::ListTurnsParams; use crate::SortDirection; @@ -90,6 +91,8 @@ mod tests { cursor: None, page_size: 10, sort_direction: SortDirection::Asc, + sort_key: ItemSortKey::CreatedAtOrdinal, + after_updated_at_ordinal: None, }) .await .expect_err("default list_items should be unsupported"); diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 878b0df0e9..18b7312985 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -31,6 +31,7 @@ pub use types::DeleteThreadsParams; pub use types::ExtraConfig; pub use types::GitInfoPatch; pub use types::ItemPage; +pub use types::ItemSortKey; pub use types::ListItemsParams; pub use types::ListThreadsParams; pub use types::ListTurnsParams; diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index 616c627969..33d8f5c42c 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -312,8 +312,8 @@ WHERE thread_id = ? 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. + // creation ordinal and timestamp from the first record so item identity and age stay + // stable. Track the latest snapshot separately for incremental replay. sqlx::query( r#" INSERT INTO thread_items ( @@ -321,11 +321,13 @@ INSERT INTO thread_items ( turn_id, item_id, rollout_ordinal, + updated_at_ordinal, created_at_ms, item_type, item_json -) VALUES (?, ?, ?, ?, ?, json_extract(?, '$.type'), ?) +) VALUES (?, ?, ?, ?, ?, ?, json_extract(?, '$.type'), ?) ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET + updated_at_ordinal = excluded.updated_at_ordinal, item_type = excluded.item_type, item_json = excluded.item_json "#, @@ -334,6 +336,7 @@ ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET .bind(item.turn_id.as_str()) .bind(item_id.as_str()) .bind(rollout_ordinal) + .bind(rollout_ordinal) .bind(created_at_ms) .bind(item_json.as_str()) .bind(item_json) diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs index 94e28eba3f..fb9c1c590c 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -28,7 +28,8 @@ mod tests; #[serde(tag = "kind", rename_all = "camelCase")] pub(super) enum CursorScope { Turns, - Items, + ItemsByCreatedAtOrdinal, + ItemsByUpdatedAtOrdinal, } #[derive(Deserialize, Serialize)] @@ -126,16 +127,7 @@ pub(in crate::local) async fn list_items( validate_page_size(params.page_size)?; let lineage = store.resolve_rollout_lineage(params.thread_id).await?; let pool = store.thread_history_db().await?; - let page = page_item_rows( - pool, - params.thread_id, - &lineage, - params.turn_id.as_deref(), - params.cursor.as_deref(), - params.page_size, - params.sort_direction, - ) - .await?; + let page = page_item_rows(pool, &lineage, ¶ms).await?; Ok(ItemPage { items: page.rows.into_iter().map(|row| row.item).collect(), @@ -180,7 +172,7 @@ async fn load_summary_items( ) -> ThreadStoreResult> { let rows = sqlx::query( r#" -SELECT turn_id, item_id, created_at_ms, item_json +SELECT turn_id, item_id, updated_at_ordinal, created_at_ms, item_json FROM thread_items WHERE thread_id = ? AND turn_id = ? @@ -289,9 +281,15 @@ pub(super) fn stored_thread_item_row_for_thread( } fn stored_thread_item(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult { + let updated_at_ordinal = row.try_get::("updated_at_ordinal")?; + let updated_at_ordinal = + u64::try_from(updated_at_ordinal).map_err(|_| ThreadStoreError::Internal { + message: format!("invalid stored item updated-at ordinal: {updated_at_ordinal}"), + })?; Ok(StoredThreadItem { turn_id: row.try_get("turn_id")?, item_id: row.try_get("item_id")?, + updated_at_ordinal, created_at_ms: row.try_get("created_at_ms")?, item_json: row.try_get::("item_json")?.into_bytes(), }) 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 index 1158362e5b..3518bfc229 100644 --- a/codex-rs/thread-store/src/local/thread_history/read_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -15,6 +15,7 @@ use pretty_assertions::assert_eq; use tempfile::TempDir; use super::*; +use crate::ItemSortKey; use crate::SortDirection; use crate::StoredTurnError; use crate::StoredTurnStatus; @@ -222,6 +223,159 @@ async fn list_items_pages_whole_thread_and_per_turn_rows() { assert_eq!(item_ids(&next_turn_page), vec!["item-3"]); } +#[tokio::test] +async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let db = history_db(&store).await; + for (turn_id, item_id, ordinal) in [ + ("turn-1", "item-1", 1), + ("turn-2", "item-2", 2), + ("turn-2", "item-3", 3), + ] { + insert_item(db, thread_id, turn_id, item_id, ordinal).await; + } + sqlx::query( + "UPDATE thread_items SET updated_at_ordinal = 4 WHERE thread_id = ? AND item_id = 'item-1'", + ) + .bind(thread_id.to_string()) + .execute(db) + .await + .expect("advance first item update ordinal"); + + let item_1 = StoredThreadItem { + updated_at_ordinal: 4, + ..expected_item("turn-1", "item-1", /*rollout_ordinal*/ 1) + }; + let item_2 = expected_item("turn-2", "item-2", /*rollout_ordinal*/ 2); + let item_3 = expected_item("turn-2", "item-3", /*rollout_ordinal*/ 3); + let creation_page = store + .list_items(item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + )) + .await + .expect("creation-ordered item page"); + assert_eq!(creation_page.items, vec![item_1.clone(), item_2.clone()]); + for (sort_direction, expected) in [ + (SortDirection::Asc, vec![item_1.clone(), item_3.clone()]), + (SortDirection::Desc, vec![item_3.clone(), item_1.clone()]), + ] { + let page = store + .list_items(ListItemsParams { + after_updated_at_ordinal: Some(2), + ..item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + sort_direction, + ) + }) + .await + .expect("creation-ordered filtered item page"); + assert_eq!(page.items, expected); + } + + let first_page = store + .list_items(updated_item_params( + thread_id, /*after_updated_at_ordinal*/ 0, + )) + .await + .expect("first filtered item page"); + assert_eq!(first_page.items, vec![item_2.clone(), item_3.clone()]); + for params in [ + ListItemsParams { + cursor: creation_page.next_cursor, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }, + item_params( + thread_id, + /*turn_id*/ None, + first_page.next_cursor.clone(), + /*page_size*/ 2, + SortDirection::Asc, + ), + ] { + let error = store + .list_items(params) + .await + .expect_err("creation and update cursors should not be interchangeable"); + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); + } + + let second_page = store + .list_items(ListItemsParams { + cursor: first_page.next_cursor, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect("second filtered item page"); + assert_eq!(second_page.items, vec![item_1.clone()]); + assert!(second_page.next_cursor.is_none()); + + let exclusive_page = store + .list_items(ListItemsParams { + turn_id: Some("turn-2".to_string()), + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 2) + }) + .await + .expect("exclusive filtered turn page"); + assert_eq!(exclusive_page.items, vec![item_3.clone()]); + + let descending_page = store + .list_items(ListItemsParams { + sort_direction: SortDirection::Desc, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect("descending update-ordered item page"); + assert_eq!(descending_page.items, vec![item_1, item_3]); + let descending_next_page = store + .list_items(ListItemsParams { + cursor: descending_page.next_cursor, + sort_direction: SortDirection::Desc, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect("next descending update-ordered item page"); + assert_eq!(descending_next_page.items, vec![item_2]); + + let error = store + .list_items(ListItemsParams { + sort_key: ItemSortKey::UpdatedAtOrdinal, + ..item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + ) + }) + .await + .expect_err("update-ordinal sorting should require a watermark"); + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); +} + +#[tokio::test] +async fn list_items_rejects_update_ordinals_outside_sqlite_integer_range() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + + for sort_key in [ItemSortKey::CreatedAtOrdinal, ItemSortKey::UpdatedAtOrdinal] { + let error = store + .list_items(ListItemsParams { + sort_key, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ u64::MAX) + }) + .await + .expect_err("out-of-range SQLite update ordinal should fail"); + + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); + } +} + #[tokio::test] async fn list_history_keeps_legacy_threads_unsupported() { let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Legacy).await; @@ -403,6 +557,17 @@ async fn lineage_reads_page_across_parent_and_child_segments() { vec!["root-user", "root-agent"] ); + for sort_key in [ItemSortKey::CreatedAtOrdinal, ItemSortKey::UpdatedAtOrdinal] { + let error = store + .list_items(ListItemsParams { + sort_key, + ..updated_item_params(child_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect_err("incremental replay should reject forked lineages"); + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); + } + let (_other_home, other_store, other_thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; let error = other_store @@ -657,12 +822,13 @@ async fn insert_item( rollout_ordinal: i64, ) { sqlx::query( - "INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_json) VALUES (?, ?, ?, ?, ?, ?)", + "INSERT INTO thread_items (thread_id, turn_id, item_id, rollout_ordinal, updated_at_ordinal, created_at_ms, item_json) VALUES (?, ?, ?, ?, ?, ?, ?)", ) .bind(thread_id.to_string()) .bind(turn_id) .bind(item_id) .bind(rollout_ordinal) + .bind(rollout_ordinal) .bind(rollout_ordinal * 1_000) .bind(format!(r#"{{"type":"userMessage","id":"{item_id}","content":[]}}"#)) .execute(db) @@ -701,6 +867,22 @@ fn item_params( cursor, page_size, sort_direction, + sort_key: ItemSortKey::CreatedAtOrdinal, + after_updated_at_ordinal: None, + } +} + +fn updated_item_params(thread_id: ThreadId, after_updated_at_ordinal: u64) -> ListItemsParams { + ListItemsParams { + sort_key: ItemSortKey::UpdatedAtOrdinal, + after_updated_at_ordinal: Some(after_updated_at_ordinal), + ..item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + ) } } @@ -708,6 +890,7 @@ fn expected_item(turn_id: &str, item_id: &str, rollout_ordinal: u64) -> StoredTh StoredThreadItem { turn_id: turn_id.to_string(), item_id: item_id.to_string(), + updated_at_ordinal: rollout_ordinal, 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(), diff --git a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs index b85d796222..8cba3b90b7 100644 --- a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs +++ b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs @@ -15,6 +15,8 @@ use super::read::serialize_cursor; use super::read::stored_thread_item_row_for_thread; use super::read::stored_turn_row; use super::thread_history_error; +use crate::ItemSortKey; +use crate::ListItemsParams; use crate::SortDirection; use crate::ThreadStoreError; use crate::ThreadStoreResult; @@ -87,34 +89,56 @@ WHERE thread_id = pub(super) async fn page_item_rows( pool: &sqlx::SqlitePool, - requested_thread_id: ThreadId, lineage: &RolloutLineage, - turn_id: Option<&str>, - cursor: Option<&str>, - page_size: usize, - direction: SortDirection, + params: &ListItemsParams, ) -> ThreadStoreResult> { - let cursor = parse_cursor(cursor, requested_thread_id, CursorScope::Items)?; + // Update ordinals are local to a physical rollout. Forked lineages need a structured + // watermark before incremental replay can safely span their segments. + if params.after_updated_at_ordinal.is_some() && lineage.segments().len() > 1 { + return Err(ThreadStoreError::InvalidRequest { + message: "incremental item replay is not supported for forked threads".to_string(), + }); + } + if matches!(params.sort_key, ItemSortKey::UpdatedAtOrdinal) { + let Some(after_updated_at_ordinal) = params.after_updated_at_ordinal else { + return Err(ThreadStoreError::InvalidRequest { + message: "update-ordinal item sorting requires an update watermark".to_string(), + }); + }; + return page_updated_item_rows(pool, params, after_updated_at_ordinal).await; + } + let cursor = parse_cursor( + params.cursor.as_deref(), + params.thread_id, + CursorScope::ItemsByCreatedAtOrdinal, + )?; let mut rows = Vec::new(); - for (segment, segment_cursor) in segments_from_cursor(lineage, direction, cursor.as_ref())? { - let remaining = remaining_limit(page_size, rows.len())?; + for (segment, segment_cursor) in + segments_from_cursor(lineage, params.sort_direction, cursor.as_ref())? + { + let remaining = remaining_limit(params.page_size, rows.len())?; if remaining == 0 { break; } let mut query = QueryBuilder::::new( r#" -SELECT turn_id, item_id, rollout_ordinal, created_at_ms, item_json +SELECT turn_id, item_id, rollout_ordinal, updated_at_ordinal, created_at_ms, item_json FROM thread_items WHERE thread_id = "#, ); query.push_bind(segment.thread_id().to_string()); push_segment_range(&mut query, segment)?; - if let Some(turn_id) = turn_id { + if let Some(after_updated_at_ordinal) = params.after_updated_at_ordinal { + query + .push(" AND updated_at_ordinal > ") + .push_bind(sqlite_integer(after_updated_at_ordinal)?); + } + if let Some(turn_id) = params.turn_id.as_deref() { query.push(" AND turn_id = ").push_bind(turn_id); } - push_cursor_clause(&mut query, direction, segment_cursor)?; - push_order_and_limit(&mut query, direction, remaining); + push_cursor_clause(&mut query, params.sort_direction, segment_cursor)?; + push_order_and_limit(&mut query, params.sort_direction, remaining); rows.extend( query .build() @@ -126,7 +150,80 @@ WHERE thread_id = .collect::>>()?, ); } - finish_page(requested_thread_id, CursorScope::Items, rows, page_size) + finish_page( + params.thread_id, + CursorScope::ItemsByCreatedAtOrdinal, + rows, + params.page_size, + ) +} + +async fn page_updated_item_rows( + pool: &sqlx::SqlitePool, + params: &ListItemsParams, + after_updated_at_ordinal: u64, +) -> ThreadStoreResult> { + let cursor = parse_cursor( + params.cursor.as_deref(), + params.thread_id, + CursorScope::ItemsByUpdatedAtOrdinal, + )?; + if cursor + .as_ref() + .is_some_and(|cursor| cursor.physical_thread_id != params.thread_id) + { + return Err(invalid_cursor("unknown physical segment")); + } + let mut query = QueryBuilder::::new( + r#" +SELECT turn_id, item_id, updated_at_ordinal AS rollout_ordinal, updated_at_ordinal, created_at_ms, item_json +FROM thread_items +WHERE thread_id = + "#, + ); + query + .push_bind(params.thread_id.to_string()) + .push(" AND updated_at_ordinal > ") + .push_bind(sqlite_integer(after_updated_at_ordinal)?); + if let Some(turn_id) = params.turn_id.as_deref() { + query.push(" AND turn_id = ").push_bind(turn_id); + } + if let Some(cursor) = cursor { + let comparator = match (params.sort_direction, cursor.include_anchor) { + (SortDirection::Asc, true) => ">=", + (SortDirection::Asc, false) => ">", + (SortDirection::Desc, true) => "<=", + (SortDirection::Desc, false) => "<", + }; + query + .push(" AND updated_at_ordinal ") + .push(comparator) + .push(" ") + .push_bind(sqlite_integer(cursor.rollout_ordinal)?); + } + let order = match params.sort_direction { + SortDirection::Asc => "ASC", + SortDirection::Desc => "DESC", + }; + query + .push(" ORDER BY updated_at_ordinal ") + .push(order) + .push(" LIMIT ") + .push_bind(remaining_limit(params.page_size, /*row_count*/ 0)?); + let rows = query + .build() + .fetch_all(pool) + .await + .map_err(thread_history_error)? + .into_iter() + .map(|row| stored_thread_item_row_for_thread(params.thread_id, row)) + .collect::>>()?; + finish_page( + params.thread_id, + CursorScope::ItemsByUpdatedAtOrdinal, + rows, + params.page_size, + ) } fn segments_from_cursor<'a>( diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 99516e44c9..ac44109e9e 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -491,7 +491,7 @@ async fn subagent_prefix_advances_projection_without_materializing_history() { } #[tokio::test] -async fn replayed_item_snapshot_updates_content_without_reordering() { +async fn replayed_item_snapshot_preserves_creation_ordinal_and_advances_update_ordinal() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); let thread_id = ThreadId::default(); @@ -545,9 +545,9 @@ async fn replayed_item_snapshot_updates_content_without_reordering() { .await .expect("append replayed item snapshot"); - let item = sqlx::query_as::<_, (i64, i64, String)>( + let item = sqlx::query_as::<_, (i64, i64, i64, String)>( r#" -SELECT rollout_ordinal, created_at_ms, item_json +SELECT rollout_ordinal, updated_at_ordinal, created_at_ms, item_json FROM thread_items WHERE thread_id = ? AND turn_id = ? AND item_id = ? "#, @@ -558,10 +558,9 @@ WHERE thread_id = ? AND turn_id = ? AND item_id = ? .fetch_one(&pool) .await .expect("read projected item"); - assert_eq!(item.0, 2); - assert_eq!(item.1, first_created_at_ms); + assert_eq!((item.0, item.1, item.2), (2, 3, first_created_at_ms)); assert_eq!( - serde_json::from_str::(item.2.as_str()).expect("parse projected item"), + serde_json::from_str::(item.3.as_str()).expect("parse projected item"), ThreadItem::UserMessage { id: "user-1".to_string(), client_id: Some("updated".to_string()), diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index b81f4cd4b8..ccfe7c0f71 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -399,8 +399,21 @@ pub struct ListItemsParams { pub cursor: Option, /// Maximum number of items to return. pub page_size: usize, - /// Sort direction requested by the caller. + /// Direction to sort items by the selected ordinal. pub sort_direction: SortDirection, + /// Ordinal to sort items by. Update-ordinal sorting requires an update watermark. + pub sort_key: ItemSortKey, + /// Filters out items with an update ordinal less than or equal to the provided value. + pub after_updated_at_ordinal: Option, +} + +/// The ordinal to use when listing persisted items. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum ItemSortKey { + /// Sort by the ordinal where the item was first projected. + CreatedAtOrdinal, + /// Sort by the ordinal where the item was last updated. + UpdatedAtOrdinal, } /// A projected app-server `ThreadItem` snapshot within a turn. @@ -410,6 +423,8 @@ pub struct StoredThreadItem { pub turn_id: String, /// Stable item identifier within the turn. pub item_id: String, + /// Rollout ordinal of the latest persisted update to this item. + pub updated_at_ordinal: u64, /// Unix timestamp (milliseconds) when this logical item was first projected. pub created_at_ms: i64, /// Serialized app-server ThreadItem snapshot.