mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Avoid cloning excluded turn items during thread resume (#46511)
## Why Resuming a running thread with a summary initial page cloned every active-turn item before discarding those outside the summary, including potentially large reasoning payloads. ## What changed - Snapshot the active turn using the requested `TurnItemsView`, copying only the first user message and final agent message for summaries and no items for `NotLoaded`. - Preserve full snapshots when the resume response includes full turns, even if its initial page requests a summary. - Share item selection between snapshots and page responses, and skip selection when a turn already has the requested view. ## Testing Add unit coverage for summary selection, completed-turn metadata, and metadata-only snapshots. Extend running-thread resume coverage for default summaries, explicit full pages, and full turns returned alongside a summary page. GitOrigin-RevId: 6ddcd93c44dbc8083827ac82f442b31730a4edde
This commit is contained in:
@@ -8,5 +8,6 @@ mod mappers;
|
||||
mod serde_helpers;
|
||||
pub mod thread_history;
|
||||
pub mod thread_history_projection;
|
||||
mod turn_items_view;
|
||||
pub mod v1;
|
||||
pub mod v2;
|
||||
|
||||
@@ -280,10 +280,15 @@ impl ThreadHistoryBuilder {
|
||||
}
|
||||
|
||||
pub fn active_turn_snapshot(&self) -> Option<Turn> {
|
||||
self.active_turn_snapshot_with_items_view(TurnItemsView::Full)
|
||||
}
|
||||
|
||||
/// Snapshots the active or last finished turn, copying only the requested items.
|
||||
pub fn active_turn_snapshot_with_items_view(&self, items_view: TurnItemsView) -> Option<Turn> {
|
||||
self.current_turn
|
||||
.as_ref()
|
||||
.map(Turn::from)
|
||||
.or_else(|| self.turns.last().map(Turn::from))
|
||||
.or_else(|| self.turns.last())
|
||||
.map(|turn| turn.snapshot(items_view))
|
||||
}
|
||||
|
||||
/// Snapshots active turn metadata without cloning its items.
|
||||
@@ -1714,6 +1719,19 @@ impl PendingTurn {
|
||||
self.started_at = started_at;
|
||||
self
|
||||
}
|
||||
|
||||
fn snapshot(&self, items_view: TurnItemsView) -> Turn {
|
||||
Turn {
|
||||
id: self.id.clone(),
|
||||
items: items_view.project_items(&self.items),
|
||||
items_view,
|
||||
error: self.error.clone(),
|
||||
status: self.status.clone(),
|
||||
started_at: self.started_at,
|
||||
completed_at: self.completed_at,
|
||||
duration_ms: self.duration_ms,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<PendingTurn> for Turn {
|
||||
@@ -1733,16 +1751,7 @@ impl From<PendingTurn> for Turn {
|
||||
|
||||
impl From<&PendingTurn> for Turn {
|
||||
fn from(value: &PendingTurn) -> Self {
|
||||
Self {
|
||||
id: value.id.clone(),
|
||||
items: value.items.clone(),
|
||||
items_view: TurnItemsView::Full,
|
||||
error: value.error.clone(),
|
||||
status: value.status.clone(),
|
||||
started_at: value.started_at,
|
||||
completed_at: value.completed_at,
|
||||
duration_ms: value.duration_ms,
|
||||
}
|
||||
value.snapshot(TurnItemsView::Full)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3961,6 +3970,78 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn active_turn_summary_preserves_selection_and_completed_metadata() {
|
||||
let mut builder = ThreadHistoryBuilder::new();
|
||||
assert_eq!(
|
||||
builder.active_turn_snapshot_with_items_view(TurnItemsView::Summary),
|
||||
None
|
||||
);
|
||||
builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-1".into(),
|
||||
root_turn_id: None,
|
||||
trace_id: None,
|
||||
started_at: Some(100),
|
||||
model_context_window: None,
|
||||
collaboration_mode_kind: Default::default(),
|
||||
}));
|
||||
builder.handle_event(&EventMsg::UserMessage(UserMessageEvent {
|
||||
message: "first".into(),
|
||||
..Default::default()
|
||||
}));
|
||||
let first_user = builder.active_turn_snapshot().unwrap().items[0].clone();
|
||||
builder.handle_event(&EventMsg::UserMessage(UserMessageEvent {
|
||||
message: "steering".into(),
|
||||
..Default::default()
|
||||
}));
|
||||
for message in ["draft", "answer"] {
|
||||
builder.handle_event(&EventMsg::AgentMessage(AgentMessageEvent {
|
||||
message: message.into(),
|
||||
phase: None,
|
||||
memory_citation: None,
|
||||
delivery: None,
|
||||
questions: None,
|
||||
}));
|
||||
}
|
||||
let final_agent = builder
|
||||
.active_turn_snapshot()
|
||||
.unwrap()
|
||||
.items
|
||||
.last()
|
||||
.unwrap()
|
||||
.clone();
|
||||
builder.handle_event(&EventMsg::AgentReasoning(AgentReasoningEvent {
|
||||
text: "not included in the summary".repeat(1024),
|
||||
}));
|
||||
let mut expected = builder.active_turn_snapshot().unwrap();
|
||||
expected.items = vec![first_user, final_agent];
|
||||
expected.items_view = TurnItemsView::Summary;
|
||||
assert_eq!(
|
||||
builder.active_turn_snapshot_with_items_view(TurnItemsView::Summary),
|
||||
Some(expected.clone())
|
||||
);
|
||||
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_snapshot_with_items_view(TurnItemsView::Summary),
|
||||
Some(expected)
|
||||
);
|
||||
assert_eq!(
|
||||
builder.active_turn_snapshot_with_items_view(TurnItemsView::NotLoaded),
|
||||
builder.active_turn_metadata_snapshot().map(Turn::from)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn active_turn_metadata_snapshot_tracks_turn_lifecycle() {
|
||||
let mut builder = ThreadHistoryBuilder::new();
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
//! Projects a turn's items without copying payloads excluded by the requested view.
|
||||
|
||||
use super::v2::ThreadItem;
|
||||
use super::v2::TurnItemsView;
|
||||
|
||||
impl TurnItemsView {
|
||||
/// Copies only the items included in this view, preserving summary selection order.
|
||||
pub fn project_items(self, items: &[ThreadItem]) -> Vec<ThreadItem> {
|
||||
match self {
|
||||
Self::NotLoaded => Vec::new(),
|
||||
Self::Full => items.to_vec(),
|
||||
Self::Summary => {
|
||||
let first_user_message = items
|
||||
.iter()
|
||||
.find(|item| matches!(item, ThreadItem::UserMessage { .. }));
|
||||
let final_agent_message = items
|
||||
.iter()
|
||||
.rev()
|
||||
.find(|item| matches!(item, ThreadItem::AgentMessage { .. }));
|
||||
match (first_user_message, final_agent_message) {
|
||||
(Some(user_message), Some(agent_message))
|
||||
if user_message.id() != agent_message.id() =>
|
||||
{
|
||||
vec![user_message.clone(), agent_message.clone()]
|
||||
}
|
||||
(Some(user_message), _) => vec![user_message.clone()],
|
||||
(None, Some(agent_message)) => vec![agent_message.clone()],
|
||||
(None, None) => Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -576,16 +576,16 @@ pub(super) async fn handle_pending_thread_resume_request(
|
||||
) {
|
||||
let (active_turn_metadata, active_turn) = {
|
||||
let state = thread_state.lock().await;
|
||||
let needs_items = pending.include_turns
|
||||
|| pending
|
||||
let items_view = if pending.include_turns {
|
||||
Some(TurnItemsView::Full)
|
||||
} else {
|
||||
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
|
||||
.map(|page| page.items_view.unwrap_or(TurnItemsView::Summary))
|
||||
};
|
||||
let active_turn =
|
||||
items_view.and_then(|view| state.active_turn_snapshot_with_items_view(view));
|
||||
(state.active_turn_metadata_snapshot(), active_turn)
|
||||
};
|
||||
tracing::debug!(
|
||||
|
||||
@@ -5731,39 +5731,10 @@ pub(super) fn build_thread_resume_initial_turns_page(
|
||||
|
||||
pub(super) 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;
|
||||
}
|
||||
if !matches!(items_view, TurnItemsView::Full) && turn.items_view != items_view {
|
||||
turn.items = items_view.project_items(&turn.items);
|
||||
}
|
||||
turn.items_view = items_view;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ use codex_app_server_protocol::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadSettings;
|
||||
use codex_app_server_protocol::Turn;
|
||||
use codex_app_server_protocol::TurnError;
|
||||
use codex_app_server_protocol::TurnItemsView;
|
||||
use codex_core::CodexThread;
|
||||
use codex_core::ThreadConfigSnapshot;
|
||||
use codex_file_watcher::WatchRegistration;
|
||||
@@ -171,6 +172,14 @@ impl ThreadState {
|
||||
self.current_turn_history.active_turn_id()
|
||||
}
|
||||
|
||||
pub(crate) fn active_turn_snapshot_with_items_view(
|
||||
&self,
|
||||
items_view: TurnItemsView,
|
||||
) -> Option<Turn> {
|
||||
self.current_turn_history
|
||||
.active_turn_snapshot_with_items_view(items_view)
|
||||
}
|
||||
|
||||
pub(crate) fn active_turn_metadata_snapshot(&self) -> Option<ThreadHistoryTurnMetadata> {
|
||||
self.current_turn_history.active_turn_metadata_snapshot()
|
||||
}
|
||||
|
||||
@@ -5298,15 +5298,19 @@ 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] {
|
||||
for (exclude_turns, items_view) in [
|
||||
(true, None),
|
||||
(true, Some(TurnItemsView::Full)),
|
||||
(false, Some(TurnItemsView::Summary)),
|
||||
] {
|
||||
let resume_id = primary
|
||||
.send_thread_resume_request(ThreadResumeParams {
|
||||
thread_id: thread.id.clone(),
|
||||
exclude_turns: true,
|
||||
exclude_turns,
|
||||
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
|
||||
limit: Some(1),
|
||||
sort_direction: Some(SortDirection::Desc),
|
||||
items_view: Some(items_view),
|
||||
items_view,
|
||||
}),
|
||||
..Default::default()
|
||||
})
|
||||
@@ -5317,8 +5321,17 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
|
||||
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_eq!(
|
||||
page.data[0].items_view,
|
||||
items_view.unwrap_or(TurnItemsView::Summary)
|
||||
);
|
||||
assert!(!page.data[0].items.is_empty());
|
||||
if !exclude_turns {
|
||||
let full_turn = resumed.thread.turns.last().expect("full active turn");
|
||||
assert_eq!(full_turn.id, running_turn.id);
|
||||
assert_eq!(full_turn.items_view, TurnItemsView::Full);
|
||||
assert!(!full_turn.items.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
let asc_resume_id = primary
|
||||
|
||||
Reference in New Issue
Block a user