Fix cancelled review leaving MCP startup busy (#31189)

## Summary

Cancelling an inline review could leave the TUI stuck in MCP startup
state, so subsequent `/review` commands were rejected as though another
task were still running. Specifically, the child’s “Starting MCP
servers” event was incorrectly forwarded to the parent TUI, so the
parent was marking itself as busy.

This keeps delegate-session MCP startup events inside the delegate
instead of exposing them as parent-session state.

## Reproduction

Here's the before -- notice that after I cancel the first review, I'm
prevented from running `/review` again:


https://github.com/user-attachments/assets/571a0793-3253-4bcc-8f10-3782d176162f

And the after -- notice that after I cancel, I can immediately start a
new `/review`:


https://github.com/user-attachments/assets/dd98b1d7-6a71-4a1f-a4b2-3295bc4848e5

codex-rs/core/src/codex_delegate.rs

@@ -299,11 +299,11 @@ async fn forward_events(
                 match event {
                     Event {
                         id: _,
-                        msg: EventMsg::TokenCount(_),
-                    } => {}
-                    Event {
-                        id: _,
-                        msg: EventMsg::SessionConfigured(_),
+                        msg:
+                            EventMsg::TokenCount(_)
+                            | EventMsg::SessionConfigured(_)
+                            | EventMsg::McpStartupUpdate(_)
+                            | EventMsg::McpStartupComplete(_),
                     } => {}
                     Event {
                         id,

codex-rs/core/src/codex_delegate_tests.rs

@@ -14,6 +14,9 @@ use codex_protocol::protocol::GuardianAssessmentAction;
 use codex_protocol::protocol::GuardianAssessmentStatus;
 use codex_protocol::protocol::GuardianCommandSource;
 use codex_protocol::protocol::McpInvocation;
+use codex_protocol::protocol::McpStartupCompleteEvent;
+use codex_protocol::protocol::McpStartupStatus;
+use codex_protocol::protocol::McpStartupUpdateEvent;
 use codex_protocol::protocol::RawResponseItemEvent;
 use codex_protocol::protocol::ReviewDecision;
 use codex_protocol::protocol::TurnAbortReason;
@@ -34,8 +37,8 @@ use tokio::sync::watch;
 use tokio::time::timeout;
 
 #[tokio::test]
-async fn forward_events_cancelled_while_send_blocked_shuts_down_delegate() {
-    let (tx_events, rx_events) = bounded(1);
+async fn forward_events_filters_private_events_before_blocked_send_is_cancelled() {
+    let (tx_events, rx_events) = bounded(SUBMISSION_CHANNEL_CAPACITY);
     let (tx_sub, rx_sub) = bounded(SUBMISSION_CHANNEL_CAPACITY);
     let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit);
     let (session, ctx, _rx_evt) = crate::session::tests::make_session_and_context_with_rx().await;
@@ -71,33 +74,53 @@ async fn forward_events_cancelled_while_send_blocked_shuts_down_delegate() {
         cancel.clone(),
     ));
 
-    tx_events
-        .send(Event {
-            id: "evt".to_string(),
-            msg: EventMsg::RawResponseItem(RawResponseItemEvent {
-                item: ResponseItem::CustomToolCall {
-                    id: None,
-                    status: None,
-                    call_id: "call-1".to_string(),
-                    name: "tool".to_string(),
-                    namespace: None,
-                    input: "{}".to_string(),
-                    internal_chat_message_metadata_passthrough: None,
-                },
-            }),
-        })
-        .await
-        .unwrap();
+    for msg in [
+        EventMsg::McpStartupUpdate(McpStartupUpdateEvent {
+            server: "pending".to_string(),
+            status: McpStartupStatus::Starting,
+        }),
+        EventMsg::McpStartupComplete(McpStartupCompleteEvent::default()),
+    ] {
+        tx_events
+            .send(Event {
+                id: "delegate-startup".to_string(),
+                msg,
+            })
+            .await
+            .unwrap();
+    }
+    let visible_msg = EventMsg::RawResponseItem(RawResponseItemEvent {
+        item: ResponseItem::CustomToolCall {
+            id: None,
+            status: None,
+            call_id: "call-1".to_string(),
+            name: "tool".to_string(),
+            namespace: None,
+            input: "{}".to_string(),
+            internal_chat_message_metadata_passthrough: None,
+        },
+    });
+    for id in ["visible-1", "visible-2", "blocked"] {
+        tx_events
+            .send(Event {
+                id: id.to_string(),
+                msg: visible_msg.clone(),
+            })
+            .await
+            .unwrap();
+    }
 
     drop(tx_events);
+    let received = rx_out.recv().await.expect("prefilled event missing");
+    assert_eq!(received.id, "full");
+    let received = rx_out.recv().await.expect("visible event missing");
+    assert_eq!(received.id, "visible-1");
     cancel.cancel();
     timeout(std::time::Duration::from_millis(1000), forward)
         .await
         .expect("forward_events hung")
         .expect("forward_events join error");
 
-    let received = rx_out.recv().await.expect("prefilled event missing");
-    assert_eq!("full", received.id);
     let mut ops = Vec::new();
     while let Ok(sub) = rx_sub.try_recv() {
         ops.push(sub.op);

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

@@ -29,6 +29,7 @@ use core_test_support::wait_for_event;
 use pretty_assertions::assert_eq;
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::time::Duration;
 use tempfile::TempDir;
 use tokio::io::AsyncWriteExt as _;
 use uuid::Uuid;
@@ -200,6 +201,95 @@ async fn review_op_emits_lifecycle_and_review_output() {
     server.verify().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn cancelled_review_does_not_forward_delegate_mcp_startup() {
+    skip_if_no_network!();
+
+    let server = start_mock_server().await;
+    let request_log = responses::mount_response_once(
+        &server,
+        responses::sse_response(responses::sse(vec![responses::ev_response_created(
+            "resp-1",
+        )]))
+        .set_delay(Duration::from_secs(30)),
+    )
+    .await;
+    let codex_home = Arc::new(TempDir::new().unwrap());
+    let codex = new_conversation_for_server(&server, codex_home.clone(), |_| {}).await;
+
+    // Consume the parent session's own empty startup round before starting the review.
+    wait_for_event(&codex, |event| {
+        matches!(event, EventMsg::McpStartupComplete(_))
+    })
+    .await;
+
+    codex
+        .submit(Op::Review {
+            review_request: ReviewRequest {
+                target: ReviewTarget::Custom {
+                    instructions: "Cancel this review".to_string(),
+                },
+                user_facing_hint: None,
+            },
+        })
+        .await
+        .unwrap();
+
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            match codex.next_event().await.expect("review event").msg {
+                event @ (EventMsg::McpStartupUpdate(_) | EventMsg::McpStartupComplete(_)) => {
+                    panic!("review forwarded delegate MCP startup: {event:?}")
+                }
+                EventMsg::EnteredReviewMode(_) => break,
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("timed out waiting for review entry");
+
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while request_log.requests().is_empty() {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("review request did not reach the server");
+
+    codex.submit(Op::Interrupt).await.unwrap();
+
+    let mut exited_review = false;
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            match codex
+                .next_event()
+                .await
+                .expect("review cancellation event")
+                .msg
+            {
+                event @ (EventMsg::McpStartupUpdate(_) | EventMsg::McpStartupComplete(_)) => {
+                    panic!("cancelled review forwarded delegate MCP startup: {event:?}")
+                }
+                EventMsg::ExitedReviewMode(ExitedReviewModeEvent { review_output }) => {
+                    assert_eq!(review_output, None);
+                    exited_review = true;
+                }
+                EventMsg::TurnAborted(_) if exited_review => break,
+                EventMsg::TurnAborted(_) => panic!("review turn aborted before review mode exited"),
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("timed out waiting for review cancellation");
+
+    assert_eq!(request_log.requests().len(), 1);
+
+    let _codex_home_guard = codex_home;
+    server.verify().await;
+}
+
 /// When the model returns plain text that is not JSON, ensure the child
 /// lifecycle still occurs and the plain text is surfaced via
 /// ExitedReviewMode(Some(..)) as the overall_explanation.