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 7c974ae4ae..b841d5243c 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 @@ -18787,6 +18787,22 @@ } ] }, + "ThreadItemEntry": { + "properties": { + "item": { + "$ref": "#/definitions/v2/ThreadItem" + }, + "turnId": { + "description": "Turn containing this item.", + "type": "string" + } + }, + "required": [ + "item", + "turnId" + ], + "type": "object" + }, "ThreadListCwdFilter": { "anyOf": [ { 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 7c5d0060ae..0d05498df3 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 @@ -16566,6 +16566,22 @@ } ] }, + "ThreadItemEntry": { + "properties": { + "item": { + "$ref": "#/definitions/ThreadItem" + }, + "turnId": { + "description": "Turn containing this item.", + "type": "string" + } + }, + "required": [ + "item", + "turnId" + ], + "type": "object" + }, "ThreadListCwdFilter": { "anyOf": [ { diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts new file mode 100644 index 0000000000..c59564f265 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts @@ -0,0 +1,10 @@ +// 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 { ThreadItem } from "./ThreadItem"; + +export type ThreadItemEntry = { +/** + * Turn containing this item. + */ +turnId: string, item: ThreadItem, }; 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 2cfb40140c..835ca745d4 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -425,6 +425,7 @@ export type { ThreadHistoryMode } from "./ThreadHistoryMode"; export type { ThreadInjectItemsParams } from "./ThreadInjectItemsParams"; export type { ThreadInjectItemsResponse } from "./ThreadInjectItemsResponse"; export type { ThreadItem } from "./ThreadItem"; +export type { ThreadItemEntry } from "./ThreadItemEntry"; export type { ThreadListParams } from "./ThreadListParams"; export type { ThreadListResponse } from "./ThreadListResponse"; export type { ThreadLoadedListParams } from "./ThreadLoadedListParams"; 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 5c675097d7..c0df9d92ad 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -256,8 +256,11 @@ fn thread_items_list_round_trips() { }) ); let response = ThreadItemsListResponse { - data: vec![ThreadItem::ContextCompaction { - id: "item_1".to_string(), + data: vec![ThreadItemEntry { + turn_id: "turn_456".to_string(), + item: ThreadItem::ContextCompaction { + id: "item_1".to_string(), + }, }], next_cursor: None, backwards_cursor: Some("cursor_0".to_string()), @@ -266,7 +269,10 @@ fn thread_items_list_round_trips() { assert_eq!( serde_json::to_value(&response).expect("serialize response"), json!({ - "data": [{"type": "contextCompaction", "id": "item_1"}], + "data": [{ + "turnId": "turn_456", + "item": {"type": "contextCompaction", "id": "item_1"}, + }], "nextCursor": null, "backwardsCursor": "cursor_0", }) 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 2a22b76993..d1a734aee1 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1353,11 +1353,20 @@ pub struct ThreadItemsListParams { pub sort_direction: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadItemEntry { + /// Turn containing this item. + pub turn_id: String, + pub item: ThreadItem, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadItemsListResponse { - pub data: Vec, + 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, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 489d0864a9..5fa96cac00 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -502,6 +502,8 @@ Use `thread/turns/list` with `capabilities.experimentalApi = true` to page a sto Every returned `Turn` includes `itemsView`, which tells clients whether the `items` array was omitted intentionally (`notLoaded`), contains only summary items (`summary`), or contains every item available from persisted app-server history (`full`). Pass `itemsView` to choose the returned detail level; omitted `itemsView` defaults to `"summary"`. +Paginated threads support `summary` and `notLoaded`; `full` returns an invalid-request error for them, so use `thread/items/list` to page complete items. + ```json { "method": "thread/turns/list", "id": 24, "params": { "threadId": "thr_123", @@ -527,7 +529,10 @@ Every returned `Turn` includes `itemsView`, which tells clients whether the `ite } } ``` -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`. +Each returned entry includes the containing `turnId` and its full `item`, so clients can group +unfiltered pages into turns. 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/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 53bcaa2dd0..86c5af95e8 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -222,6 +222,7 @@ 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::ThreadItemEntry; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadListCwdFilter; @@ -457,12 +458,16 @@ 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::ListTurnsParams as StoreListTurnsParams; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams; use codex_thread_store::ReadThreadParams as StoreReadThreadParams; use codex_thread_store::SearchThreadsParams as StoreSearchThreadsParams; use codex_thread_store::SortDirection as StoreSortDirection; use codex_thread_store::StoredThread; +use codex_thread_store::StoredTurn; +use codex_thread_store::StoredTurnItemsView; +use codex_thread_store::StoredTurnStatus; use codex_thread_store::ThreadMetadataPatch as StoreThreadMetadataPatch; use codex_thread_store::ThreadRelationFilter as StoreThreadRelationFilter; use codex_thread_store::ThreadSortKey as StoreThreadSortKey; 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 60481c60f4..376b7d6332 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2429,6 +2429,38 @@ impl ThreadRequestProcessor { } = params; let thread_uuid = ThreadId::from_string(&thread_id) .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + match self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id: thread_uuid, + include_archived: true, + include_history: false, + }) + .await + { + Ok(thread) if thread.history_mode == ThreadHistoryMode::Paginated => { + return self + .paginated_thread_turns_list_response( + thread_uuid, + cursor, + limit, + sort_direction, + items_view, + ) + .await; + } + Ok(_) => {} + Err(ThreadStoreError::InvalidRequest { message }) + if message == format!("no rollout found for thread id {thread_uuid}") => {} + Err(ThreadStoreError::ThreadNotFound { thread_id }) if thread_id == thread_uuid => {} + Err(ThreadStoreError::InvalidRequest { message }) => { + return Err(invalid_request(message)); + } + Err(ThreadStoreError::Unsupported { operation }) => { + return Err(unsupported_thread_store_operation(operation)); + } + Err(err) => return Err(internal_error(format!("failed to read thread: {err}"))), + } let items = self .load_thread_turns_list_history(thread_uuid) @@ -2470,6 +2502,76 @@ impl ThreadRequestProcessor { ) } + async fn paginated_thread_turns_list_response( + &self, + thread_id: ThreadId, + cursor: Option, + limit: Option, + sort_direction: Option, + items_view: Option, + ) -> Result { + let items_view = items_view.unwrap_or(TurnItemsView::Summary); + let page_size = limit + .map(|value| value as usize) + .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) + .clamp(1, THREAD_TURNS_MAX_LIMIT); + let sort_direction = match sort_direction.unwrap_or(SortDirection::Desc) { + SortDirection::Asc => StoreSortDirection::Asc, + SortDirection::Desc => StoreSortDirection::Desc, + }; + let stored_items_view = match items_view { + TurnItemsView::NotLoaded => StoredTurnItemsView::NotLoaded, + TurnItemsView::Summary => StoredTurnItemsView::Summary, + TurnItemsView::Full => { + return Err(invalid_request( + "thread/turns/list itemsView full is not supported for paginated threads; use thread/items/list", + )); + } + }; + let page = self + .thread_store + .list_turns(StoreListTurnsParams { + thread_id, + include_archived: true, + cursor, + page_size, + sort_direction, + items_view: stored_items_view, + }) + .await + .map_err(|err| match err { + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::Unsupported { operation } => { + unsupported_thread_store_operation(operation) + } + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("no rollout found for thread id {thread_id}")) + } + err => internal_error(format!("failed to list thread history: {err}")), + })?; + let mut turns = Vec::with_capacity(page.turns.len()); + for turn in page.turns { + turns.push(stored_turn_to_api_turn(turn, items_view)?); + } + let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok(); + let has_live_running_thread = match loaded_thread.as_ref() { + Some(thread) => matches!(thread.agent_status().await, AgentStatus::Running), + None => false, + }; + normalize_thread_turns_status( + &mut turns, + self.thread_watch_manager + .loaded_status_for_thread(&thread_id.to_string()) + .await, + has_live_running_thread, + ); + Ok(ThreadTurnsListResponse { + data: turns, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) + } + async fn thread_items_list_response_inner( &self, params: ThreadItemsListParams, @@ -2514,13 +2616,10 @@ impl ThreadRequestProcessor { let data = page .items .into_iter() - .map(|item| { - serde_json::from_slice::(&item.item_json).map_err(|err| { - internal_error(format!( - "failed to deserialize stored thread item {}: {err}", - item.item_id - )) - }) + .map(|stored_item| { + let turn_id = stored_item.turn_id.clone(); + let item = deserialize_stored_thread_item(stored_item)?; + Ok(ThreadItemEntry { turn_id, item }) }) .collect::, _>>()?; @@ -4162,6 +4261,49 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { } } +fn deserialize_stored_thread_item( + item: codex_thread_store::StoredThreadItem, +) -> Result { + serde_json::from_slice::(&item.item_json).map_err(|err| { + internal_error(format!( + "failed to deserialize stored thread item {}: {err}", + item.item_id + )) + }) +} + +fn stored_turn_to_api_turn( + turn: StoredTurn, + items_view: TurnItemsView, +) -> Result { + let status = match turn.status { + StoredTurnStatus::Completed => TurnStatus::Completed, + StoredTurnStatus::Interrupted => TurnStatus::Interrupted, + StoredTurnStatus::Failed => TurnStatus::Failed, + StoredTurnStatus::InProgress => TurnStatus::InProgress, + }; + let error = turn.error.map(|error| TurnError { + message: error.message, + codex_error_info: error.codex_error_info, + additional_details: error.additional_details, + }); + let items = turn + .items + .into_iter() + .map(deserialize_stored_thread_item) + .collect::, _>>()?; + Ok(Turn { + id: turn.turn_id, + items, + items_view, + status, + error, + started_at: turn.started_at, + completed_at: turn.completed_at, + duration_ms: turn.duration_ms, + }) +} + pub(super) fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError { method_not_found(format!("{operation} 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 4162e7ebd6..191d8f9b69 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -21,6 +21,7 @@ use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadItemsListParams; +use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadNameUpdatedNotification; @@ -36,6 +37,7 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; +use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; @@ -48,18 +50,27 @@ use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_core::config::ConfigBuilder; use codex_exec_server::EnvironmentManager; use codex_feedback::CodexFeedback; +use codex_protocol::items::AgentMessageContent; +use codex_protocol::items::AgentMessageItem; +use codex_protocol::items::TurnItem as CoreTurnItem; +use codex_protocol::items::UserMessageItem; use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionSource as ProtocolSessionSource; use codex_protocol::protocol::ThreadMemoryMode; +use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use codex_thread_store::AppendThreadItemsParams; use codex_thread_store::CreateThreadParams; use codex_thread_store::InMemoryThreadStore; +use codex_thread_store::LocalThreadStore; +use codex_thread_store::LocalThreadStoreConfig; use codex_thread_store::ThreadMetadataPatch; use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadStore; @@ -206,7 +217,7 @@ async fn thread_read_can_include_turns() -> Result<()> { } #[tokio::test] -async fn paginated_stored_thread_allows_metadata_discovery_and_rejects_legacy_history_paths() +async fn paginated_stored_thread_routes_projected_turns_and_rejects_legacy_history_paths() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; @@ -299,12 +310,18 @@ async fn paginated_stored_thread_allows_metadata_discovery_and_rejects_legacy_hi items_view: None, }) .await?; - assert_paginated_threads_unsupported( - timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(turns_list_id)), - ) - .await??, + let turns_list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turns_list_id)), + ) + .await??; + assert_eq!( + to_response::(turns_list_resp)?, + ThreadTurnsListResponse { + data: Vec::new(), + next_cursor: None, + backwards_cursor: None, + } ); let resume_id = mcp @@ -1303,6 +1320,202 @@ async fn thread_turns_list_rejects_unmaterialized_loaded_thread() -> Result<()> Ok(()) } +#[tokio::test] +async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let thread_id = codex_protocol::ThreadId::default(); + let state_db = codex_state::StateRuntime::init( + codex_home.path().to_path_buf(), + "mock_provider".to_string(), + ) + .await?; + let store = LocalThreadStore::new( + LocalThreadStoreConfig { + codex_home: codex_home.path().to_path_buf(), + sqlite_home: codex_home.path().to_path_buf(), + default_model_provider_id: "mock_provider".to_string(), + }, + Some(state_db), + ); + store + .create_thread(CreateThreadParams { + session_id: thread_id.into(), + thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: ProtocolSessionSource::Cli, + thread_source: None, + originator: "test_originator".to_string(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + selected_capability_roots: Vec::new(), + multi_agent_version: None, + history_mode: codex_protocol::protocol::ThreadHistoryMode::Paginated, + initial_window_id: Uuid::now_v7().to_string(), + metadata: ThreadPersistenceMetadata { + cwd: Some(codex_home.path().to_path_buf()), + model_provider: "mock_provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await?; + store.persist_thread(thread_id).await?; + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + paginated_turn_started("turn-1"), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::AgentMessage(AgentMessageItem { + id: "agent-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "first".to_string(), + }], + phase: None, + memory_citation: None, + }), + ), + paginated_turn_completed("turn-1"), + paginated_turn_started("turn-2"), + paginated_completed_item( + thread_id, + "turn-2", + CoreTurnItem::UserMessage(UserMessageItem { + id: "user-2".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + ], + }) + .await?; + store.shutdown_thread(thread_id).await?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let first_page = read_turns_page( + &mut mcp, + thread_id, + /*cursor*/ None, + Some(1), + Some(TurnItemsView::Summary), + ) + .await?; + assert_eq!( + first_page.data, + vec![Turn { + id: "turn-1".to_string(), + items: vec![ + ThreadItem::UserMessage { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }, + ThreadItem::AgentMessage { + id: "agent-1".to_string(), + text: "first".to_string(), + phase: None, + memory_citation: None, + }, + ], + items_view: TurnItemsView::Summary, + status: TurnStatus::Completed, + error: None, + started_at: Some(10), + completed_at: Some(20), + duration_ms: Some(10_000), + }] + ); + let next_cursor = first_page.next_cursor.expect("next turn cursor"); + let second_page = read_turns_page( + &mut mcp, + thread_id, + Some(next_cursor), + Some(1), + Some(TurnItemsView::NotLoaded), + ) + .await?; + assert_eq!( + second_page.data, + vec![Turn { + id: "turn-2".to_string(), + items: Vec::new(), + items_view: TurnItemsView::NotLoaded, + status: TurnStatus::Interrupted, + error: None, + started_at: Some(10), + completed_at: None, + duration_ms: None, + }] + ); + + let full_request_id = mcp + .send_thread_turns_list_request(ThreadTurnsListParams { + thread_id: thread_id.to_string(), + cursor: None, + limit: Some(1), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::Full), + }) + .await?; + let full_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(full_request_id)), + ) + .await??; + assert_eq!(full_err.error.code, -32600); + assert_eq!( + full_err.error.message, + "thread/turns/list itemsView full is not supported for paginated threads; use thread/items/list" + ); + + let first_items_page = read_items_page(&mut mcp, thread_id, /*cursor*/ None, Some(1)).await?; + assert_eq!(first_items_page.data.len(), 1); + assert_eq!(first_items_page.data[0].turn_id, "turn-1"); + assert_eq!(first_items_page.data[0].item.id(), "user-1"); + let second_items_page = read_items_page( + &mut mcp, + thread_id, + Some(first_items_page.next_cursor.expect("next item cursor")), + Some(1), + ) + .await?; + assert_eq!(second_items_page.data.len(), 1); + assert_eq!(second_items_page.data[0].turn_id, "turn-1"); + assert_eq!(second_items_page.data[0].item.id(), "agent-1"); + let third_items_page = read_items_page( + &mut mcp, + thread_id, + Some(second_items_page.next_cursor.expect("next item cursor")), + Some(1), + ) + .await?; + assert_eq!(third_items_page.data.len(), 1); + assert_eq!(third_items_page.data[0].turn_id, "turn-2"); + assert_eq!(third_items_page.data[0].item.id(), "user-2"); + + Ok(()) +} + #[tokio::test] async fn thread_items_list_returns_unsupported() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -1488,6 +1701,88 @@ async fn read_single_turn_items_view( Ok(data.remove(0)) } +async fn read_turns_page( + mcp: &mut TestAppServer, + thread_id: codex_protocol::ThreadId, + cursor: Option, + limit: Option, + items_view: Option, +) -> Result { + let request_id = mcp + .send_thread_turns_list_request(ThreadTurnsListParams { + thread_id: thread_id.to_string(), + cursor, + limit, + sort_direction: Some(SortDirection::Asc), + items_view, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + to_response(response) +} + +async fn read_items_page( + mcp: &mut TestAppServer, + thread_id: codex_protocol::ThreadId, + cursor: Option, + limit: Option, +) -> Result { + let request_id = mcp + .send_thread_items_list_request(ThreadItemsListParams { + thread_id: thread_id.to_string(), + turn_id: None, + cursor, + limit, + sort_direction: Some(SortDirection::Asc), + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + to_response(response) +} + +fn paginated_turn_started(turn_id: &str) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn_id.to_string(), + trace_id: None, + started_at: Some(10), + model_context_window: None, + collaboration_mode_kind: Default::default(), + })) +} + +fn paginated_turn_completed(turn_id: &str) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn_id.to_string(), + last_agent_message: None, + error: None, + started_at: Some(10), + completed_at: Some(20), + duration_ms: Some(10_000), + time_to_first_token_ms: None, + })) +} + +fn paginated_completed_item( + thread_id: codex_protocol::ThreadId, + turn_id: &str, + item: CoreTurnItem, +) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id, + turn_id: turn_id.to_string(), + item, + completed_at_ms: 1, + })) +} + fn turn_user_texts(turns: &[codex_app_server_protocol::Turn]) -> Vec<&str> { turns .iter() diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 922a09c5ac..4671e2ba74 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -344,6 +344,14 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { list_threads::list_threads(self, params).await }) } + fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreFuture<'_, TurnPage> { + Box::pin(LocalThreadStore::list_turns(self, params)) + } + + fn list_items(&self, params: ListItemsParams) -> ThreadStoreFuture<'_, ItemPage> { + Box::pin(LocalThreadStore::list_items(self, params)) + } + fn search_threads( &self, params: SearchThreadsParams,