From 1882719b30b01b50145fbca3772fb1537ed4bb2d Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 23 Jun 2026 13:57:08 -0700 Subject: [PATCH] feat(app-server): thread/turns/items/list -> thread/items/list (#29705) ## Description Rename the experimental app-server item pagination API from `thread/turns/items/list` to `thread/items/list` and make `turnId` optional. Clients can now page persisted items across a thread, or still filter to one turn when needed. ## What changed - Rename the request/response protocol types and JSON-RPC method to `ThreadItemsList*` / `thread/items/list`. - Pass optional `turnId` through to `ThreadStore::list_items`. - Update app-server docs and focused protocol/app-server tests. ## Validation - `just test -p codex-app-server-protocol thread_items_list_round_trips` - `just test -p codex-app-server thread_items_list_returns_unsupported` --- .../src/protocol/common.rs | 16 ++-- .../src/protocol/v2/tests.rs | 34 ++++++++- .../src/protocol/v2/thread.rs | 8 +- codex-rs/app-server/README.md | 8 +- codex-rs/app-server/src/message_processor.rs | 4 +- codex-rs/app-server/src/request_processors.rs | 4 +- .../request_processors/thread_processor.rs | 74 +++++++++++++++++-- .../tests/common/test_app_server.rs | 10 +-- .../app-server/tests/suite/v2/thread_read.rs | 12 +-- 9 files changed, 132 insertions(+), 38 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 7fe20a84d..51a37fbaa 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -641,12 +641,12 @@ client_request_definitions! { serialization: None, response: v2::ThreadTurnsListResponse, }, - #[experimental("thread/turns/items/list")] - ThreadTurnsItemsList => "thread/turns/items/list" { - params: v2::ThreadTurnsItemsListParams, + #[experimental("thread/items/list")] + ThreadItemsList => "thread/items/list" { + params: v2::ThreadItemsListParams, // Explicitly concurrent: this primarily reads append-only rollout storage. serialization: None, - response: v2::ThreadTurnsItemsListResponse, + response: v2::ThreadItemsListResponse, }, /// Append raw Responses API items to the thread history without starting a user turn. ThreadInjectItems => "thread/inject_items" { @@ -2090,17 +2090,17 @@ mod tests { }; assert_eq!(thread_turns_list.serialization_scope(), None); - let thread_turns_items_list = ClientRequest::ThreadTurnsItemsList { + let thread_items_list = ClientRequest::ThreadItemsList { request_id: request_id(), - params: v2::ThreadTurnsItemsListParams { + params: v2::ThreadItemsListParams { thread_id: "thread-1".to_string(), - turn_id: "turn-1".to_string(), + turn_id: None, cursor: None, limit: None, sort_direction: None, }, }; - assert_eq!(thread_turns_items_list.serialization_scope(), None); + assert_eq!(thread_items_list.serialization_scope(), None); let mcp_resource_read = ClientRequest::McpResourceRead { request_id: request_id(), 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 85a95a272..7c78fee4c 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -224,10 +224,10 @@ fn thread_resume_response_round_trips_initial_turns_page() { } #[test] -fn thread_turns_items_list_round_trips() { - let params = ThreadTurnsItemsListParams { +fn thread_items_list_round_trips() { + let params = ThreadItemsListParams { thread_id: "thr_123".to_string(), - turn_id: "turn_456".to_string(), + turn_id: Some("turn_456".to_string()), cursor: Some("cursor_1".to_string()), limit: Some(50), sort_direction: Some(SortDirection::Asc), @@ -243,7 +243,7 @@ fn thread_turns_items_list_round_trips() { "sortDirection": "asc", }) ); - let response = ThreadTurnsItemsListResponse { + let response = ThreadItemsListResponse { data: vec![ThreadItem::ContextCompaction { id: "item_1".to_string(), }], @@ -259,6 +259,32 @@ fn thread_turns_items_list_round_trips() { "backwardsCursor": "cursor_0", }) ); + + let params_without_turn = ThreadItemsListParams { + thread_id: "thr_123".to_string(), + turn_id: None, + cursor: None, + limit: None, + sort_direction: None, + }; + + assert_eq!( + serde_json::to_value(¶ms_without_turn).expect("serialize params without turn"), + json!({ + "threadId": "thr_123", + "turnId": null, + "cursor": null, + "limit": null, + "sortDirection": null, + }) + ); + assert_eq!( + serde_json::from_value::(json!({ + "threadId": "thr_123", + })) + .expect("deserialize params without turn"), + params_without_turn + ); } #[test] 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 6a51f71e5..cba4f1b24 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1316,9 +1316,11 @@ pub struct ThreadTurnsListResponse { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] -pub struct ThreadTurnsItemsListParams { +pub struct ThreadItemsListParams { pub thread_id: String, - pub turn_id: String, + /// 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. #[ts(optional = nullable)] pub cursor: Option, @@ -1333,7 +1335,7 @@ pub struct ThreadTurnsItemsListParams { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] -pub struct ThreadTurnsItemsListResponse { +pub struct ThreadItemsListResponse { pub data: Vec, /// Opaque cursor to pass to the next call to continue after the last item. /// if None, there are no more items to return. diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 50f005209..514b82b72 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -145,7 +145,7 @@ Example with notification opt-out: - `thread/loaded/list` — list the thread ids currently loaded in memory. - `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. - `thread/turns/list` — experimental; page through a stored thread’s turn history without resuming it; supports cursor-based pagination with `sortDirection`, `itemsView`, `nextCursor`, and `backwardsCursor`. -- `thread/turns/items/list` — experimental; reserved for paging full items for one turn. The API shape is present, but app-server currently returns an unsupported-method JSON-RPC error. +- `thread/items/list` — experimental; page through persisted thread items without resuming the thread. Pass `turnId` to restrict results to one turn, or omit it to page items across the thread. The active thread store must support item pagination. - `thread/metadata/update` — patch stored thread metadata in sqlite; currently supports updating persisted `gitInfo` fields and returns the refreshed `thread`. - `thread/settings/update` — experimental; queue a partial update to a loaded thread’s next-turn settings without starting a turn or adding transcript items. Omitted fields leave settings unchanged; `serviceTier: null` clears the tier; `multiAgentMode` selects `none`, `explicitRequestOnly`, or `proactive` for subsequent turns; `sandboxPolicy` and `permissions` cannot be combined. Returns `{}` when the update is accepted and emits `thread/settings/updated` with the full effective settings only if they actually change. `turn/start` settings overrides emit the same notification when they change the stored settings. - `thread/memoryMode/set` — experimental; set a thread’s persisted memory eligibility to `"enabled"` or `"disabled"` for either a loaded thread or a stored rollout; returns `{}` on success. @@ -514,10 +514,10 @@ Every returned `Turn` includes `itemsView`, which tells clients whether the `ite } } ``` -`thread/turns/items/list` is the planned hydration API for fetching full items for one turn: +`thread/items/list` pages full persisted items across a thread, optionally filtered to one turn: ```json -{ "method": "thread/turns/items/list", "id": 25, "params": { +{ "method": "thread/items/list", "id": 25, "params": { "threadId": "thr_123", "turnId": "turn_456", "limit": 100, @@ -525,7 +525,7 @@ Every returned `Turn` includes `itemsView`, which tells clients whether the `ite } } ``` -This method currently returns JSON-RPC `-32601` with message `thread/turns/items/list is not supported yet`. +Omit `turnId` or pass `null` to page items across the thread. Thread stores that do not implement item pagination return JSON-RPC `-32601` with message `thread/items/list is not supported yet`. ### Example: Update stored thread metadata diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 33cf142fe..17f34a7b2 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1223,8 +1223,8 @@ impl MessageProcessor { ClientRequest::ThreadTurnsList { params, .. } => { self.thread_processor.thread_turns_list(params).await } - ClientRequest::ThreadTurnsItemsList { params, .. } => { - self.thread_processor.thread_turns_items_list(params).await + ClientRequest::ThreadItemsList { params, .. } => { + self.thread_processor.thread_items_list(params).await } ClientRequest::ThreadShellCommand { params, .. } => { self.thread_processor diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 35df59d5c..509cbeb2e 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -211,6 +211,8 @@ use codex_app_server_protocol::ThreadIncrementElicitationResponse; use codex_app_server_protocol::ThreadInjectItemsParams; use codex_app_server_protocol::ThreadInjectItemsResponse; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadItemsListParams; +use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadListCwdFilter; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; @@ -256,7 +258,6 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; use codex_app_server_protocol::ThreadStatus; -use codex_app_server_protocol::ThreadTurnsItemsListParams; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::ThreadUnarchiveParams; @@ -436,6 +437,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::ListItemsParams as StoreListItemsParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams; 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 89ef93f63..a2966243c 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -683,13 +683,13 @@ impl ThreadRequestProcessor { .map(|response| Some(response.into())) } - pub(crate) async fn thread_turns_items_list( + pub(crate) async fn thread_items_list( &self, - _params: ThreadTurnsItemsListParams, + params: ThreadItemsListParams, ) -> Result, JSONRPCErrorError> { - Err(method_not_found( - "thread/turns/items/list is not supported yet", - )) + self.thread_items_list_response_inner(params) + .await + .map(|response| Some(response.into())) } pub(crate) async fn thread_shell_command( @@ -2388,6 +2388,68 @@ impl ThreadRequestProcessor { ) } + async fn thread_items_list_response_inner( + &self, + params: ThreadItemsListParams, + ) -> Result { + let ThreadItemsListParams { + thread_id, + turn_id, + cursor, + limit, + sort_direction, + } = params; + let thread_id = ThreadId::from_string(&thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + let page_size = limit + .map(|value| value as usize) + .unwrap_or(THREAD_ITEMS_DEFAULT_LIMIT) + .clamp(1, THREAD_ITEMS_MAX_LIMIT); + let page = self + .thread_store + .list_items(StoreListItemsParams { + thread_id, + turn_id, + include_archived: true, + cursor, + page_size, + sort_direction: match sort_direction.unwrap_or(SortDirection::Asc) { + SortDirection::Asc => StoreSortDirection::Asc, + SortDirection::Desc => StoreSortDirection::Desc, + }, + }) + .await + .map_err(|err| match err { + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::Unsupported { .. } => { + method_not_found("thread/items/list is not supported yet") + } + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("no rollout found for thread id {thread_id}")) + } + err => internal_error(format!("failed to list thread items: {err}")), + })?; + let data = + page.items + .into_iter() + .map(|item| { + serde_json::from_slice::(&item.materialized_thread_item_json) + .map_err(|err| { + internal_error(format!( + "failed to deserialize stored thread item {}: {err}", + item.item_key + )) + }) + }) + .collect::, _>>()?; + + Ok(ThreadItemsListResponse { + data, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) + } + async fn load_thread_turns_list_history( &self, thread_id: ThreadId, @@ -3714,6 +3776,8 @@ fn xcode_26_4_mcp_elicitations_auto_deny( const THREAD_TURNS_DEFAULT_LIMIT: usize = 25; const THREAD_TURNS_MAX_LIMIT: usize = 100; +const THREAD_ITEMS_DEFAULT_LIMIT: usize = 25; +const THREAD_ITEMS_MAX_LIMIT: usize = 100; fn thread_backwards_cursor_for_sort_key( thread: &StoredThread, diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index e940d9a42..8c2efa35e 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -84,6 +84,7 @@ use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadDeleteParams; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadInjectItemsParams; +use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadMemoryModeSetParams; @@ -102,7 +103,6 @@ use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSettingsUpdateParams; use codex_app_server_protocol::ThreadShellCommandParams; use codex_app_server_protocol::ThreadStartParams; -use codex_app_server_protocol::ThreadTurnsItemsListParams; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadUnarchiveParams; use codex_app_server_protocol::ThreadUnsubscribeParams; @@ -602,13 +602,13 @@ impl TestAppServer { self.send_request("thread/turns/list", params).await } - /// Send a `thread/turns/items/list` JSON-RPC request. - pub async fn send_thread_turns_items_list_request( + /// Send a `thread/items/list` JSON-RPC request. + pub async fn send_thread_items_list_request( &mut self, - params: ThreadTurnsItemsListParams, + params: ThreadItemsListParams, ) -> anyhow::Result { let params = Some(serde_json::to_value(params)?); - self.send_request("thread/turns/items/list", params).await + self.send_request("thread/items/list", params).await } /// Send a `model/list` JSON-RPC request. 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 58990aaf0..76239b4e4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -19,6 +19,7 @@ use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadNameUpdatedNotification; @@ -32,7 +33,6 @@ use codex_app_server_protocol::ThreadSetNameResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; -use codex_app_server_protocol::ThreadTurnsItemsListParams; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::TurnItemsView; @@ -1137,7 +1137,7 @@ async fn thread_turns_list_rejects_unmaterialized_loaded_thread() -> Result<()> } #[tokio::test] -async fn thread_turns_items_list_returns_unsupported() -> Result<()> { +async fn thread_items_list_returns_unsupported() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -1146,9 +1146,9 @@ async fn thread_turns_items_list_returns_unsupported() -> Result<()> { timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; let read_id = mcp - .send_thread_turns_items_list_request(ThreadTurnsItemsListParams { - thread_id: "thr_123".to_string(), - turn_id: "turn_456".to_string(), + .send_thread_items_list_request(ThreadItemsListParams { + thread_id: "00000000-0000-4000-8000-000000000123".to_string(), + turn_id: None, cursor: None, limit: None, sort_direction: None, @@ -1163,7 +1163,7 @@ async fn thread_turns_items_list_returns_unsupported() -> Result<()> { assert_eq!(read_err.error.code, -32601); assert_eq!( read_err.error.message, - "thread/turns/items/list is not supported yet" + "thread/items/list is not supported yet" ); Ok(())