From 05c582992359e47afaa298c045c62af42001a463 Mon Sep 17 00:00:00 2001 From: Thibault Sottiaux Date: Mon, 13 Apr 2026 22:09:51 -0700 Subject: [PATCH] [codex] drain mailbox only at request boundaries (#17749) This changes multi-agent v2 mailbox handling so incoming inter-agent messages no longer preempt an in-flight sampling stream at reasoning or commentary output-item boundaries. --- codex-rs/core/src/codex.rs | 27 ------------- codex-rs/core/tests/suite/pending_input.rs | 39 +++++++++---------- ...ng_input_queued_mail_after_commentary.snap | 6 ++- ...ing_input_queued_mail_after_reasoning.snap | 3 +- 4 files changed, 24 insertions(+), 51 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index bab91b177..2fd36376d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -338,7 +338,6 @@ use codex_protocol::config_types::ServiceTier; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::models::ContentItem; use codex_protocol::models::DeveloperInstructions; -use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; @@ -7836,25 +7835,6 @@ async fn try_run_sampling_request( cancellation_token: cancellation_token.child_token(), }; - let preempt_for_mailbox_mail = match &item { - ResponseItem::Message { role, phase, .. } => { - role == "assistant" && matches!(phase, Some(MessagePhase::Commentary)) - } - ResponseItem::Reasoning { .. } => true, - ResponseItem::LocalShellCall { .. } - | ResponseItem::FunctionCall { .. } - | ResponseItem::ToolSearchCall { .. } - | ResponseItem::FunctionCallOutput { .. } - | ResponseItem::CustomToolCall { .. } - | ResponseItem::CustomToolCallOutput { .. } - | ResponseItem::ToolSearchOutput { .. } - | ResponseItem::WebSearchCall { .. } - | ResponseItem::ImageGenerationCall { .. } - | ResponseItem::GhostSnapshot { .. } - | ResponseItem::Compaction { .. } - | ResponseItem::Other => false, - }; - let output_result = match handle_output_item_done(&mut ctx, item, previously_active_item) .instrument(handle_responses) @@ -7870,13 +7850,6 @@ async fn try_run_sampling_request( last_agent_message = Some(agent_message); } needs_follow_up |= output_result.needs_follow_up; - // todo: remove before stabilizing multi-agent v2 - if preempt_for_mailbox_mail && sess.mailbox_rx.lock().await.has_pending() { - break Ok(SamplingRequestResult { - needs_follow_up: true, - last_agent_message, - }); - } } ResponseEvent::OutputItemAdded(item) => { if let Some(turn_item) = handle_non_tool_response_item( diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index 406d6a3d5..719cc1f0a 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -313,7 +313,7 @@ async fn injected_user_input_triggers_follow_up_request_with_deltas() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn queued_inter_agent_mail_triggers_follow_up_after_reasoning_item() { +async fn queued_inter_agent_mail_waits_for_request_boundary_after_reasoning_item() { let (gate_reasoning_done_tx, gate_reasoning_done_rx) = oneshot::channel(); let first_chunks = vec![ @@ -323,14 +323,18 @@ async fn queued_inter_agent_mail_triggers_follow_up_after_reasoning_item() { gate_reasoning_done_rx, vec![ ev_reasoning_item("reason-1", &["thinking"], &[]), - ev_function_call( - "call-stale", - "shell", - r#"{"command":"echo stale tool call"}"#, - ), - ev_message_item_added("msg-stale", ""), - ev_output_text_delta("stale final"), - ev_message_item_done("msg-stale", "stale final"), + ev_message_item_added("msg-preserved", ""), + ev_output_text_delta("preserved commentary"), + json!({ + "type": "response.output_item.done", + "item": { + "type": "message", + "role": "assistant", + "id": "msg-preserved", + "content": [{"type": "output_text", "text": "preserved commentary"}], + "phase": "commentary", + } + }), ev_completed("resp-1"), ], ), @@ -358,7 +362,7 @@ async fn queued_inter_agent_mail_triggers_follow_up_after_reasoning_item() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn queued_inter_agent_mail_triggers_follow_up_after_commentary_message_item() { +async fn queued_inter_agent_mail_waits_for_request_boundary_after_commentary_message_item() { let (gate_message_done_tx, gate_message_done_rx) = oneshot::channel(); let first_chunks = vec![ @@ -367,25 +371,18 @@ async fn queued_inter_agent_mail_triggers_follow_up_after_commentary_message_ite gated_chunk( gate_message_done_rx, vec![ - ev_output_text_delta("first answer"), + ev_output_text_delta("first commentary"), json!({ "type": "response.output_item.done", "item": { "type": "message", "role": "assistant", "id": "msg-1", - "content": [{"type": "output_text", "text": "first answer"}], + "content": [{"type": "output_text", "text": "first commentary"}], "phase": "commentary", } }), - ev_function_call( - "call-stale", - "shell", - r#"{"command":"echo stale tool call"}"#, - ), - ev_message_item_added("msg-stale", ""), - ev_output_text_delta("stale final"), - ev_message_item_done("msg-stale", "stale final"), + ev_function_call("call-preserved", "test_tool", "{}"), ev_completed("resp-1"), ], ), @@ -411,7 +408,7 @@ async fn queued_inter_agent_mail_triggers_follow_up_after_commentary_message_ite let _ = gate_message_done_tx.send(()); - wait_for_agent_message(&codex, "first answer").await; + wait_for_agent_message(&codex, "first commentary").await; wait_for_turn_complete(&codex).await; diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap index e65a5f34f..d3889fcd3 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap @@ -13,5 +13,7 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot 00:message/developer: 01:message/user:> 02:message/user:first prompt -03:message/assistant:first answer -04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false} +03:message/assistant:first commentary +04:function_call/test_tool +05:function_call_output:unsupported call: test_tool +06:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap index 004196f97..8e23dfe52 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap @@ -14,4 +14,5 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot 01:message/user:> 02:message/user:first prompt 03:reasoning:summary=thinking:encrypted=true -04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false} +04:message/assistant:preserved commentary +05:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false}