From 0fb559f0f6e231a88ac02ea002d3ecd248e2b515 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sat, 18 Jul 2026 23:53:21 +0000 Subject: [PATCH] Support legacy views for paginated thread history (#34085) ## Why Clients that use full-history resume or request full items in turn pages should work consistently across legacy and paginated threads. ## What changed - Materialize complete turns and items from the paginated projection for `thread/resume` and `thread/turns/list` when requested. - Support `initialTurnsPage` during paginated resume, including merging a live active turn while preserving the requested page size and cursors. - Keep metadata-only resume behavior and backwards cursors available through `excludeTurns: true`. ## Testing - Extend app-server coverage for full paginated history, full item views, and initial resume pages on running threads. GitOrigin-RevId: e340f44d631ce135895c95d7e1baa7bfecd86e71 --- codex-rs/app-server/README.md | 4 +- .../request_processors/thread_lifecycle.rs | 70 +++- .../request_processors/thread_processor.rs | 304 ++++++++++++++---- codex-rs/app-server/src/thread_state.rs | 4 + .../app-server/tests/suite/v2/thread_read.rs | 163 ++++++++-- .../tests/suite/v2/thread_resume.rs | 90 +++--- 6 files changed, 479 insertions(+), 156 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 2e9abbbaf4..ec60f46c73 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -322,7 +322,7 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage. -Paginated threads require `excludeTurns: true`. Their response leaves `thread.turns` empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. Full-history resume and `initialTurnsPage` remain unavailable for paginated threads. +Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. @@ -516,7 +516,7 @@ 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. +Paginated threads support the same views. Their `full` view is materialized from the paginated item projection before app-server returns the turn page. ```json { "method": "thread/turns/list", "id": 24, "params": { diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 3af419da1d..f0e743d27c 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -530,7 +530,7 @@ pub(super) async fn handle_pending_thread_resume_request( thread_watch_manager: &ThreadWatchManager, outgoing: &Arc, pending_thread_unloads: &Arc>>, - pending: crate::thread_state::PendingThreadResumeRequest, + mut pending: crate::thread_state::PendingThreadResumeRequest, ) { let active_turn = { let state = thread_state.lock().await; @@ -554,11 +554,18 @@ pub(super) async fn handle_pending_thread_resume_request( let connection_id = request_id.connection_id; let mut thread = pending.thread_summary; if pending.include_turns { - populate_thread_turns_from_history( - &mut thread, - &pending.history_items, - active_turn.as_ref(), - ); + if let Some(turns) = pending.paginated_turns.take() { + thread.turns = turns; + } else { + populate_thread_turns_from_history( + &mut thread, + &pending.history_items, + /*active_turn*/ None, + ); + } + if let Some(active_turn) = active_turn.as_ref() { + merge_turn_history_with_active_turn(&mut thread.turns, active_turn.clone()); + } } let thread_status = thread_watch_manager @@ -567,11 +574,33 @@ pub(super) async fn handle_pending_thread_resume_request( set_thread_status_and_interrupt_stale_turns( &mut thread, - thread_status, + thread_status.clone(), has_live_in_progress_turn, ); let token_usage_thread = pending.include_turns.then(|| thread.clone()); - let mut initial_turns_page = if let Some(params) = pending.initial_turns_page.as_ref() { + let mut initial_turns_page = if let Some(mut page) = pending.paginated_initial_turns_page.take() + { + if let (Some(active_turn), Some(params)) = + (active_turn, pending.initial_turns_page.as_ref()) + { + let sort_direction = params.sort_direction.unwrap_or(SortDirection::Desc); + let active_turn_is_in_page = page.data.iter().any(|turn| turn.id == active_turn.id); + if matches!(sort_direction, SortDirection::Desc) + && !active_turn_is_in_page + && let Some(page_with_active_slot) = + pending.paginated_initial_turns_page_with_active_slot.take() + { + page = page_with_active_slot; + } + merge_active_turn_into_page(&mut page, active_turn, params); + } + super::thread_processor::normalize_thread_turns_status( + &mut page.data, + thread_status, + has_live_in_progress_turn, + ); + Some(page) + } else if let Some(params) = pending.initial_turns_page.as_ref() { match super::thread_processor::build_thread_resume_initial_turns_page( &pending.history_items, thread.status.clone(), @@ -799,6 +828,31 @@ pub(super) fn merge_turn_history_with_active_turn(turns: &mut Vec, active_ turns.push(active_turn); } +fn merge_active_turn_into_page( + page: &mut codex_app_server_protocol::TurnsPage, + mut active_turn: Turn, + params: &codex_app_server_protocol::ThreadResumeInitialTurnsPageParams, +) { + super::thread_processor::apply_thread_turns_items_view( + std::slice::from_mut(&mut active_turn), + params.items_view.unwrap_or(TurnItemsView::Summary), + ); + let sort_direction = params.sort_direction.unwrap_or(SortDirection::Desc); + let page_size = super::thread_processor::thread_turns_page_size(params.limit); + let active_turn_is_in_page = page.data.iter().any(|turn| turn.id == active_turn.id); + page.data.retain(|turn| turn.id != active_turn.id); + match sort_direction { + SortDirection::Asc + if active_turn_is_in_page + || (page.data.len() < page_size && page.next_cursor.is_none()) => + { + page.data.push(active_turn); + } + SortDirection::Asc => {} + SortDirection::Desc => page.data.insert(0, active_turn), + } +} + pub(super) fn set_thread_status_and_interrupt_stale_turns( thread: &mut Thread, loaded_status: ThreadStatus, 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 6da21bd82b..80675f30a0 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2625,22 +2625,17 @@ impl ThreadRequestProcessor { 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 page_size = thread_turns_page_size(limit); let sort_direction = match sort_direction.unwrap_or(SortDirection::Desc) { SortDirection::Asc => StoreSortDirection::Asc, SortDirection::Desc => StoreSortDirection::Desc, }; + // `Full` is only a temporary compatibility path. Keep it out of ThreadStore's API: + // load turn shells here, then hydrate their items below. 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", - )); - } + TurnItemsView::Full => StoredTurnItemsView::NotLoaded, }; let page = self .thread_store @@ -2665,7 +2660,13 @@ impl ThreadRequestProcessor { })?; 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 mut turn = stored_turn_to_api_turn(turn, items_view)?; + if matches!(items_view, TurnItemsView::Full) { + turn.items = self + .paginated_turn_full_items(thread_id, turn.id.as_str()) + .await?; + } + turns.push(turn); } let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok(); let has_live_running_thread = match loaded_thread.as_ref() { @@ -2686,6 +2687,116 @@ impl ThreadRequestProcessor { }) } + // Older clients still request `itemsView: "full"` from turn pages. Keep this + // app-server-only hydration path until those clients use `thread/items/list`. + async fn paginated_turn_full_items( + &self, + thread_id: ThreadId, + turn_id: &str, + ) -> Result, JSONRPCErrorError> { + let mut cursor = None; + let mut items = Vec::new(); + loop { + let page = self + .thread_store + .list_items(StoreListItemsParams { + thread_id, + turn_id: Some(turn_id.to_string()), + include_archived: true, + cursor: cursor.clone(), + page_size: THREAD_ITEMS_MAX_LIMIT, + sort_direction: StoreSortDirection::Asc, + }) + .await + .map_err(paginated_history_list_error)?; + for item in page.items { + items.push(deserialize_stored_thread_item(item)?); + } + let Some(next_cursor) = page.next_cursor else { + return Ok(items); + }; + if cursor.as_ref() == Some(&next_cursor) { + return Err(internal_error(format!( + "failed to load full turn items for {turn_id}: thread store returned a repeated cursor" + ))); + } + cursor = Some(next_cursor); + } + } + + // Older clients omit `excludeTurns` and expect full `thread.turns` on resume. + // Remove this slow path once all clients use paginated resume bootstrap. + async fn paginated_thread_full_turns( + &self, + thread_id: ThreadId, + ) -> Result, JSONRPCErrorError> { + let mut cursor = None; + let mut turns = Vec::new(); + loop { + let page = self + .paginated_thread_turns_list_response( + thread_id, + cursor.clone(), + Some(THREAD_TURNS_MAX_LIMIT as u32), + Some(SortDirection::Asc), + Some(TurnItemsView::Full), + ) + .await?; + turns.extend(page.data); + let Some(next_cursor) = page.next_cursor else { + return Ok(turns); + }; + if cursor.as_ref() == Some(&next_cursor) { + return Err(internal_error(format!( + "failed to load full thread turns for {thread_id}: thread store returned a repeated cursor" + ))); + } + cursor = Some(next_cursor); + } + } + + async fn paginated_resume_initial_turns_page( + &self, + thread_id: ThreadId, + params: &ThreadResumeInitialTurnsPageParams, + ) -> Result { + self.paginated_thread_turns_list_response( + thread_id, + /*cursor*/ None, + params.limit, + params.sort_direction, + params.items_view, + ) + .await + .map(Into::into) + } + + async fn paginated_resume_initial_turns_page_with_active_slot( + &self, + thread_id: ThreadId, + params: &ThreadResumeInitialTurnsPageParams, + ) -> Result { + // A running resume overlays the newest live turn on this durable page. + // Reserve one row so the overlay keeps the requested limit and the + // durable next cursor still starts after the last returned stored turn. + let page_size = thread_turns_page_size(params.limit); + if page_size == 1 { + // ThreadStore does not accept an empty page. Use its head cursor as + // the next cursor so the omitted durable row is returned next. + let mut page = self + .paginated_resume_initial_turns_page(thread_id, params) + .await?; + page.next_cursor = page.backwards_cursor.clone(); + page.data.clear(); + return Ok(page); + } + + let mut params = params.clone(); + params.limit = Some((page_size - 1) as u32); + self.paginated_resume_initial_turns_page(thread_id, ¶ms) + .await + } + pub(super) async fn paginated_resume_backwards_cursors( thread_store: &dyn ThreadStore, thread_id: ThreadId, @@ -2994,13 +3105,9 @@ impl ThreadRequestProcessor { .await .map(|thread_history| (thread_history, None)) } else if let Some(stored_thread) = stored_thread_from_running_probe { - self.load_resume_initial_history_from_stored_thread( - *stored_thread, - exclude_turns, - initial_turns_page.is_some(), - ) - .await - .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))) + self.load_resume_initial_history_from_stored_thread(*stored_thread) + .await + .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))) } else { match self .read_stored_thread_for_resume( @@ -3011,11 +3118,7 @@ impl ThreadRequestProcessor { .await { Ok(stored_thread) => self - .load_resume_initial_history_from_stored_thread( - stored_thread, - exclude_turns, - initial_turns_page.is_some(), - ) + .load_resume_initial_history_from_stored_thread(stored_thread) .await .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))), Err(error) => Err(error), @@ -3028,6 +3131,12 @@ impl ThreadRequestProcessor { return Ok(()); } }; + let paginated_thread_id = resume_source_thread.as_ref().and_then(|thread| { + matches!(thread.history_mode, ThreadHistoryMode::Paginated).then_some(thread.thread_id) + }); + let paginated_resume = paginated_thread_id.is_some(); + let needs_paginated_projection = + paginated_resume && (include_turns || initial_turns_page.is_some()); let history_cwd = thread_history.session_cwd(); let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); @@ -3113,6 +3222,30 @@ impl ThreadRequestProcessor { self.outgoing.send_error(request_id, error).await; return Ok(()); }; + // Paginated JSONL is canonical, but its SQLite projection can lag after a + // previous write failure. Persist after reopening the live writer so legacy + // response hydration reads the latest durable turns and items. + if needs_paginated_projection + && let Err(error) = self + .thread_store + .persist_thread(thread_id) + .await + .map_err(thread_store_resume_read_error) + { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + let materialized_turns = if paginated_resume && include_turns { + match self.paginated_thread_full_turns(thread_id).await { + Ok(turns) => Some(turns), + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + } + } else { + None + }; // Auto-attach a thread listener when resuming a thread. log_listener_attach_result( self.ensure_conversation_listener( @@ -3133,7 +3266,7 @@ impl ThreadRequestProcessor { &response_history, rollout_path.as_path(), resume_source_thread, - include_turns, + include_turns && !paginated_resume, ) .await { @@ -3150,6 +3283,9 @@ impl ThreadRequestProcessor { .await .thread_source .map(Into::into); + if let Some(materialized_turns) = materialized_turns { + thread.turns = materialized_turns; + } self.thread_watch_manager .upsert_thread(thread.clone()) @@ -3189,13 +3325,19 @@ impl ThreadRequestProcessor { ); let token_usage_thread = include_turns.then(|| thread.clone()); let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() { - match build_thread_resume_initial_turns_page( - response_history.get_rollout_items(), - thread.status.clone(), - /*has_live_running_thread*/ false, - /*active_turn*/ None, - params, - ) { + let initial_turns_page_result = if paginated_resume { + self.paginated_resume_initial_turns_page(thread_id, params) + .await + } else { + build_thread_resume_initial_turns_page( + response_history.get_rollout_items(), + thread.status.clone(), + /*has_live_running_thread*/ false, + /*active_turn*/ None, + params, + ) + }; + match initial_turns_page_result { Ok(page) => Some(page), Err(error) => { self.outgoing.send_error(request_id, error).await; @@ -3344,11 +3486,8 @@ impl ThreadRequestProcessor { }; if let Some((existing_thread_id, existing_thread, mut source_thread)) = running_thread { - ensure_paginated_resume_supported( - source_thread.history_mode, - params.exclude_turns, - params.initial_turns_page.is_some(), - )?; + let paginated_resume = + matches!(source_thread.history_mode, ThreadHistoryMode::Paginated); let existing_thread_rollout_path = existing_thread.rollout_path(); let active_path = existing_thread_rollout_path .as_ref() @@ -3409,7 +3548,8 @@ impl ThreadRequestProcessor { let redact_resume_payloads = should_redact_thread_resume_payloads(app_server_client_name.as_deref()); let include_turns = !params.exclude_turns; - let needs_history = include_turns || params.initial_turns_page.is_some(); + let needs_history = + !paginated_resume && (include_turns || params.initial_turns_page.is_some()); if needs_history { let source_thread_id = source_thread.thread_id.to_string(); let source_rollout_path = source_thread.rollout_path.clone(); @@ -3479,9 +3619,53 @@ impl ThreadRequestProcessor { .thread_goal_processor .pending_resume_goal_state(existing_thread.as_ref()) .await; - let resume_cursor_store = - matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) - .then(|| Arc::clone(&self.thread_store)); + if paginated_resume && (include_turns || params.initial_turns_page.is_some()) { + // Paginated JSONL is canonical, but its SQLite projection can lag after a + // previous write failure. Persist before legacy response hydration reads the + // latest durable turns and items. + self.thread_store + .persist_thread(existing_thread_id) + .await + .map_err(thread_store_resume_read_error)?; + } + let paginated_turns = if paginated_resume && include_turns { + Some(self.paginated_thread_full_turns(existing_thread_id).await?) + } else { + None + }; + let paginated_initial_turns_page = if paginated_resume { + match params.initial_turns_page.as_ref() { + Some(params) => Some( + self.paginated_resume_initial_turns_page(existing_thread_id, params) + .await?, + ), + None => None, + } + } else { + None + }; + let paginated_initial_turns_page_with_active_slot = if paginated_resume { + match params.initial_turns_page.as_ref() { + Some(params) + if matches!( + params.sort_direction.unwrap_or(SortDirection::Desc), + SortDirection::Desc + ) => + { + Some( + self.paginated_resume_initial_turns_page_with_active_slot( + existing_thread_id, + params, + ) + .await?, + ) + } + Some(_) | None => None, + } + } else { + None + }; + let resume_cursor_store = paginated_resume.then(|| Arc::clone(&self.thread_store)); let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse( Box::new(crate::thread_state::PendingThreadResumeRequest { @@ -3494,6 +3678,9 @@ impl ThreadRequestProcessor { thread_goal_state_db, include_turns, initial_turns_page: params.initial_turns_page.clone(), + paginated_turns, + paginated_initial_turns_page, + paginated_initial_turns_page_with_active_slot, resume_cursor_store, redact_resume_payloads, }), @@ -3528,14 +3715,7 @@ impl ThreadRequestProcessor { async fn load_resume_initial_history_from_stored_thread( &self, stored_thread: StoredThread, - exclude_turns: bool, - has_initial_turns_page: bool, ) -> Result<(InitialHistory, StoredThread), JSONRPCErrorError> { - ensure_paginated_resume_supported( - stored_thread.history_mode, - exclude_turns, - has_initial_turns_page, - )?; if matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) { let model_context = self .thread_store @@ -4295,6 +4475,13 @@ const THREAD_ITEMS_MAX_LIMIT: usize = 100; const THREAD_SEARCH_OCCURRENCES_DEFAULT_LIMIT: usize = 50; const THREAD_SEARCH_OCCURRENCES_MAX_LIMIT: usize = 250; +pub(super) fn thread_turns_page_size(limit: Option) -> usize { + limit + .map(|value| value as usize) + .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) + .clamp(1, THREAD_TURNS_MAX_LIMIT) +} + fn thread_backwards_cursor_for_sort_key( thread: &StoredThread, sort_key: StoreThreadSortKey, @@ -4472,7 +4659,7 @@ pub(super) fn build_thread_resume_initial_turns_page( .map(Into::into) } -fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { +pub(super) fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { for turn in turns { match items_view { TurnItemsView::NotLoaded => { @@ -4528,7 +4715,7 @@ fn reconstruct_thread_turns_for_turns_list( turns } -fn normalize_thread_turns_status( +pub(super) fn normalize_thread_turns_status( turns: &mut [Turn], loaded_status: ThreadStatus, has_live_in_progress_turn: bool, @@ -4616,27 +4803,6 @@ fn stored_turn_to_api_turn( }) } -fn ensure_paginated_resume_supported( - history_mode: ThreadHistoryMode, - exclude_turns: bool, - has_initial_turns_page: bool, -) -> Result<(), JSONRPCErrorError> { - if !matches!(history_mode, ThreadHistoryMode::Paginated) { - return Ok(()); - } - if !exclude_turns { - return Err(invalid_request( - "paginated threads do not support full-history thread/resume; pass excludeTurns=true", - )); - } - if has_initial_turns_page { - return Err(invalid_request( - "paginated threads do not support initialTurnsPage; use turnsBackwardsCursor and itemsBackwardsCursor", - )); - } - Ok(()) -} - 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/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 0e634e3f7f..3843172100 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -40,6 +40,10 @@ pub(crate) struct PendingThreadResumeRequest { pub(crate) include_turns: bool, pub(crate) initial_turns_page: Option, + pub(crate) paginated_turns: Option>, + pub(crate) paginated_initial_turns_page: Option, + pub(crate) paginated_initial_turns_page_with_active_slot: + Option, pub(crate) resume_cursor_store: Option>, pub(crate) redact_resume_payloads: bool, } 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 ed23ac88b5..78de837e76 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1551,6 +1551,15 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { content: Vec::new(), }), ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::UserMessage(UserMessageItem { + id: "steer-1".to_string(), + client_id: None, + content: Vec::new(), + }), + ), paginated_completed_item( thread_id, "turn-1", @@ -1586,6 +1595,94 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { .await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let expected_turn_1_full = Turn { + id: "turn-1".to_string(), + items: vec![ + ThreadItem::UserMessage { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }, + ThreadItem::UserMessage { + id: "steer-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::Full, + status: TurnStatus::Completed, + error: None, + started_at: Some(10), + completed_at: Some(20), + duration_ms: Some(10_000), + }; + let expected_turn_2_full = Turn { + id: "turn-2".to_string(), + items: vec![ThreadItem::UserMessage { + id: "user-2".to_string(), + client_id: None, + content: Vec::new(), + }], + items_view: TurnItemsView::Full, + status: TurnStatus::Interrupted, + error: None, + started_at: Some(10), + completed_at: None, + duration_ms: None, + }; + let expected_full_turns = vec![expected_turn_1_full.clone(), expected_turn_2_full.clone()]; + + let legacy_resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.to_string(), + ..Default::default() + }) + .await?; + let legacy_resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(legacy_resume_id)), + ) + .await??; + let ThreadResumeResponse { + thread: legacy_thread, + .. + } = to_response::(legacy_resume_resp)?; + assert_eq!(legacy_thread.turns, expected_full_turns); + + let initial_page_resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.to_string(), + exclude_turns: true, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: Some(1), + sort_direction: Some(SortDirection::Desc), + items_view: Some(TurnItemsView::Full), + }), + ..Default::default() + }) + .await?; + let initial_page_resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(initial_page_resume_id)), + ) + .await??; + let ThreadResumeResponse { + thread: initial_page_thread, + initial_turns_page, + .. + } = to_response::(initial_page_resume_resp)?; + assert!(initial_page_thread.turns.is_empty()); + assert_eq!( + initial_turns_page.expect("initial turns page").data, + vec![expected_turn_2_full] + ); + let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: thread_id.to_string(), @@ -1663,7 +1760,7 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { data.into_iter() .map(|entry| entry.item.id().to_string()) .collect::>(), - vec!["user-2", "agent-1", "user-1"] + vec!["user-2", "agent-1", "steer-1"] ); let ThreadItemsListResponse { data, .. } = read_items_page( @@ -1679,7 +1776,7 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { data.into_iter() .map(|entry| entry.item.id().to_string()) .collect::>(), - vec!["agent-1", "user-1"] + vec!["agent-1", "steer-1"] ); let first_page = read_turns_page( @@ -1740,25 +1837,16 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { }] ); - 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)), + let full_page = read_turns_page( + &mut mcp, + thread_id, + /*cursor*/ None, + Some(1), + SortDirection::Asc, + Some(TurnItemsView::Full), ) - .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" - ); + .await?; + assert_eq!(full_page.data, vec![expected_turn_1_full]); let first_items_page = read_items_page( &mut mcp, @@ -1783,19 +1871,44 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { .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"); + assert_eq!(second_items_page.data[0].item.id(), "steer-1"); let third_items_page = read_items_page( &mut mcp, thread_id, /*turn_id*/ None, Some(second_items_page.next_cursor.expect("next item cursor")), - Some(1), + Some(2), SortDirection::Asc, ) .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"); + assert_eq!(third_items_page.data.len(), 2); + assert_eq!(third_items_page.data[0].turn_id, "turn-1"); + assert_eq!(third_items_page.data[0].item.id(), "agent-1"); + assert_eq!(third_items_page.data[1].turn_id, "turn-2"); + assert_eq!(third_items_page.data[1].item.id(), "user-2"); + + let turn_start_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread_id.to_string(), + client_user_message_id: None, + input: vec![UserInput::Text { + text: "continue after legacy resume".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_start_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let _: TurnStartResponse = to_response::(turn_start_response)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index f12fd89e50..8b906c9721 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -32,6 +32,7 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::SessionSource; +use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadGoalClearResponse; use codex_app_server_protocol::ThreadGoalSetResponse; use codex_app_server_protocol::ThreadGoalStatus; @@ -127,7 +128,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; #[tokio::test] -async fn thread_resume_paginated_uses_model_context_without_history() -> Result<()> { +async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -147,48 +148,6 @@ async fn thread_resume_paginated_uses_model_context_without_history() -> Result< .await?; timeout(DEFAULT_READ_TIMEOUT, primary.initialize()).await??; - let full_resume_id = primary - .send_thread_resume_request(ThreadResumeParams { - thread_id: conversation_id.clone(), - ..Default::default() - }) - .await?; - let full_resume_err: JSONRPCError = timeout( - DEFAULT_READ_TIMEOUT, - primary.read_stream_until_error_message(RequestId::Integer(full_resume_id)), - ) - .await??; - assert_eq!(full_resume_err.error.code, -32600); - assert_eq!( - full_resume_err.error.message, - "paginated threads do not support full-history thread/resume; pass excludeTurns=true" - ); - - let initial_page_resume_id = primary - .send_thread_resume_request(ThreadResumeParams { - thread_id: conversation_id.clone(), - exclude_turns: true, - initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { - limit: None, - sort_direction: None, - items_view: None, - }), - ..Default::default() - }) - .await?; - let initial_page_err: JSONRPCError = timeout( - DEFAULT_READ_TIMEOUT, - primary.read_stream_until_error_message(RequestId::Integer(initial_page_resume_id)), - ) - .await??; - assert_eq!(initial_page_err.error.code, -32600); - assert_eq!( - initial_page_err.error.message, - "paginated threads do not support initialTurnsPage; use turnsBackwardsCursor and itemsBackwardsCursor" - ); - - // LocalThreadStore rejects paginated full-history reads, so this cold resume only succeeds - // when app-server asks for the bounded latest model context. let resume_id = primary .send_thread_resume_request(ThreadResumeParams { thread_id: conversation_id.clone(), @@ -207,7 +166,6 @@ async fn thread_resume_paginated_uses_model_context_without_history() -> Result< assert_eq!(resumed.id, conversation_id); assert_eq!(resumed.history_mode, ThreadHistoryMode::Paginated); assert!(resumed.turns.is_empty()); - Ok(()) } @@ -3325,7 +3283,7 @@ async fn thread_resume_rejects_mismatched_path_for_running_thread_id() -> Result } #[tokio::test] -async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> Result<()> { +async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> Result<()> { let server = responses::start_mock_server().await; let first_response = responses::sse_response(responses::sse(vec![ responses::ev_response_created("resp-1"), @@ -3337,7 +3295,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R responses::ev_assistant_message("msg-2", "Done"), responses::ev_completed("resp-2"), ])) - .set_delay(std::time::Duration::from_millis(500)); + .set_delay(std::time::Duration::from_secs(2)); let _response_mock = responses::mount_response_sequence(&server, vec![first_response, second_response]).await; let codex_home = TempDir::new()?; @@ -3352,6 +3310,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R let start_id = primary .send_thread_start_request_with_auto_env(ThreadStartParams { model: Some("gpt-5.4".to_string()), + history_mode: Some(ThreadHistoryMode::Paginated), ..Default::default() }) .await?; @@ -3373,11 +3332,12 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R ..Default::default() }) .await?; - timeout( + let seed_turn_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_response_message(RequestId::Integer(seed_turn_id)), ) .await??; + let TurnStartResponse { turn: seed_turn } = to_response::(seed_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_notification_message("turn/completed"), @@ -3415,9 +3375,9 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R model: Some("not-the-running-model".to_string()), cwd: Some("/tmp".to_string()), initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { - limit: None, - sort_direction: None, - items_view: None, + limit: Some(1), + sort_direction: Some(SortDirection::Desc), + items_view: Some(TurnItemsView::NotLoaded), }), ..Default::default() }) @@ -3435,15 +3395,41 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R } = to_response::(resume_resp)?; assert_eq!(model, "gpt-5.4"); let initial_turns_page = initial_turns_page.expect("resume should include initial turns page"); + assert_eq!(initial_turns_page.data.len(), 1); let resumed_running_turn = initial_turns_page .data .first() .expect("resume page should include the running turn"); assert_eq!(resumed_running_turn.id, running_turn.id); - assert_eq!(resumed_running_turn.items_view, TurnItemsView::Summary); + assert_eq!(resumed_running_turn.items_view, TurnItemsView::NotLoaded); + assert!(resumed_running_turn.items.is_empty()); assert_eq!(resumed_running_turn.status, TurnStatus::InProgress); assert!(initial_turns_page.backwards_cursor.is_some()); - assert_eq!(initial_turns_page.next_cursor, None); + assert!(initial_turns_page.next_cursor.is_some()); + + let asc_resume_id = primary + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + exclude_turns: true, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: Some(1), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::NotLoaded), + }), + ..Default::default() + }) + .await?; + let asc_resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + primary.read_stream_until_response_message(RequestId::Integer(asc_resume_id)), + ) + .await??; + let ThreadResumeResponse { + initial_turns_page, .. + } = to_response::(asc_resume_resp)?; + let initial_turns_page = initial_turns_page.expect("resume should include initial turns page"); + assert_eq!(initial_turns_page.data.len(), 1); + assert_eq!(initial_turns_page.data[0].id, seed_turn.id); // The running-thread resume response is queued onto the thread listener task. // If the in-flight turn completes before that queued command runs, the response // can legitimately observe the thread as idle.