diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index a671f0dd74..4cba4bddc5 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/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, diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 9513dcc56c..5cb2251a4a 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/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); diff --git a/codex-rs/core/tests/suite/review.rs b/codex-rs/core/tests/suite/review.rs index af44baf6ca..e403b7e1ff 100644 --- a/codex-rs/core/tests/suite/review.rs +++ b/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.