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

@@ -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;

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 {

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 {

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();

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);

codex-rs/thread-store/src/in_memory.rs

@@ -50,6 +50,7 @@ fn stores() -> &'static Mutex<HashMap<String, Arc<InMemoryThreadStore>>> {
 #[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");

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;

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)

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, &params).await?;
 
     Ok(ItemPage {
         items: page.rows.into_iter().map(|row| row.item).collect(),
@@ -180,7 +172,7 @@ async fn load_summary_items(
 ) -> ThreadStoreResult<Vec<StoredThreadItem>> {
     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<StoredThreadItem> {
+    let updated_at_ordinal = row.try_get::<i64, _>("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::<String, _>("item_json")?.into_bytes(),
     })

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(),

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<SegmentPage<StoredThreadItemRow>> {
-    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::<Sqlite>::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::<ThreadStoreResult<Vec<_>>>()?,
         );
     }
-    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<SegmentPage<StoredThreadItemRow>> {
+    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::<Sqlite>::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::<ThreadStoreResult<Vec<_>>>()?;
+    finish_page(
+        params.thread_id,
+        CursorScope::ItemsByUpdatedAtOrdinal,
+        rows,
+        params.page_size,
+    )
 }
 
 fn segments_from_cursor<'a>(

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::<ThreadItem>(item.2.as_str()).expect("parse projected item"),
+        serde_json::from_str::<ThreadItem>(item.3.as_str()).expect("parse projected item"),
         ThreadItem::UserMessage {
             id: "user-1".to_string(),
             client_id: Some("updated".to_string()),

codex-rs/thread-store/src/types.rs

@@ -399,8 +399,21 @@ pub struct ListItemsParams {
     pub cursor: Option<String>,
     /// 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<u64>,
+}
+
+/// 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.