openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-03 19:53:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support legacy views for paginated thread history (#34085)

Browse Source 
## 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
This commit is contained in:
Owen Lin
2026-07-19 00:01:51 +00:00
committed by copyberry
parent 643de86a19
commit 0fb559f0f6
6 changed files with 479 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+62 -8
View File
@@ -530,7 +530,7 @@ pub(super) async fn handle_pending_thread_resume_request(
thread_watch_manager: &ThreadWatchManager,
outgoing: &Arc<OutgoingMessageSender>,
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
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<Turn>, 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,
codex-rs/app-server/src/request_processors/thread_processor.rs
+235 -69
View File
@@ -2625,22 +2625,17 @@ impl ThreadRequestProcessor {
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 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<Vec<ThreadItem>, 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<Vec<Turn>, 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<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> {
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<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> {
// 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, &params)
.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<u32>) -> 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"))
}