From 105ab5ba4a5eeb077248f8dd148fe5e830600ece Mon Sep 17 00:00:00 2001 From: Roy Han Date: Tue, 5 May 2026 15:04:02 -0700 Subject: [PATCH] add thread turn item paging --- .../src/protocol/common.rs | 7 + .../app-server-protocol/src/protocol/v2.rs | 32 +++ codex-rs/app-server/README.md | 1 + codex-rs/app-server/src/message_processor.rs | 3 + codex-rs/app-server/src/request_processors.rs | 2 + .../request_processors/thread_processor.rs | 186 ++++++++++++++++++ .../app-server/tests/common/mcp_process.rs | 10 + 7 files changed, 241 insertions(+) diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 5ab2e5ea01..4176b8e2da 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -577,6 +577,13 @@ client_request_definitions! { serialization: None, response: v2::ThreadTurnsListResponse, }, + #[experimental("thread/turns/items/list")] + ThreadTurnsItemsList => "thread/turns/items/list" { + params: v2::ThreadTurnsItemsListParams, + // Explicitly concurrent: this primarily reads append-only rollout storage. + serialization: None, + response: v2::ThreadTurnsItemsListResponse, + }, /// Append raw Responses API items to the thread history without starting a user turn. ThreadInjectItems => "thread/inject_items" { params: v2::ThreadInjectItemsParams, diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 6dd4eaa370..da62956465 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -4694,6 +4694,38 @@ pub struct ThreadTurnsListResponse { pub backwards_cursor: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadTurnsItemsListParams { + pub thread_id: String, + pub turn_id: String, + /// Opaque cursor to pass to the next call to continue after the last item. + #[ts(optional = nullable)] + pub cursor: Option, + /// Optional item page size. + #[ts(optional = nullable)] + pub limit: Option, + /// Optional item pagination direction; defaults to ascending. + #[ts(optional = nullable)] + pub sort_direction: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadTurnsItemsListResponse { + 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. + pub next_cursor: Option, + /// Opaque cursor to pass as `cursor` when reversing `sortDirection`. + /// This is only populated when the page contains at least one item. + /// Use it with the opposite `sortDirection` to include the anchor item again + /// and catch updates to that item. + pub backwards_cursor: Option, +} + #[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 115fe1c1d2..18a9d80821 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -150,6 +150,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`, `nextCursor`, and `backwardsCursor`. +- `thread/turns/items/list` — experimental; page through one stored turn’s items with the same cursor model. - `thread/metadata/update` — patch stored thread metadata in sqlite; currently supports updating persisted `gitInfo` fields and returns the refreshed `thread`. - `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. - `memory/reset` — experimental; clear the current `CODEX_HOME/memories` directory and reset persisted memory stage data in sqlite while preserving existing thread memory modes; returns `{}` on success. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 8d26cec70f..1790563b2b 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1044,6 +1044,9 @@ 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::ThreadShellCommand { params, .. } => { self.thread_processor .thread_shell_command(&request_id, params) diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index be6d559866..75da930f3b 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -209,6 +209,8 @@ 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::ThreadTurnsItemsListResponse; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::ThreadUnarchiveParams; 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 1921fb69b1..3dd3d3426d 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -537,6 +537,15 @@ impl ThreadRequestProcessor { .map(|response| Some(response.into())) } + pub(crate) async fn thread_turns_items_list( + &self, + params: ThreadTurnsItemsListParams, + ) -> Result, JSONRPCErrorError> { + self.thread_turns_items_list_response_inner(params) + .await + .map(|response| Some(response.into())) + } + pub(crate) async fn thread_shell_command( &self, request_id: &ConnectionRequestId, @@ -2075,6 +2084,62 @@ impl ThreadRequestProcessor { }) } + async fn thread_turns_items_list_response_inner( + &self, + params: ThreadTurnsItemsListParams, + ) -> Result { + let ThreadTurnsItemsListParams { + thread_id, + turn_id, + cursor, + limit, + sort_direction, + } = params; + + let thread_uuid = ThreadId::from_string(&thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + let items = self + .load_thread_turns_list_history(thread_uuid) + .await + .map_err(thread_read_view_error)?; + let loaded_thread = self.thread_manager.get_thread(thread_uuid).await.ok(); + let has_live_running_thread = match loaded_thread.as_ref() { + Some(thread) => matches!(thread.agent_status().await, AgentStatus::Running), + None => false, + }; + let active_turn = if loaded_thread.is_some() { + let thread_state = self.thread_state_manager.thread_state(thread_uuid).await; + let state = thread_state.lock().await; + state.active_turn_snapshot() + } else { + None + }; + let turns = reconstruct_thread_turns_for_turns_list( + &items, + self.thread_watch_manager + .loaded_status_for_thread(&thread_uuid.to_string()) + .await, + has_live_running_thread, + active_turn, + ); + let turn_items = turns + .into_iter() + .find(|turn| turn.id == turn_id) + .ok_or_else(|| invalid_request(format!("turn not found: {turn_id}")))? + .items; + let page = paginate_thread_items( + turn_items, + cursor.as_deref(), + limit, + sort_direction.unwrap_or(SortDirection::Asc), + )?; + Ok(ThreadTurnsItemsListResponse { + data: page.items, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) + } + async fn load_thread_turns_list_history( &self, thread_id: ThreadId, @@ -3258,6 +3323,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 = 100; +const THREAD_ITEMS_MAX_LIMIT: usize = 500; fn thread_backwards_cursor_for_sort_key( thread: &StoredThread, @@ -3283,6 +3350,12 @@ struct ThreadTurnsPage { pub(super) backwards_cursor: Option, } +struct ThreadItemsPage { + pub(super) items: Vec, + pub(super) next_cursor: Option, + pub(super) backwards_cursor: Option, +} + #[derive(serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] struct ThreadTurnsCursor { @@ -3290,6 +3363,13 @@ struct ThreadTurnsCursor { include_anchor: bool, } +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct ThreadItemsCursor { + item_id: String, + include_anchor: bool, +} + fn paginate_thread_turns( turns: Vec, cursor: Option<&str>, @@ -3384,6 +3464,112 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result, + cursor: Option<&str>, + limit: Option, + sort_direction: SortDirection, +) -> Result { + if items.is_empty() { + return Ok(ThreadItemsPage { + items: Vec::new(), + next_cursor: None, + backwards_cursor: None, + }); + } + + let anchor = cursor.map(parse_thread_items_cursor).transpose()?; + let page_size = limit + .map(|value| value as usize) + .unwrap_or(THREAD_ITEMS_DEFAULT_LIMIT) + .clamp(1, THREAD_ITEMS_MAX_LIMIT); + + let anchor_index = anchor + .as_ref() + .and_then(|anchor| items.iter().position(|item| item.id() == anchor.item_id)); + if anchor.is_some() && anchor_index.is_none() { + return Err(JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: "invalid cursor: anchor item is no longer present".to_string(), + data: None, + }); + } + + let mut keyed_items: Vec<_> = items.into_iter().enumerate().collect(); + match sort_direction { + SortDirection::Asc => { + if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { + keyed_items.retain(|(index, _)| { + if anchor.include_anchor { + *index >= anchor_index + } else { + *index > anchor_index + } + }); + } + } + SortDirection::Desc => { + keyed_items.reverse(); + if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { + keyed_items.retain(|(index, _)| { + if anchor.include_anchor { + *index <= anchor_index + } else { + *index < anchor_index + } + }); + } + } + } + + let more_items_available = keyed_items.len() > page_size; + keyed_items.truncate(page_size); + let backwards_cursor = keyed_items + .first() + .map(|(_, item)| serialize_thread_items_cursor(item.id(), /*include_anchor*/ true)) + .transpose()?; + let next_cursor = if more_items_available { + keyed_items + .last() + .map(|(_, item)| { + serialize_thread_items_cursor(item.id(), /*include_anchor*/ false) + }) + .transpose()? + } else { + None + }; + let items = keyed_items.into_iter().map(|(_, item)| item).collect(); + + Ok(ThreadItemsPage { + items, + next_cursor, + backwards_cursor, + }) +} + +fn serialize_thread_items_cursor( + item_id: &str, + include_anchor: bool, +) -> Result { + serde_json::to_string(&ThreadItemsCursor { + item_id: item_id.to_string(), + include_anchor, + }) + .map_err(|err| JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to serialize cursor: {err}"), + data: None, + }) +} + +fn parse_thread_items_cursor(cursor: &str) -> Result { + serde_json::from_str(cursor).map_err(|_| JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid cursor: {cursor}"), + data: None, + }) +} + fn reconstruct_thread_turns_for_turns_list( items: &[RolloutItem], loaded_status: ThreadStatus, diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index c2a49d8fa6..3d508b618f 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -74,6 +74,7 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadInjectItemsParams; +use codex_app_server_protocol::ThreadTurnsItemsListParams; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadMemoryModeSetParams; @@ -522,6 +523,15 @@ impl McpProcess { 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( + &mut self, + params: ThreadTurnsItemsListParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/turns/items/list", params).await + } + /// Send a `model/list` JSON-RPC request. pub async fn send_list_models_request( &mut self,