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 b37330deae..130bb85a9e 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3596,6 +3596,7 @@ impl ThreadRequestProcessor { sort_direction: StoreSortDirection::Asc, sort_key: StoreItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .map_err(paginated_history_list_error)?; @@ -3713,6 +3714,7 @@ impl ThreadRequestProcessor { sort_direction: StoreSortDirection::Desc, sort_key: StoreItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .map_err(paginated_history_list_error)?; @@ -3766,6 +3768,7 @@ impl ThreadRequestProcessor { }, sort_key: StoreItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .map_err(|err| match err { @@ -3837,6 +3840,7 @@ impl ThreadRequestProcessor { sort_direction: StoreSortDirection::Asc, sort_key: StoreItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .map_err(|err| match err { diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 049e89cafc..a3e34aaeb0 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -138,6 +138,7 @@ mod tests { sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .expect_err("default list_items should be unsupported"); diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 6bc91d55a4..ee59f60074 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -508,6 +508,7 @@ async fn migration_projects_explicit_and_implicit_legacy_completed_items() { sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .expect("read projected items"); @@ -730,6 +731,7 @@ async fn migration_keeps_late_completions_in_their_original_turn() { sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .expect("read projected late completion"); @@ -1221,6 +1223,7 @@ async fn migration_keeps_late_completions_for_surviving_turns_across_rollback() sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, }) .await .expect("read late-completion rollback items"); 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 e2afaa6423..b00acc3199 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -175,6 +175,7 @@ pub(in crate::local) async fn list_items( if params.turn_id.is_none() || params.position.is_some() || params.after_updated_at_ordinal.is_some() + || params.before_updated_at_ordinal.is_some() || params.sort_key != crate::ItemSortKey::CreatedAtOrdinal || !(1..=100).contains(&ids.len()) { 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 b974327dc8..62b1f5680c 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 @@ -335,7 +335,7 @@ async fn item_anchors_page_exclusively_while_the_turn_grows() { } #[tokio::test] -async fn item_anchors_require_creation_order_without_an_update_watermark() { +async fn item_anchors_require_creation_order_without_an_update_boundary() { let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; insert_item( history_db(&store).await, @@ -345,9 +345,10 @@ async fn item_anchors_require_creation_order_without_an_update_watermark() { /*rollout_ordinal*/ 1, ) .await; - for (sort_key, after_updated_at_ordinal) in [ - (ItemSortKey::UpdatedAtOrdinal, None), - (ItemSortKey::CreatedAtOrdinal, Some(0)), + for (sort_key, after_updated_at_ordinal, before_updated_at_ordinal) in [ + (ItemSortKey::UpdatedAtOrdinal, None, None), + (ItemSortKey::CreatedAtOrdinal, Some(0), None), + (ItemSortKey::CreatedAtOrdinal, None, Some(2)), ] { let params = ListItemsParams { position: Some(ListItemsPosition::ItemAnchor { @@ -355,6 +356,7 @@ async fn item_anchors_require_creation_order_without_an_update_watermark() { }), sort_key, after_updated_at_ordinal, + before_updated_at_ordinal, ..item_params( thread_id, Some("turn"), @@ -368,7 +370,7 @@ async fn item_anchors_require_creation_order_without_an_update_watermark() { .await .expect_err("invalid anchor paging mode"); assert!(matches!(error, ThreadStoreError::InvalidRequest { message } - if message == "item anchors require creation-order paging without an update watermark")); + if message == "item anchors require creation-order paging without an update boundary")); } } @@ -844,6 +846,25 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { .expect("creation-ordered filtered item page"); assert_eq!(page.items, expected); } + for (sort_direction, expected) in [ + (SortDirection::Asc, vec![item_2.clone(), item_3.clone()]), + (SortDirection::Desc, vec![item_3.clone(), item_2.clone()]), + ] { + let page = store + .list_items(ListItemsParams { + before_updated_at_ordinal: Some(4), + ..item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + sort_direction, + ) + }) + .await + .expect("creation-ordered before-filtered item page"); + assert_eq!(page.items, expected); + } let first_page = store .list_items(updated_item_params( @@ -898,7 +919,7 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { }) .await .expect("descending update-ordered item page"); - assert_eq!(descending_page.items, vec![item_1, item_3]); + assert_eq!(descending_page.items, vec![item_1, item_3.clone()]); let descending_next_page = store .list_items(ListItemsParams { position: descending_page.next_cursor.map(ListItemsPosition::Cursor), @@ -907,7 +928,43 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { }) .await .expect("next descending update-ordered item page"); - assert_eq!(descending_next_page.items, vec![item_2]); + assert_eq!(descending_next_page.items, vec![item_2.clone()]); + + let before_descending_page = store + .list_items(ListItemsParams { + after_updated_at_ordinal: None, + before_updated_at_ordinal: Some(4), + page_size: 1, + sort_direction: SortDirection::Desc, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect("descending before-filtered item page"); + assert_eq!(before_descending_page.items, vec![item_3.clone()]); + let before_descending_next_page = store + .list_items(ListItemsParams { + after_updated_at_ordinal: None, + before_updated_at_ordinal: Some(4), + position: before_descending_page + .next_cursor + .map(ListItemsPosition::Cursor), + page_size: 1, + sort_direction: SortDirection::Desc, + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect("next descending before-filtered item page"); + assert_eq!(before_descending_next_page.items, vec![item_2]); + assert!(before_descending_next_page.next_cursor.is_none()); + + let error = store + .list_items(ListItemsParams { + before_updated_at_ordinal: Some(4), + ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) + }) + .await + .expect_err("after and before update ordinals should be mutually exclusive"); + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); let error = store .list_items(ListItemsParams { @@ -929,16 +986,28 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { 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"); + for (after_updated_at_ordinal, before_updated_at_ordinal) in + [(Some(u64::MAX), None), (None, Some(u64::MAX))] + { + for sort_key in [ItemSortKey::CreatedAtOrdinal, ItemSortKey::UpdatedAtOrdinal] { + let error = store + .list_items(ListItemsParams { + sort_key, + after_updated_at_ordinal, + before_updated_at_ordinal, + ..item_params( + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Asc, + ) + }) + .await + .expect_err("out-of-range SQLite update ordinal should fail"); - assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); + assert!(matches!(error, ThreadStoreError::InvalidRequest { .. })); + } } } @@ -1211,16 +1280,73 @@ async fn lineage_reads_page_across_parent_and_child_segments() { .is_empty() ); - 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 mut creation_order_params = + updated_item_params(child_id, /*after_updated_at_ordinal*/ 0); + creation_order_params.page_size = 10; + creation_order_params.sort_key = ItemSortKey::CreatedAtOrdinal; + let creation_order_update_filter = store + .list_items(creation_order_params) + .await + .expect("creation-ordered lineage items with update filter"); + assert_eq!( + item_ids(&creation_order_update_filter), + vec!["root-user", "root-agent", "root-2-item", "child-item"] + ); + + let first_updated_page = store + .list_items(updated_item_params( + child_id, /*after_updated_at_ordinal*/ 0, + )) + .await + .expect("first update-ordered lineage page"); + assert_eq!( + item_ids(&first_updated_page), + vec!["root-user", "root-agent"] + ); + let mut second_updated_params = + updated_item_params(child_id, /*after_updated_at_ordinal*/ 0); + second_updated_params.position = Some(ListItemsPosition::Cursor( + first_updated_page + .next_cursor + .expect("first update-ordered page has a cursor"), + )); + let second_updated_page = store + .list_items(second_updated_params) + .await + .expect("second update-ordered lineage page"); + assert_eq!( + item_ids(&second_updated_page), + vec!["root-2-item", "child-item"] + ); + assert!(second_updated_page.next_cursor.is_none()); + + let mut backward_updated_params = + updated_item_params(child_id, /*after_updated_at_ordinal*/ 0); + backward_updated_params.after_updated_at_ordinal = None; + backward_updated_params.before_updated_at_ordinal = Some(9); + backward_updated_params.sort_direction = SortDirection::Desc; + let first_backward_updated_page = store + .list_items(backward_updated_params.clone()) + .await + .expect("first backward update-ordered lineage page"); + assert_eq!( + item_ids(&first_backward_updated_page), + vec!["child-item", "root-2-item"] + ); + backward_updated_params.position = Some(ListItemsPosition::Cursor( + first_backward_updated_page + .next_cursor + .expect("first backward update-ordered page has a cursor"), + )); + let second_backward_updated_page = store + .list_items(backward_updated_params) + .await + .expect("second backward update-ordered lineage page"); + assert_eq!( + item_ids(&second_backward_updated_page), + vec!["root-agent", "root-user"] + ); + assert!(second_backward_updated_page.next_cursor.is_none()); let first_occurrences = store .search_thread_occurrences(SearchThreadOccurrencesParams { @@ -1739,6 +1865,7 @@ fn item_params( sort_direction, sort_key: ItemSortKey::CreatedAtOrdinal, after_updated_at_ordinal: None, + before_updated_at_ordinal: None, } } 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 69cc63244c..86a13ce1c9 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 @@ -31,6 +31,46 @@ pub(super) struct SegmentPage { pub backwards_cursor: Option, } +#[derive(Clone, Copy)] +enum UpdateOrdinalBoundary { + After(u64), + Before(u64), +} + +#[derive(Clone, Copy)] +enum HistoryOrdinalColumn { + Rollout, + UpdatedAt, +} + +impl HistoryOrdinalColumn { + fn push(self, query: &mut QueryBuilder) { + query.push(match self { + Self::Rollout => "rollout_ordinal", + Self::UpdatedAt => "updated_at_ordinal", + }); + } +} + +fn push_update_ordinal_boundary( + query: &mut QueryBuilder, + boundary: UpdateOrdinalBoundary, +) -> ThreadStoreResult<()> { + match boundary { + UpdateOrdinalBoundary::After(ordinal) => { + query + .push(" AND updated_at_ordinal > ") + .push_bind(sqlite_integer(ordinal)?); + } + UpdateOrdinalBoundary::Before(ordinal) => { + query + .push(" AND updated_at_ordinal < ") + .push_bind(sqlite_integer(ordinal)?); + } + } + Ok(()) +} + pub(super) fn validate_page_size(page_size: usize) -> ThreadStoreResult<()> { if page_size == 0 { return Err(ThreadStoreError::InvalidRequest { @@ -52,6 +92,7 @@ pub(super) async fn page_turn_rows( items_view: StoredTurnItemsView, ) -> ThreadStoreResult> { let cursor = parse_cursor(cursor, requested_thread_id, CursorScope::Turns)?; + let ordinal_column = HistoryOrdinalColumn::Rollout; let mut rows = Vec::new(); for (segment_index, segment, segment_cursor) in segments_from_cursor(lineage, direction, cursor.as_ref())? @@ -81,7 +122,7 @@ WHERE thread_id = "#, ); query.push_bind(segment.rollout_id().to_string()); - push_segment_range(&mut query, segment)?; + push_segment_range(&mut query, segment, ordinal_column)?; for newer_segment in &lineage.segments()[segment_index + 1..] { query .push(" AND NOT EXISTS (SELECT 1 FROM thread_turns AS newer_turn WHERE newer_turn.thread_id = ") @@ -95,8 +136,8 @@ WHERE thread_id = } query.push(")"); } - push_cursor_clause(&mut query, direction, segment_cursor)?; - push_order_and_limit(&mut query, direction, remaining); + push_cursor_clause(&mut query, direction, segment_cursor, ordinal_column)?; + push_order_and_limit(&mut query, direction, remaining, ordinal_column); if matches!(items_view, StoredTurnItemsView::Summary) { query.push( r#" @@ -201,41 +242,34 @@ pub(super) async fn page_item_rows( Some(ListItemsPosition::ItemAnchor { item_id }) => (None, Some(item_id.as_str())), None => (None, None), }; + let update_boundary = match ( + params.after_updated_at_ordinal, + params.before_updated_at_ordinal, + ) { + (Some(_), Some(_)) => { + return Err(ThreadStoreError::InvalidRequest { + message: "after and before update ordinals are mutually exclusive".to_string(), + }); + } + (Some(ordinal), None) => Some(UpdateOrdinalBoundary::After(ordinal)), + (None, Some(ordinal)) => Some(UpdateOrdinalBoundary::Before(ordinal)), + (None, None) => None, + }; if anchor_item_id.is_some() - && (params.sort_key != ItemSortKey::CreatedAtOrdinal - || params.after_updated_at_ordinal.is_some()) + && (params.sort_key != ItemSortKey::CreatedAtOrdinal || update_boundary.is_some()) { return Err(ThreadStoreError::InvalidRequest { - message: "item anchors require creation-order paging without an update watermark" + message: "item anchors require creation-order paging without an update boundary" .to_string(), }); } - // Update ordinals are local to a 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 { + let Some(update_boundary) = update_boundary else { return Err(ThreadStoreError::InvalidRequest { - message: "update-ordinal item sorting requires an update watermark".to_string(), + message: "update-ordinal item sorting requires an update boundary".to_string(), }); }; - let [segment] = lineage.segments() else { - return Err(ThreadStoreError::Internal { - message: "update-ordinal item paging requires one rollout segment".to_string(), - }); - }; - return page_updated_item_rows( - pool, - segment.rollout_id(), - params, - cursor, - after_updated_at_ordinal, - ) - .await; + return page_updated_item_rows(pool, lineage, params, cursor, update_boundary).await; } let cursor = if let Some(anchor_item_id) = anchor_item_id { let turn_id = params @@ -255,7 +289,7 @@ pub(super) async fn page_item_rows( ); query.push_bind(segment.rollout_id().to_string()); query.push(" AND item_id = ").push_bind(anchor_item_id); - push_segment_range(&mut query, segment)?; + push_segment_range(&mut query, segment, HistoryOrdinalColumn::Rollout)?; query.push(" AND turn_id = ").push_bind(turn_id); if let Some(ordinal) = query .build_query_scalar::() @@ -281,6 +315,7 @@ pub(super) async fn page_item_rows( CursorScope::ItemsByCreatedAtOrdinal, )? }; + let ordinal_column = HistoryOrdinalColumn::Rollout; let mut rows = Vec::new(); for (_, segment, segment_cursor) in segments_from_cursor(lineage, params.sort_direction, cursor.as_ref())? @@ -297,18 +332,17 @@ FROM thread_items"#, if params.item_ids.is_some() { // Seek each requested ID instead of scanning the turn in rollout order. query.push(" INDEXED BY sqlite_autoindex_thread_items_1"); - } else if params.turn_id.is_some() && params.after_updated_at_ordinal.is_none() { + } else if params.turn_id.is_some() && update_boundary.is_none() { // A thread-wide index can scan earlier turns when a turn's final page is underfilled. query.push(" INDEXED BY idx_thread_items_by_turn_page"); } query .push(" WHERE thread_id = ") .push_bind(segment.rollout_id().to_string()); - push_segment_range(&mut query, segment)?; - 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)?); + push_segment_range(&mut query, segment, ordinal_column)?; + if let Some(update_boundary) = update_boundary { + push_segment_range(&mut query, segment, HistoryOrdinalColumn::UpdatedAt)?; + push_update_ordinal_boundary(&mut query, update_boundary)?; } if let Some(turn_id) = params.turn_id.as_deref() { query.push(" AND turn_id = ").push_bind(turn_id); @@ -321,8 +355,13 @@ FROM thread_items"#, } ids.push_unseparated(")"); } - push_cursor_clause(&mut query, params.sort_direction, segment_cursor)?; - push_order_and_limit(&mut query, params.sort_direction, remaining); + push_cursor_clause( + &mut query, + params.sort_direction, + segment_cursor, + ordinal_column, + )?; + push_order_and_limit(&mut query, params.sort_direction, remaining, ordinal_column); rows.extend( query .build() @@ -344,60 +383,56 @@ FROM thread_items"#, async fn page_updated_item_rows( pool: &sqlx::SqlitePool, - rollout_id: ThreadId, + lineage: &RolloutLineage, params: &ListItemsParams, cursor: Option<&str>, - after_updated_at_ordinal: u64, + update_boundary: UpdateOrdinalBoundary, ) -> ThreadStoreResult> { let cursor = parse_cursor( cursor, params.thread_id, CursorScope::ItemsByUpdatedAtOrdinal, )?; - let mut query = QueryBuilder::::new( - r#" + let ordinal_column = HistoryOrdinalColumn::UpdatedAt; + let mut rows = Vec::new(); + 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, updated_at_ordinal AS rollout_ordinal, updated_at_ordinal, created_at_ms, started_at_ms, completed_at_ms, item_json FROM thread_items WHERE thread_id = "#, - ); - query - .push_bind(rollout_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); + ); + query.push_bind(segment.rollout_id().to_string()); + push_segment_range(&mut query, segment, ordinal_column)?; + push_update_ordinal_boundary(&mut query, update_boundary)?; + if let Some(turn_id) = params.turn_id.as_deref() { + query.push(" AND turn_id = ").push_bind(turn_id); + } + push_cursor_clause( + &mut query, + params.sort_direction, + segment_cursor, + ordinal_column, + )?; + push_order_and_limit(&mut query, params.sort_direction, remaining, ordinal_column); + rows.extend( + query + .build() + .fetch_all(pool) + .await + .map_err(thread_history_error)? + .into_iter() + .map(stored_thread_item_row) + .collect::>>()?, + ); } - 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(stored_thread_item_row) - .collect::>>()?; finish_page( params.thread_id, CursorScope::ItemsByUpdatedAtOrdinal, @@ -442,14 +477,17 @@ fn segments_from_cursor<'a>( fn push_segment_range( query: &mut QueryBuilder, segment: &RolloutLineageSegment, + ordinal_column: HistoryOrdinalColumn, ) -> ThreadStoreResult<()> { + query.push(" AND "); + ordinal_column.push(query); query - .push(" AND rollout_ordinal >= ") + .push(" >= ") .push_bind(sqlite_integer(segment.start_ordinal())?); if let Some(end_ordinal) = segment.end_ordinal() { - query - .push(" AND rollout_ordinal < ") - .push_bind(sqlite_integer(end_ordinal)?); + query.push(" AND "); + ordinal_column.push(query); + query.push(" < ").push_bind(sqlite_integer(end_ordinal)?); } Ok(()) } @@ -458,6 +496,7 @@ fn push_cursor_clause( query: &mut QueryBuilder, direction: SortDirection, cursor: Option<&HistoryCursor>, + ordinal_column: HistoryOrdinalColumn, ) -> ThreadStoreResult<()> { if let Some(cursor) = cursor { let comparator = match (direction, cursor.include_anchor) { @@ -466,8 +505,10 @@ fn push_cursor_clause( (SortDirection::Desc, true) => "<=", (SortDirection::Desc, false) => "<", }; + query.push(" AND "); + ordinal_column.push(query); query - .push(" AND rollout_ordinal ") + .push(" ") .push(comparator) .push(" ") .push_bind(sqlite_integer(cursor.rollout_ordinal)?); @@ -475,16 +516,19 @@ fn push_cursor_clause( Ok(()) } -fn push_order_and_limit(query: &mut QueryBuilder, direction: SortDirection, limit: i64) { +fn push_order_and_limit( + query: &mut QueryBuilder, + direction: SortDirection, + limit: i64, + ordinal_column: HistoryOrdinalColumn, +) { let order = match direction { SortDirection::Asc => "ASC", SortDirection::Desc => "DESC", }; - query - .push(" ORDER BY rollout_ordinal ") - .push(order) - .push(" LIMIT ") - .push_bind(limit); + query.push(" ORDER BY "); + ordinal_column.push(query); + query.push(" ").push(order).push(" LIMIT ").push_bind(limit); } fn remaining_limit(page_size: usize, row_count: usize) -> ThreadStoreResult { diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index ebeff03117..c44b5045a7 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -514,7 +514,7 @@ pub enum ListItemsPosition { /// Opaque cursor returned by a previous list call. Cursor(String), /// Exclusive item anchor within the visible non-empty turn. - /// Requires creation order without an update watermark. + /// Requires creation order without an update boundary. ItemAnchor { item_id: String }, } @@ -535,10 +535,12 @@ pub struct ListItemsParams { pub page_size: usize, /// Direction to sort items by the selected ordinal. pub sort_direction: SortDirection, - /// Ordinal to sort items by. Update-ordinal sorting requires an update watermark. + /// Ordinal to sort items by. Update-ordinal sorting requires an update boundary. 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, + /// Filters out items with an update ordinal greater than or equal to the provided value. + pub before_updated_at_ordinal: Option, } /// The ordinal to use when listing persisted items.