agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(app-server): include turns page on thread resume (#23534)

Browse Source 
## Summary

The client currently calls `thread/resume` to establish live updates and
immediately follows it with `thread/turns/list` to hydrate recent turns.
This lets `thread/resume` return that page directly, eliminating a round
trip and the ordering/deduplication gap between the two calls.

Experimental clients opt in with `initialTurnsPage: { limit,
sortDirection, itemsView }`. The response returns `initialTurnsPage` as
a `TurnsPage`, including cursors for paging further back in history.
Keeping the controls in a nested opt-in object provides the useful
`thread/turns/list` knobs without spreading page-specific parameters
across `thread/resume`.

## Verification

- `just fmt`
- `just write-app-server-schema --experimental`
- `just write-app-server-schema`
- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-app-server
thread_resume_initial_turns_page_matches_requested_turns_list_page
--tests`
- `cargo test -p codex-app-server
thread_resume_rejoins_running_thread_even_with_override_mismatch
--tests`
- `just fix -p codex-app-server-protocol -p codex-app-server`
This commit is contained in:
Brent Traut
2026-05-28 09:18:13 -07:00
committed by GitHub
parent 2066874415
commit 2a1158b8e2
22 changed files with 730 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors.rs
+1
View File
@@ -213,6 +213,7 @@ use codex_app_server_protocol::ThreadRealtimeStartResponse;
use codex_app_server_protocol::ThreadRealtimeStartTransport;
use codex_app_server_protocol::ThreadRealtimeStopParams;
use codex_app_server_protocol::ThreadRealtimeStopResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+22 -1
View File
@@ -565,8 +565,28 @@ pub(super) async fn handle_pending_thread_resume_request(
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() {
match super::thread_processor::build_thread_resume_initial_turns_page(
&pending.history_items,
thread.status.clone(),
has_live_in_progress_turn,
active_turn,
params,
) {
Ok(page) => Some(page),
Err(error) => {
outgoing.send_error(request_id, error).await;
return;
}
}
} else {
None
};
if pending.redact_resume_payloads {
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
redact_thread_resume_payloads(&mut initial_turns_page.data);
}
}
{
@@ -635,6 +655,7 @@ pub(super) async fn handle_pending_thread_resume_request(
sandbox,
active_permission_profile,
reasoning_effort,
initial_turns_page,
};
outgoing.send_response(request_id, response).await;
// Match cold resume: metadata-only resume should attach the listener without
codex-rs/app-server/src/request_processors/thread_processor.rs
+121 -51
View File
@@ -2241,8 +2241,6 @@ impl ThreadRequestProcessor {
sort_direction,
items_view,
} = params;
let items_view = items_view.unwrap_or(TurnItemsView::Summary);
let thread_uuid = ThreadId::from_string(&thread_id)
.map_err(|err| invalid_request(format!("invalid thread id: {err}")))?;
@@ -2270,60 +2268,20 @@ impl ThreadRequestProcessor {
} else {
None
};
let mut turns = reconstruct_thread_turns_for_turns_list(
build_thread_turns_page_response(
&items,
self.thread_watch_manager
.loaded_status_for_thread(&thread_uuid.to_string())
.await,
has_live_running_thread,
active_turn,
);
for turn in &mut turns {
match items_view {
TurnItemsView::NotLoaded => {
turn.items.clear();
turn.items_view = TurnItemsView::NotLoaded;
}
TurnItemsView::Summary => {
let first_user_message = turn
.items
.iter()
.find(|item| matches!(item, ThreadItem::UserMessage { .. }))
.cloned();
let final_agent_message = turn
.items
.iter()
.rev()
.find(|item| matches!(item, ThreadItem::AgentMessage { .. }))
.cloned();
turn.items = match (first_user_message, final_agent_message) {
(Some(user_message), Some(agent_message))
if user_message.id() != agent_message.id() =>
{
vec![user_message, agent_message]
}
(Some(user_message), _) => vec![user_message],
(None, Some(agent_message)) => vec![agent_message],
(None, None) => Vec::new(),
};
turn.items_view = TurnItemsView::Summary;
}
TurnItemsView::Full => {
turn.items_view = TurnItemsView::Full;
}
}
}
let page = paginate_thread_turns(
turns,
cursor.as_deref(),
limit,
sort_direction.unwrap_or(SortDirection::Desc),
)?;
Ok(ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
})
ThreadTurnsPageOptions {
cursor: cursor.as_deref(),
limit,
sort_direction: sort_direction.unwrap_or(SortDirection::Desc),
items_view: items_view.unwrap_or(TurnItemsView::Summary),
},
)
}
async fn load_thread_turns_list_history(
@@ -2531,6 +2489,7 @@ impl ThreadRequestProcessor {
developer_instructions,
personality,
exclude_turns,
initial_turns_page,
persist_extended_history: _persist_extended_history,
} = params;
let include_turns = !exclude_turns;
@@ -2685,8 +2644,28 @@ impl ThreadRequestProcessor {
config_snapshot.active_permission_profile,
);
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,
) {
Ok(page) => Some(page),
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
}
}
} else {
None
};
if redact_resume_payloads {
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
redact_thread_resume_payloads(&mut initial_turns_page.data);
}
}
let response = ThreadResumeResponse {
@@ -2702,6 +2681,7 @@ impl ThreadRequestProcessor {
sandbox,
active_permission_profile,
reasoning_effort: session_configured.reasoning_effort,
initial_turns_page,
};
let connection_id = request_id.connection_id;
@@ -2926,6 +2906,7 @@ impl ThreadRequestProcessor {
emit_thread_goal_update,
thread_goal_state_db,
include_turns: !params.exclude_turns,
initial_turns_page: params.initial_turns_page.clone(),
redact_resume_payloads,
}),
);
@@ -3706,6 +3687,95 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCE
serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}")))
}
struct ThreadTurnsPageOptions<'a> {
cursor: Option<&'a str>,
limit: Option<u32>,
sort_direction: SortDirection,
items_view: TurnItemsView,
}
fn build_thread_turns_page_response(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_running_thread: bool,
active_turn: Option<Turn>,
options: ThreadTurnsPageOptions<'_>,
) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> {
let mut turns = reconstruct_thread_turns_for_turns_list(
items,
loaded_status,
has_live_running_thread,
active_turn,
);
apply_thread_turns_items_view(&mut turns, options.items_view);
let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?;
Ok(ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
})
}
pub(super) fn build_thread_resume_initial_turns_page(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_running_thread: bool,
active_turn: Option<Turn>,
params: &ThreadResumeInitialTurnsPageParams,
) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> {
build_thread_turns_page_response(
items,
loaded_status,
has_live_running_thread,
active_turn,
ThreadTurnsPageOptions {
cursor: None,
limit: params.limit,
sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc),
items_view: params.items_view.unwrap_or(TurnItemsView::Summary),
},
)
.map(Into::into)
}
fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) {
for turn in turns {
match items_view {
TurnItemsView::NotLoaded => {
turn.items.clear();
turn.items_view = TurnItemsView::NotLoaded;
}
TurnItemsView::Summary => {
let first_user_message = turn
.items
.iter()
.find(|item| matches!(item, ThreadItem::UserMessage { .. }))
.cloned();
let final_agent_message = turn
.items
.iter()
.rev()
.find(|item| matches!(item, ThreadItem::AgentMessage { .. }))
.cloned();
turn.items = match (first_user_message, final_agent_message) {
(Some(user_message), Some(agent_message))
if user_message.id() != agent_message.id() =>
{
vec![user_message, agent_message]
}
(Some(user_message), _) => vec![user_message],
(None, Some(agent_message)) => vec![agent_message],
(None, None) => Vec::new(),
};
turn.items_view = TurnItemsView::Summary;
}
TurnItemsView::Full => {
turn.items_view = TurnItemsView::Full;
}
}
}
}
fn reconstruct_thread_turns_for_turns_list(
items: &[RolloutItem],
loaded_status: ThreadStatus,
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
+1
View File
@@ -652,6 +652,7 @@ mod thread_processor_behavior_tests {
developer_instructions: None,
personality: None,
exclude_turns: false,
initial_turns_page: None,
persist_extended_history: false,
};
let config_snapshot = ThreadConfigSnapshot {
codex-rs/app-server/src/request_processors/thread_resume_redaction.rs
+6 -6
View File
@@ -1,6 +1,6 @@
use codex_app_server_protocol::McpToolCallResult;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::Turn;
use serde_json::Value as JsonValue;
// Temporary bandaid for remote clients: thread/resume can include large MCP and
@@ -14,8 +14,8 @@ pub(super) fn should_redact_thread_resume_payloads(client_name: Option<&str>) ->
client_name.is_some_and(|client_name| CHATGPT_REMOTE_CLIENT_NAMES.contains(&client_name))
}
pub(super) fn redact_thread_resume_payloads(thread: &mut Thread) {
for turn in &mut thread.turns {
pub(super) fn redact_thread_resume_payloads(turns: &mut [Turn]) {
for turn in turns {
turn.items.retain_mut(|item| match item {
ThreadItem::McpToolCall {
arguments,
@@ -55,8 +55,8 @@ mod tests {
use codex_app_server_protocol::McpToolCallError;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStatus;
use codex_utils_absolute_path::test_support::PathBufExt;
@@ -100,7 +100,7 @@ mod tests {
},
]);
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
assert_eq!(thread.turns[0].items.len(), 2);
assert_eq!(
@@ -146,7 +146,7 @@ mod tests {
duration_ms: Some(8),
}]);
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
assert_eq!(
thread.turns[0].items[0],
codex-rs/app-server/src/thread_state.rs
+2
View File
@@ -35,6 +35,8 @@ pub(crate) struct PendingThreadResumeRequest {
pub(crate) emit_thread_goal_update: bool,
pub(crate) thread_goal_state_db: Option<StateDbHandle>,
pub(crate) include_turns: bool,
pub(crate) initial_turns_page:
Option<codex_app_server_protocol::ThreadResumeInitialTurnsPageParams>,
pub(crate) redact_resume_payloads: bool,
}