diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 7a89eeaea7..c4e6641192 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4491,14 +4491,54 @@ ], "type": "object" }, + "ThreadItemsListAnchor": { + "description": "An exclusive item position within the requested visible turn.", + "oneOf": [ + { + "properties": { + "itemId": { + "type": "string" + }, + "type": { + "enum": [ + "item" + ], + "title": "ItemThreadItemsListAnchorType", + "type": "string" + } + }, + "required": [ + "itemId", + "type" + ], + "title": "ItemThreadItemsListAnchor", + "type": "object" + } + ] + }, + "ThreadItemsListCursor": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/definitions/ThreadItemsListAnchor" + } + ], + "description": "Starting position for an item-history page." + }, "ThreadItemsListParams": { "properties": { "cursor": { - "description": "Opaque cursor to pass to the next call to continue after the last item.", - "type": [ - "string", - "null" - ] + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemsListCursor" + }, + { + "type": "null" + } + ], + "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, "limit": { "description": "Optional item page size.", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 94cfea4a63..ecb6d4a185 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -22589,15 +22589,55 @@ ], "type": "object" }, + "ThreadItemsListAnchor": { + "description": "An exclusive item position within the requested visible turn.", + "oneOf": [ + { + "properties": { + "itemId": { + "type": "string" + }, + "type": { + "enum": [ + "item" + ], + "title": "ItemThreadItemsListAnchorType", + "type": "string" + } + }, + "required": [ + "itemId", + "type" + ], + "title": "ItemThreadItemsListAnchor", + "type": "object" + } + ] + }, + "ThreadItemsListCursor": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/definitions/v2/ThreadItemsListAnchor" + } + ], + "description": "Starting position for an item-history page." + }, "ThreadItemsListParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { "cursor": { - "description": "Opaque cursor to pass to the next call to continue after the last item.", - "type": [ - "string", - "null" - ] + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadItemsListCursor" + }, + { + "type": "null" + } + ], + "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, "limit": { "description": "Optional item page size.", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index b8a69ed326..f7127cfb0f 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -20303,15 +20303,55 @@ ], "type": "object" }, + "ThreadItemsListAnchor": { + "description": "An exclusive item position within the requested visible turn.", + "oneOf": [ + { + "properties": { + "itemId": { + "type": "string" + }, + "type": { + "enum": [ + "item" + ], + "title": "ItemThreadItemsListAnchorType", + "type": "string" + } + }, + "required": [ + "itemId", + "type" + ], + "title": "ItemThreadItemsListAnchor", + "type": "object" + } + ] + }, + "ThreadItemsListCursor": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/definitions/ThreadItemsListAnchor" + } + ], + "description": "Starting position for an item-history page." + }, "ThreadItemsListParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { "cursor": { - "description": "Opaque cursor to pass to the next call to continue after the last item.", - "type": [ - "string", - "null" - ] + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemsListCursor" + }, + { + "type": "null" + } + ], + "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, "limit": { "description": "Optional item page size.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json index 4a21804479..6fb586ee02 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json @@ -7,15 +7,55 @@ "desc" ], "type": "string" + }, + "ThreadItemsListAnchor": { + "description": "An exclusive item position within the requested visible turn.", + "oneOf": [ + { + "properties": { + "itemId": { + "type": "string" + }, + "type": { + "enum": [ + "item" + ], + "title": "ItemThreadItemsListAnchorType", + "type": "string" + } + }, + "required": [ + "itemId", + "type" + ], + "title": "ItemThreadItemsListAnchor", + "type": "object" + } + ] + }, + "ThreadItemsListCursor": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/definitions/ThreadItemsListAnchor" + } + ], + "description": "Starting position for an item-history page." } }, "properties": { "cursor": { - "description": "Opaque cursor to pass to the next call to continue after the last item.", - "type": [ - "string", - "null" - ] + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemsListCursor" + }, + { + "type": "null" + } + ], + "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, "limit": { "description": "Optional item page size.", diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 862544637b..fc37d8d959 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index ff0114aaed..da8cafb91d 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListAnchor.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListAnchor.ts new file mode 100644 index 0000000000..425b2136bf --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListAnchor.ts @@ -0,0 +1,8 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * An exclusive item position within the requested visible turn. + */ +export type ThreadItemsListAnchor = { "type": "item", itemId: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListCursor.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListCursor.ts new file mode 100644 index 0000000000..dd4c4faec7 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListCursor.ts @@ -0,0 +1,9 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadItemsListAnchor } from "./ThreadItemsListAnchor"; + +/** + * Starting position for an item-history page. + */ +export type ThreadItemsListCursor = string | ThreadItemsListAnchor; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts index c41e93f795..fe2cce685e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts @@ -2,6 +2,7 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { SortDirection } from "./SortDirection"; +import type { ThreadItemsListCursor } from "./ThreadItemsListCursor"; export type ThreadItemsListParams = { threadId: string, /** @@ -9,9 +10,11 @@ export type ThreadItemsListParams = { threadId: string, */ turnId?: string | null, /** - * Opaque cursor to pass to the next call to continue after the last item. + * Opaque continuation cursor or an exclusive item anchor in the requested visible turn. + * An item anchor requires a non-empty `turnId`; ascending (the default) returns items + * after it, and descending returns items before it. Continue with the returned string cursor. */ -cursor?: string | null, +cursor?: ThreadItemsListCursor | null, /** * Optional item page size. */ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 4cb9aa7ed2..d0c9fea6d0 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -514,6 +514,8 @@ export type { ThreadInjectItemsParams } from "./ThreadInjectItemsParams"; export type { ThreadInjectItemsResponse } from "./ThreadInjectItemsResponse"; export type { ThreadItem } from "./ThreadItem"; export type { ThreadItemEntry } from "./ThreadItemEntry"; +export type { ThreadItemsListAnchor } from "./ThreadItemsListAnchor"; +export type { ThreadItemsListCursor } from "./ThreadItemsListCursor"; export type { ThreadItemsListParams } from "./ThreadItemsListParams"; export type { ThreadItemsListResponse } from "./ThreadItemsListResponse"; export type { ThreadListParams } from "./ThreadListParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index ed2f0729d9..5bc54053e4 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -400,7 +400,7 @@ fn thread_items_list_round_trips() { let params = ThreadItemsListParams { thread_id: "thr_123".to_string(), turn_id: Some("turn_456".to_string()), - cursor: Some("cursor_1".to_string()), + cursor: Some(ThreadItemsListCursor::Opaque("cursor_1".to_string())), limit: Some(50), sort_direction: Some(SortDirection::Asc), }; @@ -415,6 +415,26 @@ fn thread_items_list_round_trips() { "sortDirection": "asc", }) ); + for cursor in [ + json!("cursor_1"), + json!({"type": "item", "itemId": "item-1"}), + serde_json::Value::Null, + ] { + let mut value = serde_json::to_value(¶ms).unwrap(); + value["cursor"] = cursor; + let request = serde_json::from_value::(value.clone()).unwrap(); + assert_eq!(serde_json::to_value(request).unwrap(), value); + } + for invalid in [ + json!({"type": "other", "itemId": "item-1"}), + json!({"type": "item"}), + json!({"type": "item", "itemId": 1}), + json!(42), + ] { + let mut value = serde_json::to_value(¶ms).unwrap(); + value["cursor"] = invalid; + assert!(serde_json::from_value::(value).is_err()); + } for (started_at_ms, completed_at_ms) in [ (Some(1_789_855_978_123), Some(1_789_855_979_456)), (Some(1_789_855_978_123), None), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index a1376843a0..2e166d5506 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1741,9 +1741,11 @@ pub struct ThreadItemsListParams { /// Optional turn id to filter by. When omitted, returns items across the thread. #[ts(optional = nullable)] pub turn_id: Option, - /// Opaque cursor to pass to the next call to continue after the last item. + /// Opaque continuation cursor or an exclusive item anchor in the requested visible turn. + /// An item anchor requires a non-empty `turnId`; ascending (the default) returns items + /// after it, and descending returns items before it. Continue with the returned string cursor. #[ts(optional = nullable)] - pub cursor: Option, + pub cursor: Option, /// Optional item page size. #[ts(optional = nullable)] pub limit: Option, @@ -1752,6 +1754,27 @@ pub struct ThreadItemsListParams { pub sort_direction: Option, } +/// Starting position for an item-history page. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(untagged)] +#[ts(export_to = "v2/")] +pub enum ThreadItemsListCursor { + Opaque(String), + Anchor(ThreadItemsListAnchor), +} + +/// An exclusive item position within the requested visible turn. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type", rename_all = "camelCase", export_to = "v2/")] +pub enum ThreadItemsListAnchor { + Item { + #[serde(rename = "itemId")] + #[ts(rename = "itemId")] + item_id: String, + }, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index e9db2b7eab..69efec23d6 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -454,3 +454,19 @@ credential discovery/signing, are disabled while restrictions apply. Supported HTTP, WebSocket, and code-mode gRPC requests use the shared destination checks. User-directed Git, SSH, shell, and other subprocess traffic retain their existing execution and sandbox policies. + +# Item history anchors + +`thread/items/list` accepts an optional nullable `cursor`: either an opaque string +from a previous response or an item anchor such as +`{"type":"item","itemId":"item-123"}`. An item anchor resumes exclusively after +that item in the requested pagination order: ascending (the default) returns newer +items, and descending returns older items. It requires a non-empty `turnId`; +otherwise the request returns invalid params (`-32602`) with +`turnId is required when cursor is an item anchor`. +The item must belong to that turn in the thread's visible history. Empty, unknown, +and out-of-scope item IDs return invalid params (`-32602`) with +`cursor.itemId does not identify an item in the requested history scope`. +Omitted or null cursors preserve normal first-page behavior. Continue anchored +pages with the returned opaque string `nextCursor`; response fields and +`backwardsCursor` semantics are unchanged. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index a00b6eb49c..2f4731ab23 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -485,6 +485,7 @@ 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::ListItemsPosition as StoreListItemsPosition; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::ListTimelineParams as StoreListTimelineParams; 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 055edd1ea6..9679a31578 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -14,6 +14,8 @@ use super::*; use crate::error_code::method_not_found; use codex_app_server_protocol::SelectedCapabilityRoot; use codex_app_server_protocol::ThreadHistoryMode as ApiThreadHistoryMode; +use codex_app_server_protocol::ThreadItemsListAnchor; +use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadRevertParams; use codex_app_server_protocol::ThreadRevertResponse; use codex_app_server_protocol::ThreadRevertedNotification; @@ -3289,7 +3291,7 @@ impl ThreadRequestProcessor { thread_id, turn_id: Some(turn_id.to_string()), include_archived: true, - cursor: cursor.clone(), + position: cursor.clone().map(StoreListItemsPosition::Cursor), page_size: THREAD_ITEMS_MAX_LIMIT, sort_direction: StoreSortDirection::Asc, sort_key: StoreItemSortKey::CreatedAtOrdinal, @@ -3405,7 +3407,7 @@ impl ThreadRequestProcessor { thread_id, turn_id: None, include_archived: true, - cursor: None, + position: None, page_size: 1, sort_direction: StoreSortDirection::Desc, sort_key: StoreItemSortKey::CreatedAtOrdinal, @@ -3427,6 +3429,18 @@ impl ThreadRequestProcessor { limit, sort_direction, } = params; + let position = cursor.map(|cursor| match cursor { + ThreadItemsListCursor::Opaque(cursor) => StoreListItemsPosition::Cursor(cursor), + ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { item_id }) => { + StoreListItemsPosition::ItemAnchor { item_id } + } + }); + let has_anchor = matches!(position, Some(StoreListItemsPosition::ItemAnchor { .. })); + if has_anchor && turn_id.as_deref().is_none_or(str::is_empty) { + return Err(invalid_params( + "turnId is required when cursor is an item anchor", + )); + } let thread_id = ThreadId::from_string(&thread_id) .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; let page_size = limit @@ -3439,7 +3453,7 @@ impl ThreadRequestProcessor { thread_id, turn_id, include_archived: true, - cursor, + position, page_size, sort_direction: match sort_direction.unwrap_or(SortDirection::Asc) { SortDirection::Asc => StoreSortDirection::Asc, @@ -3450,6 +3464,9 @@ impl ThreadRequestProcessor { }) .await .map_err(|err| match err { + ThreadStoreError::InvalidRequest { message } if has_anchor => { + invalid_params(message) + } ThreadStoreError::InvalidRequest { message } => invalid_request(message), ThreadStoreError::Unsupported { .. } => { method_not_found("thread/items/list is not supported yet") 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 06d0ec878d..f8cc0ec24d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -25,6 +25,8 @@ use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadItemEntry; +use codex_app_server_protocol::ThreadItemsListAnchor; +use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadListParams; @@ -2052,6 +2054,84 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( assert_eq!(third_items_page.data[1].turn_id, "turn-2"); assert_eq!(third_items_page.data[1].item.id(), "user-2"); + let anchor_params = ThreadItemsListParams { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".to_string()), + cursor: None, + limit: Some(1), + sort_direction: None, + }; + for direction in [None, Some(SortDirection::Desc)] { + let request = mcp + .send_thread_items_list_request(ThreadItemsListParams { + limit: Some(10), + sort_direction: direction, + ..anchor_params.clone() + }) + .await?; + let ordinary: ThreadItemsListResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request)).await??; + assert_eq!(ordinary.data.len(), 3); + let request = mcp + .send_thread_items_list_request(ThreadItemsListParams { + cursor: Some(ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { + item_id: ordinary.data[0].item.id().to_string(), + })), + sort_direction: direction, + ..anchor_params.clone() + }) + .await?; + let anchored: ThreadItemsListResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request)).await??; + assert_eq!(anchored.data, ordinary.data[1..2]); + let request = mcp + .send_thread_items_list_request(ThreadItemsListParams { + cursor: Some(ThreadItemsListCursor::Opaque( + anchored + .next_cursor + .expect("anchored page has a continuation"), + )), + sort_direction: direction, + ..anchor_params.clone() + }) + .await?; + let continued: ThreadItemsListResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request)).await??; + assert_eq!(continued.data, ordinary.data[2..]); + assert_eq!(continued.next_cursor, None); + } + for (turn_id, anchor, message) in [ + ( + None, + "steer-1", + "turnId is required when cursor is an item anchor", + ), + ( + Some("turn-1"), + "missing", + "cursor.itemId does not identify an item in the requested history scope", + ), + ] { + let request = mcp + .send_thread_items_list_request(ThreadItemsListParams { + turn_id: turn_id.map(str::to_string), + cursor: Some(ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { + item_id: anchor.to_string(), + })), + ..anchor_params.clone() + }) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request)), + ) + .await??; + assert_eq!( + (error.error.code, error.error.message.as_str()), + (-32602, message) + ); + } + let turn_start_id = mcp .send_turn_start_request(TurnStartParams { thread_id: thread_id.to_string(), @@ -2299,7 +2379,7 @@ async fn read_items_page( .send_thread_items_list_request(ThreadItemsListParams { thread_id: thread_id.to_string(), turn_id: turn_id.map(str::to_string), - cursor, + cursor: cursor.map(ThreadItemsListCursor::Opaque), limit, sort_direction: Some(sort_direction), }) diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs index d69414f758..67b4be96de 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_revert.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -17,6 +17,7 @@ use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadResumeParams; @@ -490,7 +491,7 @@ async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { params: ThreadItemsListParams { thread_id: thread.id.clone(), turn_id: None, - cursor: items_backwards_cursor, + cursor: items_backwards_cursor.map(ThreadItemsListCursor::Opaque), limit: None, sort_direction: None, }, diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 9eed3ae9e0..b7a3899260 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -131,7 +131,7 @@ mod tests { thread_id, turn_id: None, include_archived: true, - cursor: None, + position: None, page_size: 10, sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 2650a56f85..26eba76b5a 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -77,6 +77,7 @@ pub use types::GitInfoPatch; pub use types::ItemPage; pub use types::ItemSortKey; pub use types::ListItemsParams; +pub use types::ListItemsPosition; pub use types::ListThreadsParams; pub use types::ListTimelineParams; pub use types::ListTurnsParams; 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 d459944ab0..b52dc927f6 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -454,7 +454,7 @@ async fn migration_projects_explicit_and_implicit_legacy_completed_items() { thread_id, turn_id: None, include_archived: false, - cursor: None, + position: None, page_size: 10, sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, @@ -584,7 +584,7 @@ async fn migration_keeps_late_completions_in_their_original_turn() { thread_id, turn_id: None, include_archived: false, - cursor: None, + position: None, page_size: 10, sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, @@ -1064,7 +1064,7 @@ async fn migration_keeps_late_completions_for_surviving_turns_across_rollback() thread_id, turn_id: None, include_archived: false, - cursor: None, + position: None, page_size: 20, sort_direction: SortDirection::Asc, sort_key: ItemSortKey::CreatedAtOrdinal, 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 5d5ffb66ee..dccd8752e1 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 @@ -23,6 +23,7 @@ use tempfile::TempDir; use super::*; use crate::ItemSortKey; +use crate::ListItemsPosition; use crate::ListTimelineParams; use crate::SearchThreadOccurrencesParams; use crate::SortDirection; @@ -232,6 +233,169 @@ 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 item_anchors_page_exclusively_while_the_turn_grows() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let db = history_db(&store).await; + insert_turn( + db, + thread_id, + "turn", + /*rollout_ordinal*/ 1, + "inProgress", + /*error_json*/ None, + /*first_user_item_id*/ None, + /*final_agent_item_id*/ None, + ) + .await; + for (id, ordinal) in [ + ("one", 2), + ("two", 3), + ("anchor", 4), + ("four", 5), + ("five", 6), + ] { + insert_item(db, thread_id, "turn", id, ordinal).await; + } + insert_item( + db, + thread_id, + "other-turn", + "anchor", + /*rollout_ordinal*/ 8, + ) + .await; + for (direction, expected) in [ + (SortDirection::Asc, vec!["four", "five", "new"]), + (SortDirection::Desc, vec!["two", "one"]), + ] { + let mut params = item_params( + thread_id, + Some("turn"), + /*cursor*/ None, + /*page_size*/ 1, + direction, + ); + params.position = Some(ListItemsPosition::ItemAnchor { + item_id: "anchor".to_string(), + }); + let first = store.list_items(params.clone()).await.expect("anchor page"); + assert_eq!(item_ids(&first), vec![expected[0]]); + if direction == SortDirection::Asc { + insert_item(db, thread_id, "turn", "new", /*rollout_ordinal*/ 7).await; + } + let mut ids = vec![first.items[0].item_id.clone()]; + params.position = first.next_cursor.map(ListItemsPosition::Cursor); + while params.position.is_some() { + let page = store + .list_items(params.clone()) + .await + .expect("continuation"); + ids.extend(page.items.into_iter().map(|item| item.item_id)); + params.position = page.next_cursor.map(ListItemsPosition::Cursor); + } + assert_eq!(ids, expected); + params.position = first.backwards_cursor.map(ListItemsPosition::Cursor); + params.sort_direction = match direction { + SortDirection::Asc => SortDirection::Desc, + SortDirection::Desc => SortDirection::Asc, + }; + let backwards = store.list_items(params).await.expect("reverse page"); + assert_eq!(item_ids(&backwards), vec![expected[0]]); + } + for (id, direction) in [("one", SortDirection::Desc), ("new", SortDirection::Asc)] { + let mut params = item_params( + thread_id, + Some("turn"), + /*cursor*/ None, + /*page_size*/ 2, + direction, + ); + params.position = Some(ListItemsPosition::ItemAnchor { + item_id: id.to_string(), + }); + let page = store.list_items(params).await.expect("edge anchor"); + assert!(page.items.is_empty()); + assert_eq!((page.next_cursor, page.backwards_cursor), (None, None)); + } +} + +#[tokio::test] +async fn item_anchors_require_creation_order_without_an_update_watermark() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + insert_item( + history_db(&store).await, + thread_id, + "turn", + "anchor", + /*rollout_ordinal*/ 1, + ) + .await; + for (sort_key, after_updated_at_ordinal) in [ + (ItemSortKey::UpdatedAtOrdinal, None), + (ItemSortKey::CreatedAtOrdinal, Some(0)), + ] { + let params = ListItemsParams { + position: Some(ListItemsPosition::ItemAnchor { + item_id: "anchor".to_string(), + }), + sort_key, + after_updated_at_ordinal, + ..item_params( + thread_id, + Some("turn"), + /*cursor*/ None, + /*page_size*/ 1, + SortDirection::Asc, + ) + }; + let error = store + .list_items(params) + .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")); + } +} + +#[tokio::test] +async fn item_anchors_reject_items_outside_the_requested_scope() { + let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let db = history_db(&store).await; + insert_item( + db, + thread_id, + "other-turn", + "other-turn-item", + /*rollout_ordinal*/ 2, + ) + .await; + insert_item( + db, + ThreadId::default(), + "turn", + "other-thread-item", + /*rollout_ordinal*/ 2, + ) + .await; + for anchor in ["", "unknown", "other-turn-item", "other-thread-item"] { + let mut params = item_params( + thread_id, + Some("turn"), + /*cursor*/ None, + /*page_size*/ 1, + SortDirection::Asc, + ); + params.position = Some(ListItemsPosition::ItemAnchor { + item_id: anchor.to_string(), + }); + let error = store.list_items(params).await.expect_err("invalid anchor"); + assert!( + matches!(error, ThreadStoreError::InvalidRequest { message } if message == "cursor.itemId does not identify an item in the requested history scope") + ); + } +} + #[tokio::test] async fn timeline_interleaves_items_and_restores_page_boundary_session_state() { let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; @@ -676,7 +840,7 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { assert_eq!(first_page.items, vec![item_2.clone(), item_3.clone()]); for params in [ ListItemsParams { - cursor: creation_page.next_cursor, + position: creation_page.next_cursor.map(ListItemsPosition::Cursor), ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) }, item_params( @@ -696,7 +860,7 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { let second_page = store .list_items(ListItemsParams { - cursor: first_page.next_cursor, + position: first_page.next_cursor.map(ListItemsPosition::Cursor), ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) }) .await @@ -723,7 +887,7 @@ async fn list_items_filters_exclusive_update_ordinals_across_pages_and_turns() { assert_eq!(descending_page.items, vec![item_1, item_3]); let descending_next_page = store .list_items(ListItemsParams { - cursor: descending_page.next_cursor, + position: descending_page.next_cursor.map(ListItemsPosition::Cursor), sort_direction: SortDirection::Desc, ..updated_item_params(thread_id, /*after_updated_at_ordinal*/ 0) }) @@ -992,6 +1156,29 @@ async fn lineage_reads_page_across_parent_and_child_segments() { vec!["root-user", "root-agent"] ); + let mut anchored = item_params( + child_id, + Some("root-1"), + /*cursor*/ None, + /*page_size*/ 2, + SortDirection::Desc, + ); + anchored.position = Some(ListItemsPosition::ItemAnchor { + item_id: "root-agent".to_string(), + }); + let page = store + .list_items(anchored.clone()) + .await + .expect("inherited anchor"); + assert_eq!(item_ids(&page), vec!["root-user"]); + anchored.turn_id = Some("excluded-root".to_string()); + anchored.position = Some(ListItemsPosition::ItemAnchor { + item_id: "excluded-item".to_string(), + }); + assert!( + matches!(store.list_items(anchored).await, Err(ThreadStoreError::InvalidRequest { message }) if message == "cursor.itemId does not identify an item in the requested history scope") + ); + for sort_key in [ItemSortKey::CreatedAtOrdinal, ItemSortKey::UpdatedAtOrdinal] { let error = store .list_items(ListItemsParams { @@ -1498,7 +1685,7 @@ fn item_params( thread_id, turn_id: turn_id.map(str::to_owned), include_archived: false, - cursor, + position: cursor.map(ListItemsPosition::Cursor), page_size, sort_direction, sort_key: ItemSortKey::CreatedAtOrdinal, 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 9d11e31505..e16a6dc48f 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 @@ -19,6 +19,7 @@ use super::read::stored_turn_row; use super::thread_history_error; use crate::ItemSortKey; use crate::ListItemsParams; +use crate::ListItemsPosition; use crate::SortDirection; use crate::StoredTurnItemsView; use crate::ThreadStoreError; @@ -208,6 +209,20 @@ pub(super) async fn page_item_rows( lineage: &RolloutLineage, params: &ListItemsParams, ) -> ThreadStoreResult> { + let (cursor, anchor_item_id) = match params.position.as_ref() { + Some(ListItemsPosition::Cursor(cursor)) => (Some(cursor.as_str()), None), + Some(ListItemsPosition::ItemAnchor { item_id }) => (None, Some(item_id.as_str())), + None => (None, None), + }; + if anchor_item_id.is_some() + && (params.sort_key != ItemSortKey::CreatedAtOrdinal + || params.after_updated_at_ordinal.is_some()) + { + return Err(ThreadStoreError::InvalidRequest { + message: "item anchors require creation-order paging without an update watermark" + .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 { @@ -230,15 +245,55 @@ pub(super) async fn page_item_rows( pool, segment.rollout_id(), params, + cursor, after_updated_at_ordinal, ) .await; } - let cursor = parse_cursor( - params.cursor.as_deref(), - params.thread_id, - CursorScope::ItemsByCreatedAtOrdinal, - )?; + let cursor = if let Some(anchor_item_id) = anchor_item_id { + let turn_id = params + .turn_id + .as_deref() + .filter(|id| !id.is_empty()) + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: "turnId is required when cursor is an item anchor".to_string(), + })?; + if anchor_item_id.is_empty() { + return Err(invalid_item_anchor()); + } + let mut anchor = None; + for segment in lineage.segments() { + let mut query = QueryBuilder::::new( + "SELECT rollout_ordinal FROM thread_items WHERE thread_id = ", + ); + query.push_bind(segment.rollout_id().to_string()); + query.push(" AND item_id = ").push_bind(anchor_item_id); + push_segment_range(&mut query, segment)?; + query.push(" AND turn_id = ").push_bind(turn_id); + if let Some(ordinal) = query + .build_query_scalar::() + .fetch_optional(pool) + .await + .map_err(thread_history_error)? + { + anchor = Some(ordinal); + break; + } + } + Some(HistoryCursor { + requested_thread_id: params.thread_id, + rollout_ordinal: u64::try_from(anchor.ok_or_else(invalid_item_anchor)?) + .map_err(thread_history_error)?, + include_anchor: false, + scope: CursorScope::ItemsByCreatedAtOrdinal, + }) + } else { + parse_cursor( + cursor, + params.thread_id, + CursorScope::ItemsByCreatedAtOrdinal, + )? + }; let mut rows = Vec::new(); for (_, segment, segment_cursor) in segments_from_cursor(lineage, params.sort_direction, cursor.as_ref())? @@ -289,10 +344,11 @@ async fn page_updated_item_rows( pool: &sqlx::SqlitePool, rollout_id: ThreadId, params: &ListItemsParams, + cursor: Option<&str>, after_updated_at_ordinal: u64, ) -> ThreadStoreResult> { let cursor = parse_cursor( - params.cursor.as_deref(), + cursor, params.thread_id, CursorScope::ItemsByUpdatedAtOrdinal, )?; @@ -504,3 +560,10 @@ fn page_size_too_large() -> ThreadStoreError { message: "page size is too large".to_string(), } } + +fn invalid_item_anchor() -> ThreadStoreError { + ThreadStoreError::InvalidRequest { + message: "cursor.itemId does not identify an item in the requested history scope" + .to_string(), + } +} diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index d6c0d40bb4..692cb1029c 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -469,6 +469,16 @@ pub struct TurnPage { pub backwards_cursor: Option, } +/// Starting position for listing persisted items. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +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. + ItemAnchor { item_id: String }, +} + /// Parameters for listing persisted items within a thread. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ListItemsParams { @@ -478,8 +488,8 @@ pub struct ListItemsParams { pub turn_id: Option, /// Whether archived threads are eligible. pub include_archived: bool, - /// Opaque cursor returned by a previous list call. - pub cursor: Option, + /// Optional starting position; omitted for the first page. + pub position: Option, /// Maximum number of items to return. pub page_size: usize, /// Direction to sort items by the selected ordinal. diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index c183e79eca..4232dd9ac5 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -15,6 +15,7 @@ use codex_app_server_protocol::JSONRPCRequest; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadStatus; @@ -2090,7 +2091,7 @@ async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> params: ThreadItemsListParams { thread_id: thread_id.to_string(), turn_id: None, - cursor: Some(cursor.clone()), + cursor: Some(ThreadItemsListCursor::Opaque(cursor.clone())), limit: Some(crate::app_server_session::HISTORY_ITEM_PAGE_LIMIT), sort_direction: Some(SortDirection::Desc), }, diff --git a/codex-rs/tui/src/app_server_session/history.rs b/codex-rs/tui/src/app_server_session/history.rs index 1668111ee1..55ce6daf35 100644 --- a/codex-rs/tui/src/app_server_session/history.rs +++ b/codex-rs/tui/src/app_server_session/history.rs @@ -14,6 +14,7 @@ use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadRevertParams; @@ -107,7 +108,7 @@ pub(crate) fn thread_items_page_params( ThreadItemsListParams { thread_id: thread_id.to_string(), turn_id: turn_id.map(str::to_string), - cursor, + cursor: cursor.map(ThreadItemsListCursor::Opaque), limit: Some(limit), sort_direction: Some(SortDirection::Desc), } diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index ef61063212..7cef53c5a4 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -5455,14 +5455,42 @@ class ContextCompactionThreadItem(BaseModel): type: Annotated[Literal["contextCompaction"], Field(title="ContextCompactionThreadItemType")] +class ItemThreadItemsListAnchor(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + item_id: Annotated[str, Field(alias="itemId")] + type: Annotated[Literal["item"], Field(title="ItemThreadItemsListAnchorType")] + + +class ThreadItemsListAnchor(RootModel[ItemThreadItemsListAnchor]): + model_config = ConfigDict( + populate_by_name=True, + ) + root: Annotated[ + ItemThreadItemsListAnchor, + Field(description="An exclusive item position within the requested visible turn."), + ] + + +class ThreadItemsListCursor(RootModel[str | ThreadItemsListAnchor]): + model_config = ConfigDict( + populate_by_name=True, + ) + root: Annotated[ + str | ThreadItemsListAnchor, + Field(description="Starting position for an item-history page."), + ] + + class ThreadItemsListParams(BaseModel): model_config = ConfigDict( populate_by_name=True, ) cursor: Annotated[ - str | None, + ThreadItemsListCursor | None, Field( - description="Opaque cursor to pass to the next call to continue after the last item." + description="Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." ), ] = None limit: Annotated[int | None, Field(description="Optional item page size.", ge=0)] = None