diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 587325c246..a26db57902 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -444,6 +444,7 @@ use codex_state::log_db::LogDbLayer; use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; use codex_thread_store::DeleteThreadParams as StoreDeleteThreadParams; use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; +use codex_thread_store::ListItemsFilters as StoreListItemsFilters; use codex_thread_store::ListItemsParams as StoreListItemsParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::LocalThreadStore; 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 af8525873f..07d221d7ff 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2461,8 +2461,11 @@ impl ThreadRequestProcessor { .thread_store .list_items(StoreListItemsParams { thread_id, - turn_id, include_archived: true, + filters: StoreListItemsFilters { + turn_id, + updated_at: None, + }, cursor, page_size, sort_direction: match sort_direction.unwrap_or(SortDirection::Asc) { diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index c80d013ba2..0b317b06ac 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -50,9 +50,9 @@ fn stores() -> &'static Mutex>> { #[cfg(test)] mod tests { use super::*; + use crate::ListItemsFilters; use crate::ListItemsParams; use crate::ListTurnsParams; - use crate::ListUpdatedItemsParams; use crate::SortDirection; use crate::StoredTurnItemsView; use crate::ThreadPersistenceMetadata; @@ -86,8 +86,8 @@ mod tests { let items_err = store .list_items(ListItemsParams { thread_id, - turn_id: None, include_archived: true, + filters: ListItemsFilters::default(), cursor: None, page_size: 10, sort_direction: SortDirection::Asc, @@ -100,25 +100,6 @@ mod tests { operation: "list_items" } )); - - let updated_items_err = store - .list_updated_items(ListUpdatedItemsParams { - thread_id, - turn_id: None, - after_updated_at: None, - through_updated_at: None, - cursor: None, - page_size: 10, - sort_direction: SortDirection::Asc, - }) - .await - .expect_err("default list_updated_items should be unsupported"); - assert!(matches!( - updated_items_err, - ThreadStoreError::Unsupported { - operation: "list_updated_items" - } - )); } #[tokio::test] diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index ebb7b6b44d..6ebcd712e8 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -30,10 +30,10 @@ pub use types::DeleteThreadParams; pub use types::ExtraConfig; pub use types::GitInfoPatch; pub use types::ItemPage; +pub use types::ListItemsFilters; pub use types::ListItemsParams; pub use types::ListThreadsParams; pub use types::ListTurnsParams; -pub use types::ListUpdatedItemsParams; pub use types::LoadThreadHistoryParams; pub use types::ReadThreadByRolloutPathParams; pub use types::ReadThreadParams; @@ -56,4 +56,4 @@ pub use types::ThreadSearchPage; pub use types::ThreadSortKey; pub use types::TurnPage; pub use types::UpdateThreadMetadataParams; -pub use types::UpdatedItemsPage; +pub use types::UpdatedAtFilter; diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 77ba42bb58..68d572d7a2 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -12,7 +12,6 @@ use crate::ItemPage; use crate::ListItemsParams; use crate::ListThreadsParams; use crate::ListTurnsParams; -use crate::ListUpdatedItemsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; @@ -26,7 +25,6 @@ use crate::ThreadStoreError; use crate::ThreadStoreResult; use crate::TurnPage; use crate::UpdateThreadMetadataParams; -use crate::UpdatedItemsPage; /// Future returned by [`ThreadStore`] operations. pub type ThreadStoreFuture<'a, T> = Pin> + Send + 'a>>; @@ -113,7 +111,7 @@ pub trait ThreadStore: Any + Send + Sync { }) } - /// Lists persisted items within a stored thread, optionally filtered to a turn. + /// Lists persisted items within a stored thread matching the supplied filters. fn list_items(&self, _params: ListItemsParams) -> ThreadStoreFuture<'_, ItemPage> { Box::pin(async { Err(ThreadStoreError::Unsupported { @@ -122,18 +120,6 @@ pub trait ThreadStore: Any + Send + Sync { }) } - /// Lists latest item snapshots changed within a timestamp-bounded interval. - fn list_updated_items( - &self, - _params: ListUpdatedItemsParams, - ) -> ThreadStoreFuture<'_, UpdatedItemsPage> { - Box::pin(async { - Err(ThreadStoreError::Unsupported { - operation: "list_updated_items", - }) - }) - } - /// Applies a literal metadata patch and returns the updated thread. /// /// Implementations should apply the supplied fields directly. Policy such as deciding whether diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index c6f1dea7c6..505ddac81d 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -372,39 +372,44 @@ pub struct TurnPage { pub backwards_cursor: Option, } +/// Optional filters for listing persisted items. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ListItemsFilters { + /// Optional turn id to filter by. When omitted, returns items across the thread. + pub turn_id: Option, + + /// Optional item update timestamp filter. + #[serde(skip_serializing_if = "Option::is_none")] + pub updated_at: Option, +} + +/// Bounds for filtering item snapshots by their update timestamp. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct UpdatedAtFilter { + /// Exclusive lower bound: `updated_at > gt`. + #[serde(skip_serializing_if = "Option::is_none")] + pub gt: Option>, + + /// Inclusive upper bound: `updated_at <= lte`. + pub lte: DateTime, +} + /// Parameters for listing persisted items within a thread. +/// +/// Callers must reuse the same filters when following a page cursor. Implementations that forward +/// an [`UpdatedAtFilter`] must verify that the backing store applied its exact `lte` bound before +/// returning a successful page. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ListItemsParams { /// Thread id to read. pub thread_id: ThreadId, - /// Optional turn id to filter by. When omitted, returns items across the thread. - pub turn_id: Option, /// Whether archived threads are eligible. pub include_archived: bool, - /// Opaque cursor returned by a previous list call. - pub cursor: Option, - /// Maximum number of items to return. - pub page_size: usize, - /// Sort direction requested by the caller. - pub sort_direction: SortDirection, -} - -/// Parameters for listing items changed within a timestamp-bounded interval. -/// -/// When both bounds are present, implementations should interpret the interval as -/// `(after_updated_at, through_updated_at]`. Bounds remain optional so each store owns validation -/// of partial ranges. Callers should use [`crate::ThreadStore::list_items`] when neither bound is -/// present, and must reuse the same bounds when following a page cursor. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ListUpdatedItemsParams { - /// Thread id to read. - pub thread_id: ThreadId, - /// Optional turn id to filter by. When omitted, returns items across the thread. - pub turn_id: Option, - /// Exclusive lower bound for item updates. - pub after_updated_at: Option>, - /// Inclusive upper bound for item updates. - pub through_updated_at: Option>, + /// Filters applied to the listed items. + /// + /// Flattening preserves the existing serialized `turn_id` field. + #[serde(flatten)] + pub filters: ListItemsFilters, /// Opaque cursor returned by a previous list call. pub cursor: Option, /// Maximum number of items to return. @@ -434,20 +439,6 @@ pub struct ItemPage { pub backwards_cursor: Option, } -/// A page of items changed within a timestamp-bounded interval. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct UpdatedItemsPage { - /// Latest snapshots of items changed within the requested interval. - pub items: Vec, - /// Opaque cursor to continue listing with the same timestamp bounds. - pub next_cursor: Option, - /// Opaque cursor for fetching in the opposite direction with the same timestamp bounds. - pub backwards_cursor: Option, - /// Store-confirmed inclusive upper bound. On success, this must match the requested upper - /// bound, including when that bound is absent. - pub read_through_updated_at: Option>, -} - /// Store-owned thread metadata used by list/read/resume responses. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct StoredThread { @@ -775,6 +766,30 @@ mod tests { use super::*; + #[test] + fn list_items_filters_preserve_flattened_turn_id_shape() { + let params = ListItemsParams { + thread_id: ThreadId::default(), + include_archived: true, + filters: ListItemsFilters { + turn_id: Some("turn-1".to_string()), + updated_at: None, + }, + cursor: Some("next-page".to_string()), + page_size: 25, + sort_direction: SortDirection::Asc, + }; + + let value = serde_json::to_value(¶ms).expect("serialize list items params"); + assert_eq!(value["turn_id"], json!("turn-1")); + assert_eq!(value.get("filters"), None); + assert_eq!(value.get("updated_at"), None); + + let decoded: ListItemsParams = + serde_json::from_value(value).expect("deserialize legacy-shaped list items params"); + assert_eq!(decoded, params); + } + #[test] fn thread_metadata_patch_round_trips_optional_clears() { let patch = ThreadMetadataPatch {