Add incremental thread history changes

Add ThreadHistoryBuilder APIs for collecting incremental thread item and turn changes while applying rollout items.

Batch handling coalesces repeated changes so callers can get the latest incremental thread item changes for a set of rollout items without rebuilding full history.

codex-rs/app-server-protocol/src/protocol/thread_history.rs

@@ -83,12 +83,154 @@ pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
     builder.finish()
 }
 
+/// A materialized `ThreadItem` snapshot that changed while handling one input.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ThreadHistoryItemChange {
+    pub turn_id: String,
+    pub item: ThreadItem,
+}
+
+/// Lightweight turn metadata snapshot for projectors that track turn status without
+/// re-reading the full item list.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ThreadHistoryTurnChange {
+    pub turn_id: String,
+    pub status: TurnStatus,
+    pub error: Option<TurnError>,
+    pub started_at: Option<i64>,
+    pub completed_at: Option<i64>,
+    pub duration_ms: Option<i64>,
+}
+
+/// Incremental changes produced by opt-in `ThreadHistoryBuilder` handlers.
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct ThreadHistoryChangeSet {
+    pub changed_items: Vec<ThreadHistoryItemChange>,
+    pub changed_turns: Vec<ThreadHistoryTurnChange>,
+    pub removed_turn_ids: Vec<String>,
+}
+
+impl ThreadHistoryChangeSet {
+    pub fn is_empty(&self) -> bool {
+        self.changed_items.is_empty()
+            && self.changed_turns.is_empty()
+            && self.removed_turn_ids.is_empty()
+    }
+}
+
+impl ThreadHistoryTurnChange {
+    fn from_pending_turn(turn: &PendingTurn) -> Self {
+        Self {
+            turn_id: turn.id.clone(),
+            status: turn.status.clone(),
+            error: turn.error.clone(),
+            started_at: turn.started_at,
+            completed_at: turn.completed_at,
+            duration_ms: turn.duration_ms,
+        }
+    }
+
+    fn from_turn(turn: &Turn) -> Self {
+        Self {
+            turn_id: turn.id.clone(),
+            status: turn.status.clone(),
+            error: turn.error.clone(),
+            started_at: turn.started_at,
+            completed_at: turn.completed_at,
+            duration_ms: turn.duration_ms,
+        }
+    }
+}
+
+/// Coalesces per-rollout-item changes into an end-of-batch view. It preserves
+/// first-change order while replacing repeated item/turn snapshots with their
+/// latest value, and drops accumulated changes for turns removed by rollback.
+#[derive(Default)]
+struct ThreadHistoryChangeAccumulator {
+    changed_items: Vec<Option<ThreadHistoryItemChange>>,
+    changed_item_indexes: HashMap<(String, String), usize>,
+    changed_turns: Vec<Option<ThreadHistoryTurnChange>>,
+    changed_turn_indexes: HashMap<String, usize>,
+    removed_turn_ids: Vec<String>,
+    removed_turn_indexes: HashMap<String, usize>,
+}
+
+impl ThreadHistoryChangeAccumulator {
+    fn push(&mut self, changes: ThreadHistoryChangeSet) {
+        for turn_id in changes.removed_turn_ids {
+            self.push_removed_turn_id(turn_id);
+        }
+        for item_change in changes.changed_items {
+            self.push_item_change(item_change);
+        }
+        for turn_change in changes.changed_turns {
+            self.push_turn_change(turn_change);
+        }
+    }
+
+    fn finish(self) -> ThreadHistoryChangeSet {
+        ThreadHistoryChangeSet {
+            changed_items: self.changed_items.into_iter().flatten().collect(),
+            changed_turns: self.changed_turns.into_iter().flatten().collect(),
+            removed_turn_ids: self.removed_turn_ids,
+        }
+    }
+
+    fn push_item_change(&mut self, change: ThreadHistoryItemChange) {
+        let key = (change.turn_id.clone(), change.item.id().to_string());
+        if let Some(index) = self.changed_item_indexes.get(&key).copied() {
+            self.changed_items[index] = Some(change);
+            return;
+        }
+
+        self.changed_item_indexes
+            .insert(key, self.changed_items.len());
+        self.changed_items.push(Some(change));
+    }
+
+    fn push_turn_change(&mut self, change: ThreadHistoryTurnChange) {
+        if let Some(index) = self.changed_turn_indexes.get(&change.turn_id).copied() {
+            self.changed_turns[index] = Some(change);
+            return;
+        }
+
+        self.changed_turn_indexes
+            .insert(change.turn_id.clone(), self.changed_turns.len());
+        self.changed_turns.push(Some(change));
+    }
+
+    fn push_removed_turn_id(&mut self, turn_id: String) {
+        if !self.removed_turn_indexes.contains_key(&turn_id) {
+            self.removed_turn_indexes
+                .insert(turn_id.clone(), self.removed_turn_ids.len());
+            self.removed_turn_ids.push(turn_id.clone());
+        }
+
+        if let Some(index) = self.changed_turn_indexes.remove(&turn_id) {
+            self.changed_turns[index] = None;
+        }
+
+        let removed_item_keys: Vec<(String, String)> = self
+            .changed_item_indexes
+            .keys()
+            .filter(|(item_turn_id, _)| item_turn_id == &turn_id)
+            .cloned()
+            .collect();
+        for key in removed_item_keys {
+            if let Some(index) = self.changed_item_indexes.remove(&key) {
+                self.changed_items[index] = None;
+            }
+        }
+    }
+}
+
 pub struct ThreadHistoryBuilder {
     turns: Vec<Turn>,
     current_turn: Option<PendingTurn>,
     next_item_index: i64,
     current_rollout_index: usize,
     next_rollout_index: usize,
+    active_change_set: Option<ThreadHistoryChangeSet>,
 }
 
 impl Default for ThreadHistoryBuilder {
@@ -105,6 +247,7 @@ impl ThreadHistoryBuilder {
             next_item_index: 1,
             current_rollout_index: 0,
             next_rollout_index: 0,
+            active_change_set: None,
         }
     }
 
@@ -247,6 +390,42 @@ impl ThreadHistoryBuilder {
         }
     }
 
+    /// Handles one event and returns the materialized items or turn metadata
+    /// changed by that event.
+    pub fn handle_event_with_changes(&mut self, event: &EventMsg) -> ThreadHistoryChangeSet {
+        self.collect_changes(|builder| builder.handle_event(event))
+    }
+
+    /// Handles a rollout item and returns the materialized items or turn metadata
+    /// changed by that one append.
+    pub fn handle_rollout_item_with_changes(
+        &mut self,
+        item: &RolloutItem,
+    ) -> ThreadHistoryChangeSet {
+        self.collect_changes(|builder| builder.handle_rollout_item(item))
+    }
+
+    /// Handles rollout items in order and returns a coalesced end-of-batch
+    /// change set. Multiple changes to the same item or turn are deduplicated
+    /// so only the latest snapshot is emitted.
+    pub fn handle_rollout_items_with_changes(
+        &mut self,
+        items: &[RolloutItem],
+    ) -> ThreadHistoryChangeSet {
+        let mut accumulator = ThreadHistoryChangeAccumulator::default();
+        for item in items {
+            accumulator.push(self.handle_rollout_item_with_changes(item));
+        }
+        accumulator.finish()
+    }
+
+    fn collect_changes(&mut self, handle: impl FnOnce(&mut Self)) -> ThreadHistoryChangeSet {
+        debug_assert!(self.active_change_set.is_none());
+        self.active_change_set = Some(ThreadHistoryChangeSet::default());
+        handle(self);
+        self.active_change_set.take().unwrap_or_default()
+    }
+
     fn handle_response_item(&mut self, item: &codex_protocol::models::ResponseItem) {
         let codex_protocol::models::ResponseItem::Message {
             role, content, id, ..
@@ -263,7 +442,7 @@ impl ThreadHistoryBuilder {
             return;
         };
 
-        self.ensure_turn().items.push(ThreadItem::HookPrompt {
+        self.push_item_in_current_turn(ThreadItem::HookPrompt {
             id: hook_prompt.id,
             fragments: hook_prompt
                 .fragments
@@ -283,18 +462,13 @@ impl ThreadHistoryBuilder {
         {
             self.finish_current_turn();
         }
-        let mut turn = self
-            .current_turn
-            .take()
-            .unwrap_or_else(|| self.new_turn(/*id*/ None));
         let id = self.next_item_id();
         let content = self.build_user_inputs(payload);
-        turn.items.push(ThreadItem::UserMessage {
+        self.push_item_in_current_turn(ThreadItem::UserMessage {
             id,
             client_id: payload.client_id.clone(),
             content,
         });
-        self.current_turn = Some(turn);
     }
 
     fn handle_agent_message(
@@ -308,7 +482,7 @@ impl ThreadHistoryBuilder {
         }
 
         let id = self.next_item_id();
-        self.ensure_turn().items.push(ThreadItem::AgentMessage {
+        self.push_item_in_current_turn(ThreadItem::AgentMessage {
             id,
             text,
             phase,
@@ -322,14 +496,34 @@ impl ThreadHistoryBuilder {
         }
 
         // If the last item is a reasoning item, add the new text to the summary.
-        if let Some(ThreadItem::Reasoning { summary, .. }) = self.ensure_turn().items.last_mut() {
-            summary.push(payload.text.clone());
+        let existing_item_change = {
+            let tracking_changes = self.is_tracking_changes();
+            let turn = self.ensure_turn();
+            if let Some(ThreadItem::Reasoning { summary, .. }) = turn.items.last_mut() {
+                summary.push(payload.text.clone());
+                let changed_item = if tracking_changes {
+                    turn.items
+                        .last()
+                        .cloned()
+                        .map(|item| (turn.id.clone(), item))
+                } else {
+                    None
+                };
+                Some(changed_item)
+            } else {
+                None
+            }
+        };
+        if let Some(changed_item) = existing_item_change {
+            if let Some((turn_id, item)) = changed_item {
+                self.record_changed_item(turn_id, item);
+            }
             return;
         }
 
         // Otherwise, create a new reasoning item.
         let id = self.next_item_id();
-        self.ensure_turn().items.push(ThreadItem::Reasoning {
+        self.push_item_in_current_turn(ThreadItem::Reasoning {
             id,
             summary: vec![payload.text.clone()],
             content: Vec::new(),
@@ -342,14 +536,34 @@ impl ThreadHistoryBuilder {
         }
 
         // If the last item is a reasoning item, add the new text to the content.
-        if let Some(ThreadItem::Reasoning { content, .. }) = self.ensure_turn().items.last_mut() {
-            content.push(payload.text.clone());
+        let existing_item_change = {
+            let tracking_changes = self.is_tracking_changes();
+            let turn = self.ensure_turn();
+            if let Some(ThreadItem::Reasoning { content, .. }) = turn.items.last_mut() {
+                content.push(payload.text.clone());
+                let changed_item = if tracking_changes {
+                    turn.items
+                        .last()
+                        .cloned()
+                        .map(|item| (turn.id.clone(), item))
+                } else {
+                    None
+                };
+                Some(changed_item)
+            } else {
+                None
+            }
+        };
+        if let Some(changed_item) = existing_item_change {
+            if let Some((turn_id, item)) = changed_item {
+                self.record_changed_item(turn_id, item);
+            }
             return;
         }
 
         // Otherwise, create a new reasoning item.
         let id = self.next_item_id();
-        self.ensure_turn().items.push(ThreadItem::Reasoning {
+        self.push_item_in_current_turn(ThreadItem::Reasoning {
             id,
             summary: Vec::new(),
             content: vec![payload.text.clone()],
@@ -884,9 +1098,7 @@ impl ThreadHistoryBuilder {
 
     fn handle_context_compacted(&mut self, _payload: &ContextCompactedEvent) {
         let id = self.next_item_id();
-        self.ensure_turn()
-            .items
-            .push(ThreadItem::ContextCompaction { id });
+        self.push_item_in_current_turn(ThreadItem::ContextCompaction { id });
     }
 
     fn handle_entered_review_mode(&mut self, payload: &codex_protocol::protocol::ReviewRequest) {
@@ -895,9 +1107,7 @@ impl ThreadHistoryBuilder {
             .clone()
             .unwrap_or_else(|| "Review requested.".to_string());
         let id = self.next_item_id();
-        self.ensure_turn()
-            .items
-            .push(ThreadItem::EnteredReviewMode { id, review });
+        self.push_item_in_current_turn(ThreadItem::EnteredReviewMode { id, review });
     }
 
     fn handle_exited_review_mode(
@@ -910,24 +1120,28 @@ impl ThreadHistoryBuilder {
             .map(render_review_output_text)
             .unwrap_or_else(|| REVIEW_FALLBACK_MESSAGE.to_string());
         let id = self.next_item_id();
-        self.ensure_turn()
-            .items
-            .push(ThreadItem::ExitedReviewMode { id, review });
+        self.push_item_in_current_turn(ThreadItem::ExitedReviewMode { id, review });
     }
 
     fn handle_error(&mut self, payload: &ErrorEvent) {
         if !payload.affects_turn_status() {
             return;
         }
-        let Some(turn) = self.current_turn.as_mut() else {
-            return;
+        let tracking_changes = self.is_tracking_changes();
+        let changed_turn = if let Some(turn) = self.current_turn.as_mut() {
+            turn.status = TurnStatus::Failed;
+            turn.error = Some(V2TurnError {
+                message: payload.message.clone(),
+                codex_error_info: payload.codex_error_info.clone().map(Into::into),
+                additional_details: None,
+            });
+            tracking_changes.then(|| ThreadHistoryTurnChange::from_pending_turn(turn))
+        } else {
+            None
         };
-        turn.status = TurnStatus::Failed;
-        turn.error = Some(V2TurnError {
-            message: payload.message.clone(),
-            codex_error_info: payload.codex_error_info.clone().map(Into::into),
-            additional_details: None,
-        });
+        if let Some(changed_turn) = changed_turn {
+            self.record_changed_turn(changed_turn);
+        }
     }
 
     fn handle_turn_aborted(&mut self, payload: &TurnAbortedEvent) {
@@ -935,11 +1149,13 @@ impl ThreadHistoryBuilder {
             turn.status = TurnStatus::Interrupted;
             turn.completed_at = payload.completed_at;
             turn.duration_ms = payload.duration_ms;
+            ThreadHistoryTurnChange::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.
             if let Some(turn) = self.current_turn.as_mut().filter(|turn| turn.id == turn_id) {
-                apply_abort(turn);
+                let changed_turn = apply_abort(turn);
+                self.record_changed_turn(changed_turn);
                 return;
             }
 
@@ -947,24 +1163,28 @@ impl ThreadHistoryBuilder {
                 turn.status = TurnStatus::Interrupted;
                 turn.completed_at = payload.completed_at;
                 turn.duration_ms = payload.duration_ms;
+                let changed_turn = ThreadHistoryTurnChange::from_turn(turn);
+                self.record_changed_turn(changed_turn);
                 return;
             }
         }
 
         // If the event has no ID (or refers to an unknown turn), fall back to the active turn.
         if let Some(turn) = self.current_turn.as_mut() {
-            apply_abort(turn);
+            let changed_turn = apply_abort(turn);
+            self.record_changed_turn(changed_turn);
         }
     }
 
     fn handle_turn_started(&mut self, payload: &TurnStartedEvent) {
         self.finish_current_turn();
-        self.current_turn = Some(
-            self.new_turn(Some(payload.turn_id.clone()))
-                .with_status(TurnStatus::InProgress)
-                .with_started_at(payload.started_at)
-                .opened_explicitly(),
-        );
+        let turn = self
+            .new_turn(Some(payload.turn_id.clone()))
+            .with_status(TurnStatus::InProgress)
+            .with_started_at(payload.started_at)
+            .opened_explicitly();
+        self.record_changed_pending_turn(&turn);
+        self.current_turn = Some(turn);
     }
 
     fn handle_turn_complete(&mut self, payload: &TurnCompleteEvent) {
@@ -974,6 +1194,7 @@ impl ThreadHistoryBuilder {
             }
             turn.completed_at = payload.completed_at;
             turn.duration_ms = payload.duration_ms;
+            ThreadHistoryTurnChange::from_pending_turn(turn)
         };
 
         // Prefer an exact ID match from the active turn and then close it.
@@ -982,7 +1203,8 @@ impl ThreadHistoryBuilder {
             .as_mut()
             .filter(|turn| turn.id == payload.turn_id)
         {
-            mark_completed(current_turn);
+            let changed_turn = mark_completed(current_turn);
+            self.record_changed_turn(changed_turn);
             self.finish_current_turn();
             return;
         }
@@ -997,12 +1219,15 @@ impl ThreadHistoryBuilder {
             }
             turn.completed_at = payload.completed_at;
             turn.duration_ms = payload.duration_ms;
+            let changed_turn = ThreadHistoryTurnChange::from_turn(turn);
+            self.record_changed_turn(changed_turn);
             return;
         }
 
         // If the completion event cannot be matched, apply it to the active turn.
         if let Some(current_turn) = self.current_turn.as_mut() {
-            mark_completed(current_turn);
+            let changed_turn = mark_completed(current_turn);
+            self.record_changed_turn(changed_turn);
             self.finish_current_turn();
         }
     }
@@ -1020,6 +1245,18 @@ impl ThreadHistoryBuilder {
         self.finish_current_turn();
 
         let n = usize::try_from(payload.num_turns).unwrap_or(usize::MAX);
+        let removed_turn_ids = if n >= self.turns.len() {
+            self.turns.iter().map(|turn| turn.id.clone()).collect()
+        } else if n == 0 {
+            Vec::new()
+        } else {
+            self.turns[self.turns.len() - n..]
+                .iter()
+                .map(|turn| turn.id.clone())
+                .collect()
+        };
+        self.record_removed_turn_ids(removed_turn_ids);
+
         if n >= self.turns.len() {
             self.turns.clear();
         } else {
@@ -1064,7 +1301,8 @@ impl ThreadHistoryBuilder {
     fn ensure_turn(&mut self) -> &mut PendingTurn {
         if self.current_turn.is_none() {
             let turn = self.new_turn(/*id*/ None);
-            return self.current_turn.insert(turn);
+            self.record_changed_pending_turn(&turn);
+            self.current_turn = Some(turn);
         }
 
         if let Some(turn) = self.current_turn.as_mut() {
@@ -1074,16 +1312,42 @@ impl ThreadHistoryBuilder {
         unreachable!("current turn must exist after initialization");
     }
 
+    fn push_item_in_current_turn(&mut self, item: ThreadItem) {
+        let tracking_changes = self.is_tracking_changes();
+        let changed_item = {
+            let turn = self.ensure_turn();
+            let changed_item = tracking_changes.then(|| (turn.id.clone(), item.clone()));
+            turn.items.push(item);
+            changed_item
+        };
+        if let Some((turn_id, item)) = changed_item {
+            self.record_changed_item(turn_id, item);
+        }
+    }
+
     fn upsert_item_in_turn_id(&mut self, turn_id: &str, item: ThreadItem) {
+        let tracking_changes = self.is_tracking_changes();
         if let Some(turn) = self.current_turn.as_mut()
             && turn.id == turn_id
         {
-            upsert_turn_item(&mut turn.items, item);
+            let changed_item = {
+                let item = upsert_turn_item(&mut turn.items, item);
+                tracking_changes.then(|| (turn.id.clone(), item.clone()))
+            };
+            if let Some((turn_id, item)) = changed_item {
+                self.record_changed_item(turn_id, item);
+            }
             return;
         }
 
         if let Some(turn) = self.turns.iter_mut().find(|turn| turn.id == turn_id) {
-            upsert_turn_item(&mut turn.items, item);
+            let changed_item = {
+                let item = upsert_turn_item(&mut turn.items, item);
+                tracking_changes.then(|| (turn.id.clone(), item.clone()))
+            };
+            if let Some((turn_id, item)) = changed_item {
+                self.record_changed_item(turn_id, item);
+            }
             return;
         }
 
@@ -1094,8 +1358,45 @@ impl ThreadHistoryBuilder {
     }
 
     fn upsert_item_in_current_turn(&mut self, item: ThreadItem) {
-        let turn = self.ensure_turn();
-        upsert_turn_item(&mut turn.items, item);
+        let tracking_changes = self.is_tracking_changes();
+        let changed_item = {
+            let turn = self.ensure_turn();
+            let item = upsert_turn_item(&mut turn.items, item);
+            tracking_changes.then(|| (turn.id.clone(), item.clone()))
+        };
+        if let Some((turn_id, item)) = changed_item {
+            self.record_changed_item(turn_id, item);
+        }
+    }
+
+    fn is_tracking_changes(&self) -> bool {
+        self.active_change_set.is_some()
+    }
+
+    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 });
+        }
+    }
+
+    fn record_changed_pending_turn(&mut self, turn: &PendingTurn) {
+        if self.is_tracking_changes() {
+            self.record_changed_turn(ThreadHistoryTurnChange::from_pending_turn(turn));
+        }
+    }
+
+    fn record_changed_turn(&mut self, turn: ThreadHistoryTurnChange) {
+        if let Some(change_set) = self.active_change_set.as_mut() {
+            change_set.changed_turns.push(turn);
+        }
+    }
+
+    fn record_removed_turn_ids(&mut self, removed_turn_ids: Vec<String>) {
+        if let Some(change_set) = self.active_change_set.as_mut() {
+            change_set.removed_turn_ids.extend(removed_turn_ids);
+        }
     }
 
     fn next_item_id(&mut self) -> String {
@@ -1163,15 +1464,17 @@ fn convert_dynamic_tool_content_items(
         .collect()
 }
 
-fn upsert_turn_item(items: &mut Vec<ThreadItem>, item: ThreadItem) {
-    if let Some(existing_item) = items
-        .iter_mut()
-        .find(|existing_item| existing_item.id() == item.id())
+fn upsert_turn_item(items: &mut Vec<ThreadItem>, item: ThreadItem) -> &ThreadItem {
+    if let Some(existing_item_index) = items
+        .iter()
+        .position(|existing_item| existing_item.id() == item.id())
     {
-        *existing_item = item;
-        return;
+        items[existing_item_index] = item;
+        return &items[existing_item_index];
     }
+    let inserted_item_index = items.len();
     items.push(item);
+    &items[inserted_item_index]
 }
 
 struct PendingTurn {
@@ -1275,6 +1578,7 @@ mod tests {
     use codex_protocol::protocol::TurnCompleteEvent;
     use codex_protocol::protocol::TurnStartedEvent;
     use codex_protocol::protocol::UserMessageEvent;
+    use codex_protocol::protocol::WebSearchBeginEvent;
     use codex_protocol::protocol::WebSearchEndEvent;
     use codex_utils_absolute_path::test_support::PathBufExt;
     use codex_utils_absolute_path::test_support::test_path_buf;
@@ -3457,4 +3761,299 @@ mod tests {
         assert_eq!(turns.len(), 1);
         assert!(turns[0].items.is_empty());
     }
+
+    #[test]
+    fn changed_rollout_item_reports_new_item_snapshot() {
+        let mut builder = ThreadHistoryBuilder::new();
+
+        let changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(
+            EventMsg::UserMessage(UserMessageEvent {
+                client_id: Some("client-message-1".into()),
+                message: "hello".into(),
+                images: None,
+                text_elements: Vec::new(),
+                local_images: Vec::new(),
+                ..Default::default()
+            }),
+        ));
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: vec![ThreadHistoryItemChange {
+                    turn_id: "rollout-0".into(),
+                    item: ThreadItem::UserMessage {
+                        id: "item-1".into(),
+                        client_id: Some("client-message-1".into()),
+                        content: vec![UserInput::Text {
+                            text: "hello".into(),
+                            text_elements: Vec::new(),
+                        }],
+                    },
+                }],
+                changed_turns: vec![ThreadHistoryTurnChange {
+                    turn_id: "rollout-0".into(),
+                    status: TurnStatus::Completed,
+                    error: None,
+                    started_at: None,
+                    completed_at: None,
+                    duration_ms: None,
+                }],
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_item_reports_updated_existing_item_snapshot() {
+        let mut builder = ThreadHistoryBuilder::new();
+        builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(EventMsg::WebSearchBegin(
+            WebSearchBeginEvent {
+                call_id: "search-1".into(),
+            },
+        )));
+
+        let changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(
+            EventMsg::WebSearchEnd(WebSearchEndEvent {
+                call_id: "search-1".into(),
+                query: "codex".into(),
+                action: CoreWebSearchAction::Search {
+                    query: Some("codex".into()),
+                    queries: None,
+                },
+            }),
+        ));
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: vec![ThreadHistoryItemChange {
+                    turn_id: "rollout-0".into(),
+                    item: ThreadItem::WebSearch {
+                        id: "search-1".into(),
+                        query: "codex".into(),
+                        action: Some(WebSearchAction::Search {
+                            query: Some("codex".into()),
+                            queries: None,
+                        }),
+                    },
+                }],
+                changed_turns: Vec::new(),
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_item_reports_streaming_item_mutation() {
+        let mut builder = ThreadHistoryBuilder::new();
+        builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(EventMsg::AgentReasoning(
+            AgentReasoningEvent {
+                text: "summary".into(),
+            },
+        )));
+
+        let changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(
+            EventMsg::AgentReasoningRawContent(AgentReasoningRawContentEvent {
+                text: "raw content".into(),
+            }),
+        ));
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: vec![ThreadHistoryItemChange {
+                    turn_id: "rollout-0".into(),
+                    item: ThreadItem::Reasoning {
+                        id: "item-1".into(),
+                        summary: vec!["summary".into()],
+                        content: vec!["raw content".into()],
+                    },
+                }],
+                changed_turns: Vec::new(),
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_item_reports_turn_completion_metadata() {
+        let mut builder = ThreadHistoryBuilder::new();
+
+        let start_changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(
+            EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id: "turn-a".into(),
+                trace_id: None,
+                started_at: Some(10),
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            }),
+        ));
+        assert_eq!(
+            start_changes,
+            ThreadHistoryChangeSet {
+                changed_items: Vec::new(),
+                changed_turns: vec![ThreadHistoryTurnChange {
+                    turn_id: "turn-a".into(),
+                    status: TurnStatus::InProgress,
+                    error: None,
+                    started_at: Some(10),
+                    completed_at: None,
+                    duration_ms: None,
+                }],
+                removed_turn_ids: Vec::new(),
+            }
+        );
+
+        builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(EventMsg::UserMessage(
+            UserMessageEvent {
+                client_id: None,
+                message: "hello".into(),
+                images: None,
+                text_elements: Vec::new(),
+                local_images: Vec::new(),
+                ..Default::default()
+            },
+        )));
+        let complete_changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg(
+            EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id: "turn-a".into(),
+                last_agent_message: None,
+                completed_at: Some(20),
+                duration_ms: Some(123),
+                time_to_first_token_ms: None,
+            }),
+        ));
+
+        assert_eq!(
+            complete_changes,
+            ThreadHistoryChangeSet {
+                changed_items: Vec::new(),
+                changed_turns: vec![ThreadHistoryTurnChange {
+                    turn_id: "turn-a".into(),
+                    status: TurnStatus::Completed,
+                    error: None,
+                    started_at: Some(10),
+                    completed_at: Some(20),
+                    duration_ms: Some(123),
+                }],
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_items_dedupe_updated_item_snapshots() {
+        let mut builder = ThreadHistoryBuilder::new();
+        let changes = builder.handle_rollout_items_with_changes(&[
+            RolloutItem::EventMsg(EventMsg::WebSearchBegin(WebSearchBeginEvent {
+                call_id: "search-1".into(),
+            })),
+            RolloutItem::EventMsg(EventMsg::WebSearchEnd(WebSearchEndEvent {
+                call_id: "search-1".into(),
+                query: "codex".into(),
+                action: CoreWebSearchAction::Search {
+                    query: Some("codex".into()),
+                    queries: None,
+                },
+            })),
+        ]);
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: vec![ThreadHistoryItemChange {
+                    turn_id: "rollout-0".into(),
+                    item: ThreadItem::WebSearch {
+                        id: "search-1".into(),
+                        query: "codex".into(),
+                        action: Some(WebSearchAction::Search {
+                            query: Some("codex".into()),
+                            queries: None,
+                        }),
+                    },
+                }],
+                changed_turns: vec![ThreadHistoryTurnChange {
+                    turn_id: "rollout-0".into(),
+                    status: TurnStatus::Completed,
+                    error: None,
+                    started_at: None,
+                    completed_at: None,
+                    duration_ms: None,
+                }],
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_items_dedupe_turn_metadata_snapshots() {
+        let mut builder = ThreadHistoryBuilder::new();
+        let changes = builder.handle_rollout_items_with_changes(&[
+            RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id: "turn-a".into(),
+                trace_id: None,
+                started_at: Some(10),
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            })),
+            RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id: "turn-a".into(),
+                last_agent_message: None,
+                completed_at: Some(20),
+                duration_ms: Some(123),
+                time_to_first_token_ms: None,
+            })),
+        ]);
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: Vec::new(),
+                changed_turns: vec![ThreadHistoryTurnChange {
+                    turn_id: "turn-a".into(),
+                    status: TurnStatus::Completed,
+                    error: None,
+                    started_at: Some(10),
+                    completed_at: Some(20),
+                    duration_ms: Some(123),
+                }],
+                removed_turn_ids: Vec::new(),
+            }
+        );
+    }
+
+    #[test]
+    fn changed_rollout_items_drop_prior_changes_for_removed_turns() {
+        let mut builder = ThreadHistoryBuilder::new();
+        let changes = builder.handle_rollout_items_with_changes(&[
+            RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id: "turn-a".into(),
+                trace_id: None,
+                started_at: None,
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            })),
+            RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
+                client_id: None,
+                message: "hello".into(),
+                images: None,
+                text_elements: Vec::new(),
+                local_images: Vec::new(),
+                ..Default::default()
+            })),
+            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
+                num_turns: 1,
+            })),
+        ]);
+
+        assert_eq!(
+            changes,
+            ThreadHistoryChangeSet {
+                changed_items: Vec::new(),
+                changed_turns: Vec::new(),
+                removed_turn_ids: vec!["turn-a".into()],
+            }
+        );
+    }
 }