mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Support paginated thread history in app-server list APIs (#33152)
## Why Clients could discover threads with paginated history, but could not page their turns through `thread/turns/list`. ## What changed - Route `thread/turns/list` for paginated threads through the thread store, preserving cursors, sort direction, turn status, timing, errors, and the requested `summary` or `notLoaded` item view. - Reject the `full` turn item view for paginated threads and direct clients to `thread/items/list` for complete item pagination. - Return each item page entry with both its containing `turnId` and `item`, so clients can regroup thread-wide pages into turns. - Expose local thread-store turn and item pagination through the `ThreadStore` implementation. ## Testing Added app-server coverage for paging projected turns and items, item views, cursors, turn metadata, and rejection of the unsupported `full` view. GitOrigin-RevId: 9c4a127a9b4fa91313f0e87494c62fe2e3a06f19
This commit is contained in:
+16
@@ -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": [
|
||||
{
|
||||
|
||||
+16
@@ -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": [
|
||||
{
|
||||
|
||||
@@ -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, };
|
||||
@@ -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";
|
||||
|
||||
@@ -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",
|
||||
})
|
||||
|
||||
@@ -1353,11 +1353,20 @@ pub struct ThreadItemsListParams {
|
||||
pub sort_direction: Option<SortDirection>,
|
||||
}
|
||||
|
||||
#[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<ThreadItem>,
|
||||
pub data: Vec<ThreadItemEntry>,
|
||||
/// 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<String>,
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String>,
|
||||
limit: Option<u32>,
|
||||
sort_direction: Option<SortDirection>,
|
||||
items_view: Option<TurnItemsView>,
|
||||
) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> {
|
||||
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::<ThreadItem>(&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::<Result<Vec<_>, _>>()?;
|
||||
|
||||
@@ -4162,6 +4261,49 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError {
|
||||
}
|
||||
}
|
||||
|
||||
fn deserialize_stored_thread_item(
|
||||
item: codex_thread_store::StoredThreadItem,
|
||||
) -> Result<ThreadItem, JSONRPCErrorError> {
|
||||
serde_json::from_slice::<ThreadItem>(&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<Turn, JSONRPCErrorError> {
|
||||
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::<Result<Vec<_>, _>>()?;
|
||||
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"))
|
||||
}
|
||||
|
||||
@@ -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::<ThreadTurnsListResponse>(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<String>,
|
||||
limit: Option<u32>,
|
||||
items_view: Option<TurnItemsView>,
|
||||
) -> Result<ThreadTurnsListResponse> {
|
||||
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<String>,
|
||||
limit: Option<u32>,
|
||||
) -> Result<ThreadItemsListResponse> {
|
||||
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()
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user