feat(core): emit canonical review mode items (#31473)

## Description

This PR moves review-mode markers onto canonical `TurnItem` lifecycle:

- `TurnItem::EnteredReviewMode`
- `TurnItem::ExitedReviewMode`

Core now emits `ItemStarted` / `ItemCompleted` for both. The completed
items map back into the existing `EnteredReviewMode` /
`ExitedReviewMode` events, so raw core event consumers and legacy
rollout persistence keep seeing the old events.

This is the compatibility layer needed before paginated rollouts persist
review markers as `ItemCompleted(TurnItem)`.

## Why

Review markers were one of the remaining app-server thread items created
directly from legacy events. Giving them canonical items lets paginated
history persist stable turn/item IDs without changing legacy rollouts.

## What changed

- Added canonical review-mode `TurnItem`s and switched review flow to
emit their lifecycle.
- Added completed-item → legacy review event mappings with stable
turn/item IDs.
- Switched app-server live notifications to the generic canonical item
path and kept legacy replay compatible with old payloads.
- Updated `ThreadHistoryBuilder` to replay canonical review items even
though review turns still do not emit `TurnStarted`.

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

@@ -33,6 +33,9 @@ use codex_protocol::protocol::GuardianAssessmentAction;
 use codex_protocol::protocol::GuardianAssessmentEvent;
 use codex_protocol::protocol::PatchApplyBeginEvent;
 use codex_protocol::protocol::PatchApplyEndEvent;
+use codex_protocol::protocol::ReviewOutputEvent;
+use codex_protocol::review_format::REVIEW_FALLBACK_MESSAGE;
+use codex_protocol::review_format::render_review_output_text;
 use codex_shell_command::parse_command::parse_command;
 use codex_shell_command::parse_command::shlex_join;
 use codex_utils_path_uri::PathConvention;
@@ -41,6 +44,12 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 use tracing::warn;
 
+pub(crate) fn review_output_text(output: Option<&ReviewOutputEvent>) -> String {
+    output
+        .map(render_review_output_text)
+        .unwrap_or_else(|| REVIEW_FALLBACK_MESSAGE.to_string())
+}
+
 pub fn build_file_change_approval_request_item(
     payload: &ApplyPatchApprovalRequestEvent,
 ) -> ThreadItem {

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

@@ -4,6 +4,7 @@ use crate::protocol::item_builders::build_file_change_approval_request_item;
 use crate::protocol::item_builders::build_file_change_begin_item;
 use crate::protocol::item_builders::build_file_change_end_item;
 use crate::protocol::item_builders::build_item_from_guardian_event;
+use crate::protocol::item_builders::review_output_text;
 use crate::protocol::v2::CollabAgentState;
 use crate::protocol::v2::CollabAgentTool;
 use crate::protocol::v2::CollabAgentToolCallStatus;
@@ -49,7 +50,6 @@ use codex_protocol::protocol::McpToolCallBeginEvent;
 use codex_protocol::protocol::McpToolCallEndEvent;
 use codex_protocol::protocol::PatchApplyBeginEvent;
 use codex_protocol::protocol::PatchApplyEndEvent;
-use codex_protocol::protocol::ReviewOutputEvent;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::ThreadRolledBackEvent;
 use codex_protocol::protocol::TurnAbortedEvent;
@@ -59,6 +59,8 @@ use codex_protocol::protocol::UserMessageEvent;
 use codex_protocol::protocol::ViewImageToolCallEvent;
 use codex_protocol::protocol::WebSearchBeginEvent;
 use codex_protocol::protocol::WebSearchEndEvent;
+#[cfg(test)]
+use codex_protocol::review_format::REVIEW_FALLBACK_MESSAGE;
 use std::collections::HashMap;
 use tracing::warn;
 use uuid::Uuid;
@@ -590,6 +592,11 @@ impl ThreadHistoryBuilder {
         turn_id: &str,
         item: &codex_protocol::items::TurnItem,
     ) {
+        let is_review_mode_item = matches!(
+            item,
+            codex_protocol::items::TurnItem::EnteredReviewMode(_)
+                | codex_protocol::items::TurnItem::ExitedReviewMode(_)
+        );
         let should_upsert = match item {
             codex_protocol::items::TurnItem::Plan(plan) => !plan.text.is_empty(),
             codex_protocol::items::TurnItem::Sleep(_)
@@ -597,7 +604,9 @@ impl ThreadHistoryBuilder {
             | codex_protocol::items::TurnItem::DynamicToolCall(_)
             | codex_protocol::items::TurnItem::CollabAgentToolCall(_)
             | codex_protocol::items::TurnItem::SubAgentActivity(_)
-            | codex_protocol::items::TurnItem::Extension(_) => true,
+            | codex_protocol::items::TurnItem::Extension(_)
+            | codex_protocol::items::TurnItem::EnteredReviewMode(_)
+            | codex_protocol::items::TurnItem::ExitedReviewMode(_) => true,
             codex_protocol::items::TurnItem::UserMessage(_)
             | codex_protocol::items::TurnItem::HookPrompt(_)
             | codex_protocol::items::TurnItem::AgentMessage(_)
@@ -611,7 +620,12 @@ impl ThreadHistoryBuilder {
         };
 
         if should_upsert {
-            self.upsert_item_in_turn_id(turn_id, ThreadItem::from(item.clone()));
+            let item = ThreadItem::from(item.clone());
+            if is_review_mode_item {
+                self.upsert_review_mode_item(Some(turn_id), item);
+            } else {
+                self.upsert_item_in_turn_id(turn_id, item);
+            }
         }
     }
 
@@ -1108,26 +1122,55 @@ impl ThreadHistoryBuilder {
         self.push_item_in_current_turn(ThreadItem::ContextCompaction { id });
     }
 
-    fn handle_entered_review_mode(&mut self, payload: &codex_protocol::protocol::ReviewRequest) {
+    fn handle_entered_review_mode(
+        &mut self,
+        payload: &codex_protocol::protocol::EnteredReviewModeEvent,
+    ) {
         let review = payload
             .user_facing_hint
             .clone()
             .unwrap_or_else(|| "Review requested.".to_string());
-        let id = self.next_item_id();
-        self.push_item_in_current_turn(ThreadItem::EnteredReviewMode { id, review });
+        let id = payload
+            .item_id
+            .clone()
+            .unwrap_or_else(|| self.next_item_id());
+        self.upsert_review_mode_item(
+            payload.turn_id.as_deref(),
+            ThreadItem::EnteredReviewMode { id, review },
+        );
     }
 
     fn handle_exited_review_mode(
         &mut self,
         payload: &codex_protocol::protocol::ExitedReviewModeEvent,
     ) {
-        let review = payload
-            .review_output
+        let review = review_output_text(payload.review_output.as_ref());
+        let id = payload
+            .item_id
+            .clone()
+            .unwrap_or_else(|| self.next_item_id());
+        self.upsert_review_mode_item(
+            payload.turn_id.as_deref(),
+            ThreadItem::ExitedReviewMode { id, review },
+        );
+    }
+
+    fn upsert_review_mode_item(&mut self, turn_id: Option<&str>, item: ThreadItem) {
+        let Some(turn_id) = turn_id else {
+            self.upsert_item_in_current_turn(item);
+            return;
+        };
+        let current_turn_matches = self
+            .current_turn
             .as_ref()
-            .map(render_review_output_text)
-            .unwrap_or_else(|| REVIEW_FALLBACK_MESSAGE.to_string());
-        let id = self.next_item_id();
-        self.push_item_in_current_turn(ThreadItem::ExitedReviewMode { id, review });
+            .is_some_and(|turn| turn.id == turn_id);
+        if !current_turn_matches && !self.turns.iter().any(|turn| turn.id == turn_id) {
+            self.finish_current_turn();
+            let turn = self.new_turn(Some(turn_id.to_string()));
+            self.record_changed_pending_turn(&turn);
+            self.current_turn = Some(turn);
+        }
+        self.upsert_item_in_turn_id(turn_id, item);
     }
 
     fn handle_error(&mut self, payload: &ErrorEvent) {
@@ -1443,17 +1486,6 @@ impl ThreadHistoryBuilder {
     }
 }
 
-const REVIEW_FALLBACK_MESSAGE: &str = "Reviewer failed to output a response.";
-
-fn render_review_output_text(output: &ReviewOutputEvent) -> String {
-    let explanation = output.overall_explanation.trim();
-    if explanation.is_empty() {
-        REVIEW_FALLBACK_MESSAGE.to_string()
-    } else {
-        explanation.to_string()
-    }
-}
-
 fn convert_dynamic_tool_content_items(
     items: &[codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem],
 ) -> Vec<DynamicToolCallOutputContentItem> {
@@ -1558,6 +1590,8 @@ mod tests {
     use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynamicToolCallOutputContentItem;
     use codex_protocol::items::CommandExecutionItem as CoreCommandExecutionItem;
     use codex_protocol::items::CommandExecutionStatus as CoreCommandExecutionStatus;
+    use codex_protocol::items::EnteredReviewModeItem as CoreEnteredReviewModeItem;
+    use codex_protocol::items::ExitedReviewModeItem as CoreExitedReviewModeItem;
     use codex_protocol::items::HookPromptFragment as CoreHookPromptFragment;
     use codex_protocol::items::SleepItem as CoreSleepItem;
     use codex_protocol::items::TurnItem as CoreTurnItem;
@@ -1575,12 +1609,15 @@ mod tests {
     use codex_protocol::protocol::CodexErrorInfo;
     use codex_protocol::protocol::CompactedItem;
     use codex_protocol::protocol::DynamicToolCallResponseEvent;
+    use codex_protocol::protocol::EnteredReviewModeEvent;
     use codex_protocol::protocol::ExecCommandEndEvent;
     use codex_protocol::protocol::ExecCommandSource;
+    use codex_protocol::protocol::ExitedReviewModeEvent;
     use codex_protocol::protocol::ItemStartedEvent;
     use codex_protocol::protocol::McpInvocation;
     use codex_protocol::protocol::McpToolCallEndEvent;
     use codex_protocol::protocol::PatchApplyBeginEvent;
+    use codex_protocol::protocol::ReviewTarget;
     use codex_protocol::protocol::ThreadRolledBackEvent;
     use codex_protocol::protocol::TurnAbortReason;
     use codex_protocol::protocol::TurnAbortedEvent;
@@ -1705,6 +1742,111 @@ mod tests {
         );
     }
 
+    #[test]
+    fn review_mode_events_replay_persisted_ids() {
+        let events = vec![
+            EventMsg::EnteredReviewMode(EnteredReviewModeEvent {
+                target: ReviewTarget::Custom {
+                    instructions: "review this".into(),
+                },
+                user_facing_hint: Some("Review requested.".into()),
+                turn_id: Some("turn-1".into()),
+                item_id: Some("entered-review".into()),
+            }),
+            EventMsg::ExitedReviewMode(ExitedReviewModeEvent {
+                turn_id: Some("turn-1".into()),
+                item_id: Some("exited-review".into()),
+                review_output: None,
+            }),
+            EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id: "turn-1".into(),
+                last_agent_message: None,
+                completed_at: None,
+                duration_ms: None,
+                time_to_first_token_ms: None,
+            }),
+        ];
+
+        let mut builder = ThreadHistoryBuilder::new();
+        for event in &events {
+            builder.handle_event(event);
+        }
+        let turns = builder.finish();
+
+        assert_eq!(turns[0].id, "turn-1");
+        assert_eq!(
+            turns[0].items,
+            vec![
+                ThreadItem::EnteredReviewMode {
+                    id: "entered-review".into(),
+                    review: "Review requested.".into(),
+                },
+                ThreadItem::ExitedReviewMode {
+                    id: "exited-review".into(),
+                    review: REVIEW_FALLBACK_MESSAGE.into(),
+                },
+            ]
+        );
+    }
+
+    #[test]
+    fn review_mode_items_replay_without_turn_started() {
+        let thread_id = ThreadId::new();
+        let entered = CoreTurnItem::EnteredReviewMode(CoreEnteredReviewModeItem {
+            id: "entered-review".into(),
+            target: ReviewTarget::Custom {
+                instructions: "review this".into(),
+            },
+            user_facing_hint: "Review requested.".into(),
+        });
+        let exited = CoreTurnItem::ExitedReviewMode(CoreExitedReviewModeItem {
+            id: "exited-review".into(),
+            review_output: None,
+        });
+        let events = vec![
+            EventMsg::ItemCompleted(ItemCompletedEvent {
+                thread_id,
+                turn_id: "turn-1".into(),
+                item: entered,
+                completed_at_ms: 0,
+            }),
+            EventMsg::ItemCompleted(ItemCompletedEvent {
+                thread_id,
+                turn_id: "turn-1".into(),
+                item: exited,
+                completed_at_ms: 0,
+            }),
+            EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id: "turn-1".into(),
+                last_agent_message: None,
+                completed_at: None,
+                duration_ms: None,
+                time_to_first_token_ms: None,
+            }),
+        ];
+
+        let mut builder = ThreadHistoryBuilder::new();
+        for event in &events {
+            builder.handle_event(event);
+        }
+        let turns = builder.finish();
+
+        assert_eq!(turns[0].id, "turn-1");
+        assert_eq!(
+            turns[0].items,
+            vec![
+                ThreadItem::EnteredReviewMode {
+                    id: "entered-review".into(),
+                    review: "Review requested.".into(),
+                },
+                ThreadItem::ExitedReviewMode {
+                    id: "exited-review".into(),
+                    review: REVIEW_FALLBACK_MESSAGE.into(),
+                },
+            ]
+        );
+    }
+
     #[test]
     fn rebuilds_user_message_image_details_from_legacy_events() {
         let local_path = PathBuf::from("/tmp/local.png");
@@ -3932,7 +4074,6 @@ mod tests {
                 ..Default::default()
             }),
         ));
-
         assert_eq!(
             changes,
             ThreadHistoryChangeSet {
@@ -3979,7 +4120,6 @@ mod tests {
                 },
             }),
         ));
-
         assert_eq!(
             changes,
             ThreadHistoryChangeSet {
@@ -4014,7 +4154,6 @@ mod tests {
                 text: "raw content".into(),
             }),
         ));
-
         assert_eq!(
             changes,
             ThreadHistoryChangeSet {
@@ -4114,7 +4253,6 @@ mod tests {
                 },
             })),
         ]);
-
         assert_eq!(
             changes,
             ThreadHistoryChangeSet {

codex-rs/app-server-protocol/src/protocol/v2/item.rs

@@ -10,6 +10,7 @@ use super::UserInput;
 use super::shared::v2_enum_from_core;
 use crate::protocol::item_builders::command_actions_for_path_uri;
 use crate::protocol::item_builders::convert_patch_changes;
+use crate::protocol::item_builders::review_output_text;
 use codex_experimental_api_macros::ExperimentalApi;
 use codex_extension_items::ExtensionItem;
 pub use codex_extension_items::image_generation::ImageGenerationItem;
@@ -923,6 +924,14 @@ impl From<CoreTurnItem> for ThreadItem {
                     saved_path: image.saved_path,
                 })
             }
+            CoreTurnItem::EnteredReviewMode(review) => ThreadItem::EnteredReviewMode {
+                id: review.id,
+                review: review.user_facing_hint,
+            },
+            CoreTurnItem::ExitedReviewMode(review) => ThreadItem::ExitedReviewMode {
+                id: review.id,
+                review: review_output_text(review.review_output.as_ref()),
+            },
             CoreTurnItem::FileChange(file_change) => ThreadItem::FileChange {
                 id: file_change.id,
                 changes: convert_patch_changes(&file_change.changes),