[thread-store] consolidate item list filters

This commit is contained in:
Richard Lee
2026-07-06 11:20:12 -07:00
parent df70ca822a
commit 7a66cc7d72
6 changed files with 65 additions and 79 deletions
@@ -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;
@@ -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) {
+2 -21
View File
@@ -50,9 +50,9 @@ fn stores() -> &'static Mutex<HashMap<String, Arc<InMemoryThreadStore>>> {
#[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]
+2 -2
View File
@@ -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;
+1 -15
View File
@@ -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<Box<dyn Future<Output = ThreadStoreResult<T>> + 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
+55 -40
View File
@@ -372,39 +372,44 @@ pub struct TurnPage {
pub backwards_cursor: Option<String>,
}
/// 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<String>,
/// Optional item update timestamp filter.
#[serde(skip_serializing_if = "Option::is_none")]
pub updated_at: Option<UpdatedAtFilter>,
}
/// 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<DateTime<Utc>>,
/// Inclusive upper bound: `updated_at <= lte`.
pub lte: DateTime<Utc>,
}
/// 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<String>,
/// Whether archived threads are eligible.
pub include_archived: bool,
/// Opaque cursor returned by a previous list call.
pub cursor: Option<String>,
/// 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<String>,
/// Exclusive lower bound for item updates.
pub after_updated_at: Option<DateTime<Utc>>,
/// Inclusive upper bound for item updates.
pub through_updated_at: Option<DateTime<Utc>>,
/// 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<String>,
/// Maximum number of items to return.
@@ -434,20 +439,6 @@ pub struct ItemPage {
pub backwards_cursor: Option<String>,
}
/// 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<StoredThreadItem>,
/// Opaque cursor to continue listing with the same timestamp bounds.
pub next_cursor: Option<String>,
/// Opaque cursor for fetching in the opposite direction with the same timestamp bounds.
pub backwards_cursor: Option<String>,
/// 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<DateTime<Utc>>,
}
/// 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(&params).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 {