openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-10 21:00:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid cloning active turn items for metadata-only thread resumes (#46510)

Browse Source 
## Why

Resuming a running thread cloned the active turn's items even when the response did not need them.

## What changed

Add a metadata-only turn snapshot and use it for resume status checks and initial pages with `TurnItemsView::NotLoaded`. Only snapshot items when including turns or requesting an initial page with items, preserving item data for `Summary` and `Full` views.

## Testing

Add lifecycle coverage for metadata snapshots and extend running-thread resume tests to check metadata-only responses and populated `Summary` and `Full` pages.

GitOrigin-RevId: 281a2b7e1faf3a3fc07e007a065b16235eda08b6
This commit is contained in:
Charlie Marsh
2026-09-18 23:33:08 +00:00
committed by copyberry
parent 36430b3688
commit dfb265764d
6 changed files with 165 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+90 -19
View File
@@ -101,10 +101,9 @@ pub struct ThreadHistoryItemChange {
pub completed_at_ms: Option<i64>,
}
/// Lightweight turn metadata snapshot for projectors that track turn status without
/// re-reading the full item list.
/// Turn metadata for history projection and snapshots that do not need items.
#[derive(Debug, Clone, PartialEq)]
pub struct ThreadHistoryTurnChange {
pub struct ThreadHistoryTurnMetadata {
pub turn_id: String,
pub root_turn_id: Option<String>,
pub status: TurnStatus,
@@ -118,7 +117,7 @@ pub struct ThreadHistoryTurnChange {
#[derive(Debug, Default, Clone, PartialEq)]
pub struct ThreadHistoryChangeSet {
pub changed_items: Vec<ThreadHistoryItemChange>,
pub changed_turns: Vec<ThreadHistoryTurnChange>,
pub changed_turns: Vec<ThreadHistoryTurnMetadata>,
pub removed_turn_ids: Vec<String>,
}
@@ -130,7 +129,23 @@ impl ThreadHistoryChangeSet {
}
}
impl ThreadHistoryTurnChange {
impl From<ThreadHistoryTurnMetadata> for Turn {
/// Builds a wire turn whose items have not been loaded.
fn from(value: ThreadHistoryTurnMetadata) -> Self {
Self {
id: value.turn_id,
items: Vec::new(),
items_view: TurnItemsView::NotLoaded,
error: value.error,
status: value.status,
started_at: value.started_at,
completed_at: value.completed_at,
duration_ms: value.duration_ms,
}
}
}
impl ThreadHistoryTurnMetadata {
fn from_pending_turn(turn: &PendingTurn) -> Self {
Self {
turn_id: turn.id.clone(),
@@ -151,7 +166,7 @@ impl ThreadHistoryTurnChange {
struct ThreadHistoryChangeAccumulator {
changed_items: Vec<Option<ThreadHistoryItemChange>>,
changed_item_indexes: HashMap<(String, String), usize>,
changed_turns: Vec<Option<ThreadHistoryTurnChange>>,
changed_turns: Vec<Option<ThreadHistoryTurnMetadata>>,
changed_turn_indexes: HashMap<String, usize>,
removed_turn_ids: Vec<String>,
removed_turn_indexes: HashMap<String, usize>,
@@ -190,7 +205,7 @@ impl ThreadHistoryChangeAccumulator {
self.changed_items.push(Some(change));
}
fn push_turn_change(&mut self, change: ThreadHistoryTurnChange) {
fn push_turn_change(&mut self, change: ThreadHistoryTurnMetadata) {
if let Some(index) = self.changed_turn_indexes.get(&change.turn_id).copied() {
self.changed_turns[index] = Some(change);
return;
@@ -271,6 +286,15 @@ impl ThreadHistoryBuilder {
.or_else(|| self.turns.last().map(Turn::from))
}
/// Snapshots active turn metadata without cloning its items.
/// Like the full snapshot, falls back to the last finished turn.
pub fn active_turn_metadata_snapshot(&self) -> Option<ThreadHistoryTurnMetadata> {
self.current_turn
.as_ref()
.or_else(|| self.turns.last())
.map(ThreadHistoryTurnMetadata::from_pending_turn)
}
/// Returns the id of the active turn without materializing its items.
pub fn active_turn_id(&self) -> Option<&str> {
self.current_turn
@@ -1213,7 +1237,7 @@ impl ThreadHistoryBuilder {
codex_error_info: payload.codex_error_info.clone().map(Into::into),
additional_details: None,
});
tracking_changes.then(|| ThreadHistoryTurnChange::from_pending_turn(turn))
tracking_changes.then(|| ThreadHistoryTurnMetadata::from_pending_turn(turn))
} else {
None
};
@@ -1227,7 +1251,7 @@ impl ThreadHistoryBuilder {
turn.status = TurnStatus::Interrupted;
turn.completed_at = payload.completed_at;
turn.duration_ms = payload.duration_ms;
ThreadHistoryTurnChange::from_pending_turn(turn)
ThreadHistoryTurnMetadata::from_pending_turn(turn)
};
if let Some(turn_id) = payload.turn_id.as_deref() {
// Prefer an exact ID match so we interrupt the turn explicitly targeted by the event.
@@ -1241,7 +1265,7 @@ impl ThreadHistoryBuilder {
turn.status = TurnStatus::Interrupted;
turn.completed_at = payload.completed_at;
turn.duration_ms = payload.duration_ms;
let changed_turn = ThreadHistoryTurnChange::from_pending_turn(turn);
let changed_turn = ThreadHistoryTurnMetadata::from_pending_turn(turn);
self.record_changed_turn(changed_turn);
return;
}
@@ -1282,7 +1306,7 @@ impl ThreadHistoryBuilder {
}
turn.completed_at = payload.completed_at;
turn.duration_ms = payload.duration_ms;
ThreadHistoryTurnChange::from_pending_turn(turn)
ThreadHistoryTurnMetadata::from_pending_turn(turn)
};
// Prefer an exact ID match from the active turn and then close it.
@@ -1310,7 +1334,7 @@ impl ThreadHistoryBuilder {
}
turn.completed_at = payload.completed_at;
turn.duration_ms = payload.duration_ms;
let changed_turn = ThreadHistoryTurnChange::from_pending_turn(turn);
let changed_turn = ThreadHistoryTurnMetadata::from_pending_turn(turn);
self.record_changed_turn(changed_turn);
return;
}
@@ -1480,11 +1504,11 @@ impl ThreadHistoryBuilder {
fn record_changed_pending_turn(&mut self, turn: &PendingTurn) {
if self.is_tracking_changes() {
self.record_changed_turn(ThreadHistoryTurnChange::from_pending_turn(turn));
self.record_changed_turn(ThreadHistoryTurnMetadata::from_pending_turn(turn));
}
}
fn record_changed_turn(&mut self, turn: ThreadHistoryTurnChange) {
fn record_changed_turn(&mut self, turn: ThreadHistoryTurnMetadata) {
if let Some(change_set) = self.active_change_set.as_mut() {
change_set.changed_turns.push(turn);
}
@@ -3937,6 +3961,53 @@ mod tests {
);
}
#[test]
fn active_turn_metadata_snapshot_tracks_turn_lifecycle() {
let mut builder = ThreadHistoryBuilder::new();
assert_eq!(builder.active_turn_metadata_snapshot(), None);
builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent {
turn_id: "turn-1".into(),
root_turn_id: Some("root-turn".into()),
trace_id: None,
started_at: Some(100),
model_context_window: None,
collaboration_mode_kind: Default::default(),
}));
builder.handle_event(&EventMsg::UserMessage(UserMessageEvent {
message: "hello".into(),
..Default::default()
}));
let mut expected = ThreadHistoryTurnMetadata {
turn_id: "turn-1".into(),
root_turn_id: Some("root-turn".into()),
error: None,
status: TurnStatus::InProgress,
started_at: Some(100),
completed_at: None,
duration_ms: None,
};
assert_eq!(
builder.active_turn_metadata_snapshot(),
Some(expected.clone())
);
assert_eq!(builder.active_turn_snapshot().unwrap().items.len(), 1);
builder.handle_event(&EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: "turn-1".into(),
started_at: Some(100),
last_agent_message: None,
error: None,
completed_at: Some(102),
duration_ms: Some(2_000),
time_to_first_token_ms: None,
}));
expected.status = TurnStatus::Completed;
expected.completed_at = Some(102);
expected.duration_ms = Some(2_000);
assert_eq!(builder.active_turn_metadata_snapshot(), Some(expected));
assert_eq!(builder.active_turn_snapshot().unwrap().items.len(), 1);
}
#[test]
fn patch_apply_begin_updates_active_turn_snapshot_with_file_change() {
let turn_id = "turn-1";
@@ -5009,7 +5080,7 @@ mod tests {
started_at_ms: None,
completed_at_ms: None,
}],
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "rollout-0".into(),
root_turn_id: None,
status: TurnStatus::Completed,
@@ -5117,7 +5188,7 @@ mod tests {
start_changes,
ThreadHistoryChangeSet {
changed_items: Vec::new(),
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-a".into(),
root_turn_id: Some("root-turn".into()),
status: TurnStatus::InProgress,
@@ -5156,7 +5227,7 @@ mod tests {
complete_changes,
ThreadHistoryChangeSet {
changed_items: Vec::new(),
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-a".into(),
root_turn_id: Some("root-turn".into()),
status: TurnStatus::Completed,
@@ -5204,7 +5275,7 @@ mod tests {
started_at_ms: None,
completed_at_ms: None,
}],
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "rollout-0".into(),
root_turn_id: None,
status: TurnStatus::Completed,
@@ -5245,7 +5316,7 @@ mod tests {
changes,
ThreadHistoryChangeSet {
changed_items: Vec::new(),
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-a".into(),
root_turn_id: Some("root-turn".into()),
status: TurnStatus::Completed,
codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs
+4 -4
View File
@@ -9,7 +9,7 @@ use codex_rollout::RolloutLine;
use crate::protocol::thread_history::ThreadHistoryChangeSet;
use crate::protocol::thread_history::ThreadHistoryItemChange;
use crate::protocol::thread_history::ThreadHistoryTurnChange;
use crate::protocol::thread_history::ThreadHistoryTurnMetadata;
use crate::protocol::v2::ThreadItem;
use crate::protocol::v2::TurnError;
use crate::protocol::v2::TurnStatus;
@@ -21,7 +21,7 @@ use crate::protocol::v2::TurnStatus;
pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
match &line.item {
RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: event.turn_id.clone(),
root_turn_id: event.root_turn_id.clone(),
status: TurnStatus::InProgress,
@@ -33,7 +33,7 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
..Default::default()
},
RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: event.turn_id.clone(),
root_turn_id: None,
status: if event.error.is_some() {
@@ -58,7 +58,7 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
return ThreadHistoryChangeSet::default();
};
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: turn_id.clone(),
root_turn_id: None,
status: TurnStatus::Interrupted,
codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs
+3 -3
View File
@@ -57,7 +57,7 @@ fn projects_turn_lifecycle_without_prior_builder_state() {
assert_eq!(
completed,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-1".to_string(),
root_turn_id: None,
status: TurnStatus::Completed,
@@ -94,7 +94,7 @@ fn projects_failed_turn_completion_as_snapshot() {
assert_eq!(
changes,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-1".to_string(),
root_turn_id: None,
status: TurnStatus::Failed,
@@ -245,7 +245,7 @@ fn projects_identified_turn_aborts() {
assert_eq!(
changes,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
changed_turns: vec![ThreadHistoryTurnMetadata {
turn_id: "turn-1".to_string(),
root_turn_id: None,
status: TurnStatus::Interrupted,
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+21 -6
View File
@@ -574,21 +574,31 @@ pub(super) async fn handle_pending_thread_resume_request(
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
mut pending: crate::thread_state::PendingThreadResumeRequest,
) {
let active_turn = {
let (active_turn_metadata, active_turn) = {
let state = thread_state.lock().await;
state.active_turn_snapshot()
let needs_items = pending.include_turns
|| pending
.initial_turns_page
.as_ref()
.is_some_and(|page| !matches!(page.items_view, Some(TurnItemsView::NotLoaded)));
let active_turn = if needs_items {
state.active_turn_snapshot()
} else {
None
};
(state.active_turn_metadata_snapshot(), active_turn)
};
tracing::debug!(
thread_id = %conversation_id,
request_id = ?pending.request_id,
active_turn_present = active_turn.is_some(),
active_turn_id = ?active_turn.as_ref().map(|turn| turn.id.as_str()),
active_turn_status = ?active_turn.as_ref().map(|turn| &turn.status),
active_turn_present = active_turn_metadata.is_some(),
active_turn_id = ?active_turn_metadata.as_ref().map(|turn| turn.turn_id.as_str()),
active_turn_status = ?active_turn_metadata.as_ref().map(|turn| &turn.status),
"composing running thread resume response"
);
let has_live_in_progress_turn =
matches!(conversation.agent_status().await, AgentStatus::Running)
|| active_turn
|| active_turn_metadata
.as_ref()
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
@@ -619,6 +629,11 @@ pub(super) async fn handle_pending_thread_resume_request(
thread_status.clone(),
has_live_in_progress_turn,
);
let active_turn = if pending.initial_turns_page.is_some() {
active_turn.or_else(|| active_turn_metadata.map(Turn::from))
} else {
None
};
let mut initial_turns_page = if let Some(mut page) = pending.paginated_initial_turns_page.take()
{
if let (Some(active_turn), Some(params)) =
codex-rs/app-server/src/thread_state.rs
+5
View File
@@ -3,6 +3,7 @@ use crate::outgoing_message::ConnectionRequestId;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadGoal;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::ThreadHistoryTurnMetadata;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadSettings;
use codex_app_server_protocol::Turn;
@@ -170,6 +171,10 @@ impl ThreadState {
self.current_turn_history.active_turn_id()
}
pub(crate) fn active_turn_metadata_snapshot(&self) -> Option<ThreadHistoryTurnMetadata> {
self.current_turn_history.active_turn_metadata_snapshot()
}
pub(crate) fn register_shutdown_drain_waiter(&mut self) -> oneshot::Receiver<()> {
let (completion_tx, completion_rx) = oneshot::channel();
self.shutdown_drain_waiter = Some(completion_tx);
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+42
View File
@@ -5215,6 +5215,18 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
primary.read_stream_until_notification_message("turn/started"),
)
.await??;
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let started: ItemStartedNotification =
primary.read_notification("item/started").await?;
if started.turn_id == running_turn.id
&& matches!(started.item, ThreadItem::UserMessage { .. })
{
return Ok::<(), anyhow::Error>(());
}
}
})
.await??;
let resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
@@ -5267,6 +5279,13 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
primary.read_response(metadata_resume_id),
)
.await??;
assert_eq!(metadata_resume.thread.id, thread.id);
assert_eq!(
metadata_resume.thread.status,
ThreadStatus::Active {
active_flags: Vec::new()
}
);
assert!(metadata_resume.thread.turns.is_empty());
assert!(metadata_resume.initial_turns_page.is_none());
assert!(
@@ -5279,6 +5298,29 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
"hot paginated resume should wait for a real token usage update"
);
for items_view in [TurnItemsView::Summary, TurnItemsView::Full] {
let 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::Desc),
items_view: Some(items_view),
}),
..Default::default()
})
.await?;
let resumed: ThreadResumeResponse =
timeout(DEFAULT_READ_TIMEOUT, primary.read_response(resume_id)).await??;
let page = resumed.initial_turns_page.expect("initial turns page");
assert_eq!(page.data.len(), 1);
assert_eq!(page.data[0].id, running_turn.id);
assert_eq!(page.data[0].status, TurnStatus::InProgress);
assert_eq!(page.data[0].items_view, items_view);
assert!(!page.data[0].items.is_empty());
}
let asc_resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),