mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
Revert "[codex] drain mailbox only at request boundaries" (#18325)
## Summary - Reverts PR #17749 so queued inter-agent mail can again preempt after reasoning/commentary output item boundaries. - Applies the revert to the current `codex/turn.rs` module layout and restores the prior pending-input test expectations/snapshots. ## Testing - `just fmt` - `cargo test -p codex-core --test all pending_input` - `cargo test -p codex-core` failed in unrelated `tools::js_repl::tests::js_repl_imported_local_files_can_access_repl_globals`: dotslash download hit `mktemp: mkdtemp failed ... Operation not permitted` in the sandbox temp dir. Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jif-oai
committed by
GitHub
Unverified
GitHub
parent
5bb193aa88
commit
e3c2acb9cd
@@ -78,6 +78,7 @@ use codex_protocol::items::UserMessageItem;
|
||||
use codex_protocol::items::build_hook_prompt_message;
|
||||
use codex_protocol::models::BaseInstructions;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::MessagePhase;
|
||||
use codex_protocol::models::ResponseInputItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::protocol::AgentMessageContentDeltaEvent;
|
||||
@@ -1944,6 +1945,25 @@ 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)
|
||||
@@ -1959,6 +1979,13 @@ 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 ResponseItem::CustomToolCall { call_id, name, .. } = &item {
|
||||
|
||||
@@ -321,7 +321,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_waits_for_request_boundary_after_reasoning_item() {
|
||||
async fn queued_inter_agent_mail_triggers_follow_up_after_reasoning_item() {
|
||||
let (gate_reasoning_done_tx, gate_reasoning_done_rx) = oneshot::channel();
|
||||
|
||||
let first_chunks = vec![
|
||||
@@ -331,18 +331,14 @@ async fn queued_inter_agent_mail_waits_for_request_boundary_after_reasoning_item
|
||||
gate_reasoning_done_rx,
|
||||
vec![
|
||||
ev_reasoning_item("reason-1", &["thinking"], &[]),
|
||||
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_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_completed("resp-1"),
|
||||
],
|
||||
),
|
||||
@@ -370,7 +366,7 @@ async fn queued_inter_agent_mail_waits_for_request_boundary_after_reasoning_item
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn queued_inter_agent_mail_waits_for_request_boundary_after_commentary_message_item() {
|
||||
async fn queued_inter_agent_mail_triggers_follow_up_after_commentary_message_item() {
|
||||
let (gate_message_done_tx, gate_message_done_rx) = oneshot::channel();
|
||||
|
||||
let first_chunks = vec![
|
||||
@@ -379,18 +375,25 @@ async fn queued_inter_agent_mail_waits_for_request_boundary_after_commentary_mes
|
||||
gated_chunk(
|
||||
gate_message_done_rx,
|
||||
vec![
|
||||
ev_output_text_delta("first commentary"),
|
||||
ev_output_text_delta("first answer"),
|
||||
json!({
|
||||
"type": "response.output_item.done",
|
||||
"item": {
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"id": "msg-1",
|
||||
"content": [{"type": "output_text", "text": "first commentary"}],
|
||||
"content": [{"type": "output_text", "text": "first answer"}],
|
||||
"phase": "commentary",
|
||||
}
|
||||
}),
|
||||
ev_function_call("call-preserved", "test_tool", "{}"),
|
||||
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_completed("resp-1"),
|
||||
],
|
||||
),
|
||||
@@ -416,7 +419,7 @@ async fn queued_inter_agent_mail_waits_for_request_boundary_after_commentary_mes
|
||||
|
||||
let _ = gate_message_done_tx.send(());
|
||||
|
||||
wait_for_agent_message(&codex, "first commentary").await;
|
||||
wait_for_agent_message(&codex, "first answer").await;
|
||||
|
||||
wait_for_turn_complete(&codex).await;
|
||||
|
||||
|
||||
+2
-4
@@ -13,7 +13,5 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot
|
||||
00:message/developer:<PERMISSIONS_INSTRUCTIONS>
|
||||
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
|
||||
02:message/user:first prompt
|
||||
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}
|
||||
03:message/assistant:first answer
|
||||
04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false}
|
||||
|
||||
+1
-2
@@ -14,5 +14,4 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot
|
||||
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
|
||||
02:message/user:first prompt
|
||||
03:reasoning:summary=thinking:encrypted=true
|
||||
04:message/assistant:preserved commentary
|
||||
05:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false}
|
||||
04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false}
|
||||
|
||||
Reference in New Issue
Block a user