From 8d6a91b1ff926bb6006aff94ef4f9e628863cc09 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 28 Jul 2026 00:08:38 +0000 Subject: [PATCH] Preserve item timestamps in thread history projections (#35689) ## What changed - Add optional start and completion timestamps to `ThreadHistoryItemChange`. - Populate them from canonical `ItemCompleted` records during stateless thread history projection. - Treat a zero completion timestamp as absent, and leave timestamps unset for legacy builder events that do not provide them. ## Testing - Cover projected item timestamps, including missing start times and zero completion times. GitOrigin-RevId: 2187530f489ad3d86b6ab65faaaab1eab5357f1b --- .../src/protocol/thread_history.rs | 20 +++++++-- .../src/protocol/thread_history_projection.rs | 2 + .../thread_history_projection_tests.rs | 42 ++++++++++++++++++- .../thread-store/src/local/read_thread.rs | 1 + 4 files changed, 61 insertions(+), 4 deletions(-) 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 717425b949..b50d376c39 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -95,6 +95,8 @@ pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { pub struct ThreadHistoryItemChange { pub turn_id: String, pub item: ThreadItem, + pub started_at_ms: Option, + pub completed_at_ms: Option, } /// Lightweight turn metadata snapshot for projectors that track turn status without @@ -1443,9 +1445,13 @@ impl ThreadHistoryBuilder { fn record_changed_item(&mut self, turn_id: String, item: ThreadItem) { if let Some(change_set) = self.active_change_set.as_mut() { - change_set - .changed_items - .push(ThreadHistoryItemChange { turn_id, item }); + change_set.changed_items.push(ThreadHistoryItemChange { + turn_id, + item, + // Legacy events used by ThreadHistoryBuilder don't have timestamps + started_at_ms: None, + completed_at_ms: None, + }); } } @@ -4446,6 +4452,8 @@ mod tests { text_elements: Vec::new(), }], }, + started_at_ms: None, + completed_at_ms: None, }], changed_turns: vec![ThreadHistoryTurnChange { turn_id: "rollout-0".into(), @@ -4494,6 +4502,8 @@ mod tests { }), results: None, }), + started_at_ms: None, + completed_at_ms: None, }], changed_turns: Vec::new(), removed_turn_ids: Vec::new(), @@ -4525,6 +4535,8 @@ mod tests { summary: vec!["summary".into()], content: vec!["raw content".into()], }, + started_at_ms: None, + completed_at_ms: None, }], changed_turns: Vec::new(), removed_turn_ids: Vec::new(), @@ -4631,6 +4643,8 @@ mod tests { }), results: None, }), + started_at_ms: None, + completed_at_ms: None, }], changed_turns: vec![ThreadHistoryTurnChange { turn_id: "rollout-0".into(), 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 38de714bce..8b0e30a90d 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 @@ -70,6 +70,8 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet { changed_items: vec![ThreadHistoryItemChange { turn_id: event.turn_id.clone(), item: ThreadItem::from(event.item.clone()), + started_at_ms: event.started_at_ms, + completed_at_ms: (event.completed_at_ms != 0).then_some(event.completed_at_ms), }], ..Default::default() }, 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 2eb7875deb..44ea7a1386 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 @@ -130,6 +130,8 @@ fn projects_completed_canonical_turn_items() { vec![ThreadHistoryItemChange { turn_id: "turn-1".to_string(), item: ThreadItem::from(user_item), + started_at_ms: Some(100), + completed_at_ms: Some(123), }] ); assert_eq!( @@ -137,10 +139,48 @@ fn projects_completed_canonical_turn_items() { vec![ThreadHistoryItemChange { turn_id: "turn-1".to_string(), item: ThreadItem::from(agent_item), + started_at_ms: Some(100), + completed_at_ms: Some(123), }] ); } +#[test] +fn projects_optional_completed_item_lifecycle_timestamps() { + let item = TurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }); + + for (started_at_ms, completed_at_ms, expected_completed_at_ms) in + [(None, 123, Some(123)), (Some(100), 0, None)] + { + let changes = project(RolloutItem::EventMsg(EventMsg::ItemCompleted( + ItemCompletedEvent { + thread_id: ThreadId::default(), + turn_id: "turn-1".to_string(), + item: item.clone(), + started_at_ms, + completed_at_ms, + }, + ))); + + assert_eq!( + changes, + ThreadHistoryChangeSet { + changed_items: vec![ThreadHistoryItemChange { + turn_id: "turn-1".to_string(), + item: ThreadItem::from(item.clone()), + started_at_ms, + completed_at_ms: expected_completed_at_ms, + }], + ..Default::default() + } + ); + } +} + #[test] fn ignores_legacy_abort_without_turn_id_and_context_only_records() { let aborted = project(RolloutItem::EventMsg(EventMsg::TurnAborted( @@ -206,7 +246,7 @@ fn item_completed(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> Rollout thread_id, turn_id: turn_id.to_string(), item, - started_at_ms: Some(0), + started_at_ms: Some(100), completed_at_ms: 123, })) } diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 30aded9ff7..686d4d750d 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -827,6 +827,7 @@ mod tests { text: "Rollout user message".to_string(), text_elements: Vec::new(), }])), + started_at_ms: None, completed_at_ms: 0, })), )