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
This commit is contained in:
Owen Lin authored and copyberry committed 2026-07-28 00:14:22 +00:00
1 parent 8c72d61727
commit 8d6a91b1ff
4 files changed
+61 -4

No files matched your search

@@ -95,6 +95,8 @@ pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
pub struct ThreadHistoryItemChange { pub struct ThreadHistoryItemChange {
pub turn_id: String, pub turn_id: String,
pub item: ThreadItem, pub item: ThreadItem,
pub started_at_ms: Option<i64>,
pub completed_at_ms: Option<i64>,
} }
/// Lightweight turn metadata snapshot for projectors that track turn status without /// 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) { fn record_changed_item(&mut self, turn_id: String, item: ThreadItem) {
if let Some(change_set) = self.active_change_set.as_mut() { if let Some(change_set) = self.active_change_set.as_mut() {
change_set change_set.changed_items.push(ThreadHistoryItemChange {
.changed_items turn_id,
.push(ThreadHistoryItemChange { turn_id, item }); 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(), text_elements: Vec::new(),
}], }],
}, },
started_at_ms: None,
completed_at_ms: None,
}], }],
changed_turns: vec![ThreadHistoryTurnChange { changed_turns: vec![ThreadHistoryTurnChange {
turn_id: "rollout-0".into(), turn_id: "rollout-0".into(),
@@ -4494,6 +4502,8 @@ mod tests {
}), }),
results: None, results: None,
}), }),
started_at_ms: None,
completed_at_ms: None,
}], }],
changed_turns: Vec::new(), changed_turns: Vec::new(),
removed_turn_ids: Vec::new(), removed_turn_ids: Vec::new(),
@@ -4525,6 +4535,8 @@ mod tests {
summary: vec!["summary".into()], summary: vec!["summary".into()],
content: vec!["raw content".into()], content: vec!["raw content".into()],
}, },
started_at_ms: None,
completed_at_ms: None,
}], }],
changed_turns: Vec::new(), changed_turns: Vec::new(),
removed_turn_ids: Vec::new(), removed_turn_ids: Vec::new(),
@@ -4631,6 +4643,8 @@ mod tests {
}), }),
results: None, results: None,
}), }),
started_at_ms: None,
completed_at_ms: None,
}], }],
changed_turns: vec![ThreadHistoryTurnChange { changed_turns: vec![ThreadHistoryTurnChange {
turn_id: "rollout-0".into(), turn_id: "rollout-0".into(),
@@ -70,6 +70,8 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
changed_items: vec![ThreadHistoryItemChange { changed_items: vec![ThreadHistoryItemChange {
turn_id: event.turn_id.clone(), turn_id: event.turn_id.clone(),
item: ThreadItem::from(event.item.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() ..Default::default()
}, },
@@ -130,6 +130,8 @@ fn projects_completed_canonical_turn_items() {
vec![ThreadHistoryItemChange { vec![ThreadHistoryItemChange {
turn_id: "turn-1".to_string(), turn_id: "turn-1".to_string(),
item: ThreadItem::from(user_item), item: ThreadItem::from(user_item),
started_at_ms: Some(100),
completed_at_ms: Some(123),
}] }]
); );
assert_eq!( assert_eq!(
@@ -137,10 +139,48 @@ fn projects_completed_canonical_turn_items() {
vec![ThreadHistoryItemChange { vec![ThreadHistoryItemChange {
turn_id: "turn-1".to_string(), turn_id: "turn-1".to_string(),
item: ThreadItem::from(agent_item), 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] #[test]
fn ignores_legacy_abort_without_turn_id_and_context_only_records() { fn ignores_legacy_abort_without_turn_id_and_context_only_records() {
let aborted = project(RolloutItem::EventMsg(EventMsg::TurnAborted( 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, thread_id,
turn_id: turn_id.to_string(), turn_id: turn_id.to_string(),
item, item,
started_at_ms: Some(0), started_at_ms: Some(100),
completed_at_ms: 123, completed_at_ms: 123,
})) }))
} }
@@ -827,6 +827,7 @@ mod tests {
text: "Rollout user message".to_string(), text: "Rollout user message".to_string(),
text_elements: Vec::new(), text_elements: Vec::new(),
}])), }])),
started_at_ms: None,
completed_at_ms: 0, completed_at_ms: 0,
})), })),
) )