diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index df0a5db810..4cc2d56532 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -101,10 +101,9 @@ pub struct ThreadHistoryItemChange { pub completed_at_ms: Option, } -/// 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, pub status: TurnStatus, @@ -118,7 +117,7 @@ pub struct ThreadHistoryTurnChange { #[derive(Debug, Default, Clone, PartialEq)] pub struct ThreadHistoryChangeSet { pub changed_items: Vec, - pub changed_turns: Vec, + pub changed_turns: Vec, pub removed_turn_ids: Vec, } @@ -130,7 +129,23 @@ impl ThreadHistoryChangeSet { } } -impl ThreadHistoryTurnChange { +impl From 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>, changed_item_indexes: HashMap<(String, String), usize>, - changed_turns: Vec>, + changed_turns: Vec>, changed_turn_indexes: HashMap, removed_turn_ids: Vec, removed_turn_indexes: HashMap, @@ -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 { + 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, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs index a0b2a14ac7..5256731ceb 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs @@ -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, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index 4210011287..e976735ea3 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -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, 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 ba11862eff..5e7697df72 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -574,21 +574,31 @@ pub(super) async fn handle_pending_thread_resume_request( pending_thread_unloads: &Arc>>, 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)) = diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index d80ce76982..a02e855f0a 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -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 { + 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); 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 8638286c25..b31c06babe 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -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(),