Preserve review lifecycle ordering on failure (#50804)

## Why

A review can fail before the UI receives the event that enters review mode. Failed-turn completion also needs to preserve the running indicator when a queued `/review` has already been submitted and is awaiting startup.

## What changed

- Emit the review-entry lifecycle before starting the review task, after aborting existing tasks and clearing connector selection.
- When a live failed-turn completion repeats an already handled error, refresh task state only if no user turn is pending startup, and continue processing queued input.

## Testing

Add regression tests for lifecycle ordering on a server-overload response and for preserving the running indicator while a queued review awaits startup after a failed turn.

GitOrigin-RevId: 0f91d3d1a1a983542c54c2686630de9c232321b1

codex-rs/core/src/session/review.rs

@@ -165,11 +165,8 @@ pub(super) async fn spawn_review_thread(
         tc.turn_metadata_state
             .spawn_git_enrichment_task(Arc::clone(&sess.services.git_root_discovery));
     }
-    // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not
-    // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle
-    // (TurnStarted + TurnComplete) for consistency with other standalone tasks.
-    sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new())
-        .await;
+    sess.abort_all_tasks(TurnAbortReason::Replaced).await;
+    sess.clear_connector_selection().await;
 
     // Announce entering review mode so UIs can switch modes.
     let item = TurnItem::EnteredReviewMode(EnteredReviewModeItem {
@@ -179,4 +176,10 @@ pub(super) async fn spawn_review_thread(
     });
     sess.emit_turn_item_started(&tc, &item).await;
     sess.emit_turn_item_completed(&tc, item).await;
+
+    // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not
+    // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle
+    // (TurnStarted + TurnComplete) for consistency with other standalone tasks.
+    sess.start_task(Arc::clone(&tc), input, ReviewTask::new())
+        .await;
 }

codex-rs/core/tests/suite/review.rs

@@ -55,6 +55,7 @@ use tempfile::TempDir;
 use tokio::io::AsyncWriteExt as _;
 use uuid::Uuid;
 use wiremock::MockServer;
+use wiremock::ResponseTemplate;
 
 /// Verify that submitting `Op::Review` emits review item lifecycle,
 /// legacy review events, and TurnComplete when the model returns a structured review payload.
@@ -297,6 +298,51 @@ async fn review_op_emits_lifecycle_and_review_output() {
     server.verify().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_overload_preserves_lifecycle_order() {
+    skip_if_no_network!();
+
+    let server = start_mock_server().await;
+    responses::mount_response_once(
+        &server,
+        ResponseTemplate::new(503)
+            .set_body_json(serde_json::json!({ "error": { "code": "server_is_overloaded" } })),
+    )
+    .await;
+    let codex = new_conversation_for_server(&server, Arc::new(TempDir::new().unwrap()), |config| {
+        config.model_provider.request_max_retries = Some(0);
+        config.model_provider.stream_max_retries = Some(0);
+    })
+    .await;
+
+    codex
+        .submit(Op::Review {
+            review_request: ReviewRequest {
+                target: ReviewTarget::UncommittedChanges,
+                user_facing_hint: None,
+            },
+        })
+        .await
+        .unwrap();
+
+    let mut lifecycle = Vec::new();
+    loop {
+        let event = match wait_for_event(&codex, |_| true).await {
+            EventMsg::EnteredReviewMode(_) => "entered",
+            EventMsg::Error(_) => "error",
+            EventMsg::ExitedReviewMode(_) => "exited",
+            EventMsg::TurnComplete(_) => "complete",
+            _ => continue,
+        };
+        lifecycle.push(event);
+        if event == "complete" {
+            break;
+        }
+    }
+
+    assert_eq!(lifecycle, ["entered", "error", "exited", "complete"]);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn cancelled_review_does_not_forward_delegate_mcp_startup() {
     skip_if_no_network!();

codex-rs/tui/src/chatwidget/protocol.rs

@@ -512,6 +512,12 @@ impl ChatWidget {
                         == Some(&(notification.turn.id.clone(), error.message.clone()))
                     {
                         self.last_non_retry_error = None;
+                        if replay_kind.is_none() {
+                            if !self.input_queue.user_turn_pending_start {
+                                self.update_task_running_state();
+                            }
+                            self.maybe_send_next_queued_input();
+                        }
                     } else {
                         self.handle_non_retry_error(error.message, error.codex_error_info);
                     }

codex-rs/tui/src/chatwidget/tests/review_mode.rs

@@ -359,6 +359,44 @@ async fn review_restores_context_window_indicator() {
     assert!(!chat.review.is_review_mode);
 }
 
+#[tokio::test]
+async fn failed_turn_completion_preserves_queued_review_start() {
+    const ERROR: &str = "Selected model is at capacity. Please try a different model.";
+
+    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
+    chat.thread_id = Some(ThreadId::new());
+    handle_turn_started(&mut chat, "turn-1");
+    assert!(chat.queue_user_message_with_options(
+        UserMessage::from("/review check regressions"),
+        QueuedInputAction::ParseSlash,
+        Vec::new(),
+    ));
+    handle_error(&mut chat, ERROR, Some(CodexErrorInfo::ServerOverloaded));
+    assert_matches!(op_rx.try_recv(), Ok(Op::Review { .. }));
+    assert!(chat.input_queue.user_turn_pending_start);
+    assert!(chat.bottom_pane.is_task_running());
+
+    chat.handle_server_notification(
+        ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: chat.thread_id.map(|id| id.to_string()).unwrap_or_default(),
+            turn: app_server_turn(
+                "turn-1",
+                AppServerTurnStatus::Failed,
+                /*duration_ms*/ None,
+                Some(AppServerTurnError {
+                    message: ERROR.to_string(),
+                    codex_error_info: Some(CodexErrorInfo::ServerOverloaded),
+                    additional_details: None,
+                    misalignment: None,
+                }),
+            ),
+        }),
+        /*replay_kind*/ None,
+    );
+
+    assert!(chat.bottom_pane.is_task_running());
+}
+
 #[tokio::test]
 async fn restore_thread_input_state_restores_pending_steers_without_downgrading_them() {
     let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;