feat(core): emit canonical collab wait items (#31301)

This PR depends on [#31296](https://github.com/openai/codex/pull/31296) for the canonical-to-legacy event mappings. ## Description This PR makes the v1 and v2 wait paths emit canonical `TurnItem::CollabAgentToolCall` lifecycle instead of `CollabWaitingBegin` / `CollabWaitingEnd` directly. Both paths already used the same legacy waiting events before this PR. The v1 item carries receiver metadata and final agent statuses for its target agents; v2 waits for mailbox activity rather than specific agents, so it keeps those fields empty, matching the existing v2 legacy payload. App-server v2 consumes the canonical item directly and ignores the mapped legacy wait events. ## Why Wait is separate from the other collab tools because it is multi-target and has distinct timeout/status behavior. Keeping it last also lets this PR remove the old helper that only existed to shape legacy wait status entries in core. ## What changed - Emit canonical collab wait items from both v1 and v2 wait handlers. - Preserve receiver metadata and agent status snapshots on completed wait items. - Remove the old core helper for building legacy wait status entries. ## Follow-up The next stack PR, [#30188](https://github.com/openai/codex/pull/30188), writes canonical `TurnItem` values to paginated rollout files.
2026-08-23 13:09:46 +00:00 · 2026-07-07 13:59:02 -07:00
parent 058d97c5dc
commit 6b4882528e
6 changed files with 124 additions and 116 deletions
--- a/codex-rs/app-server/src/bespoke_event_handling.rs
+++ b/codex-rs/app-server/src/bespoke_event_handling.rs
@@ -823,30 +823,32 @@ pub(crate) async fn apply_bespoke_event_handling(
                on_request_permissions_response(pending_response, conversation, thread_state).await;
            });
        }
-        EventMsg::DynamicToolCallRequest(_) | EventMsg::DynamicToolCallResponse(_) => {
-            // Deprecated dynamic-tool events are still fanned out for raw-event and rollout
-            // compatibility consumers. App-server v2 receives the canonical DynamicToolCall
-            // item lifecycle and dispatches client requests from canonical starts instead.
-        }
-        EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_) => {
-            // Deprecated MCP tool-call events are still fanned out for legacy clients.
-            // App-server v2 receives the canonical TurnItem::McpToolCall lifecycle instead.
-        }
-        EventMsg::CollabAgentSpawnBegin(_)
+        EventMsg::DynamicToolCallRequest(_)
+        | EventMsg::DynamicToolCallResponse(_)
+        | EventMsg::CollabAgentSpawnBegin(_)
        | EventMsg::CollabAgentSpawnEnd(_)
        | EventMsg::CollabAgentInteractionBegin(_)
        | EventMsg::CollabAgentInteractionEnd(_)
+        | EventMsg::CollabWaitingBegin(_)
+        | EventMsg::CollabWaitingEnd(_)
        | EventMsg::CollabCloseBegin(_)
        | EventMsg::CollabCloseEnd(_)
        | EventMsg::CollabResumeBegin(_)
-        | EventMsg::CollabResumeEnd(_) => {
-            // Deprecated non-wait collaboration events are still fanned out for raw-event and
-            // rollout compatibility consumers. App-server v2 receives the canonical
-            // CollabAgentToolCall item lifecycle instead.
+        | EventMsg::CollabResumeEnd(_)
+        | EventMsg::SubAgentActivity(_)
+        | EventMsg::ExecCommandBegin(_)
+        | EventMsg::ExecCommandEnd(_) => {
+            // Deprecated item lifecycle events are still fanned out for raw-event and rollout
+            // compatibility consumers.
+            // App-server v2 receives canonical TurnItem lifecycle instead, and dispatches
+            // dynamic tool requests from canonical DynamicToolCall starts.
        }
-        msg @ (EventMsg::CollabWaitingBegin(_)
-        | EventMsg::CollabWaitingEnd(_)
-        | EventMsg::AgentMessageContentDelta(_)
+        EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_) => {
+            // Deprecated MCP tool-call events are still fanned out for raw-event and rollout
+            // compatibility consumers.
+            // App-server v2 receives the canonical TurnItem::McpToolCall lifecycle instead.
+        }
+        msg @ (EventMsg::AgentMessageContentDelta(_)
        | EventMsg::PlanDelta(_)
        | EventMsg::ReasoningContentDelta(_)
        | EventMsg::ReasoningRawContentDelta(_)
@@ -858,13 +860,9 @@ pub(crate) async fn apply_bespoke_event_handling(
            );
            outgoing.send_server_notification(notification).await;
        }
-        EventMsg::SubAgentActivity(_) => {
-            // Deprecated sub-agent activity events are still fanned out for raw-event and
-            // rollout compatibility consumers. App-server v2 receives the canonical
-            // SubAgentActivity item lifecycle instead.
-        }
        EventMsg::ContextCompacted(..) => {
-            // Core still fans out this deprecated event for legacy clients;
+            // Core still fans out this deprecated event for raw-event and rollout compatibility
+            // consumers;
            // v2 clients receive the canonical ContextCompaction item instead.
        }
        EventMsg::DeprecationNotice(event) => {
@@ -1096,14 +1094,10 @@ pub(crate) async fn apply_bespoke_event_handling(
            .await;
        }
        EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) => {
-            // Core still fans out these deprecated events for legacy clients;
+            // Core still fans out these deprecated events for raw-event and rollout compatibility
+            // consumers;
            // v2 clients receive the canonical FileChange item instead.
        }
-        EventMsg::ExecCommandBegin(_) | EventMsg::ExecCommandEnd(_) => {
-            // Deprecated command-execution events are still fanned out for raw-event and rollout
-            // compatibility consumers. App-server v2 receives the canonical CommandExecution
-            // item lifecycle instead.
-        }
        EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event) => {
            let notification = item_event_to_server_notification(
                EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event),
--- a/codex-rs/core/src/tools/handlers/multi_agents.rs
+++ b/codex-rs/core/src/tools/handlers/multi_agents.rs
@@ -25,8 +25,6 @@ use codex_protocol::items::TurnItem;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::openai_models::ReasoningEffort;
 use codex_protocol::protocol::CollabAgentRef;
-use codex_protocol::protocol::CollabWaitingBeginEvent;
-use codex_protocol::protocol::CollabWaitingEndEvent;
 use codex_protocol::user_input::UserInput;
 use codex_tools::ToolName;
 use codex_tools::ToolSearchInfo;
--- a/codex-rs/core/src/tools/handlers/multi_agents/wait.rs
+++ b/codex-rs/core/src/tools/handlers/multi_agents/wait.rs
@@ -3,7 +3,6 @@ use crate::agent::status::is_final;
 use crate::session::session::Session;
 use crate::tools::handlers::multi_agents_spec::WaitAgentTimeoutOptions;
 use crate::tools::handlers::multi_agents_spec::create_wait_agent_tool_v1;
-use crate::turn_timing::now_unix_timestamp_ms;
 use codex_protocol::error::CodexErr;
 use codex_tools::ToolSpec;
 use futures::FutureExt;
@@ -98,16 +97,20 @@ impl Handler {
        };

        session
-            .send_event(
+            .emit_turn_item_started(
                &turn,
-                CollabWaitingBeginEvent {
-                    started_at_ms: now_unix_timestamp_ms(),
+                &TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
+                    id: call_id.clone(),
+                    tool: CollabAgentTool::Wait,
+                    status: CollabAgentToolCallStatus::InProgress,
                    sender_thread_id: session.thread_id,
                    receiver_thread_ids: receiver_thread_ids.clone(),
                    receiver_agents: receiver_agents.clone(),
-                    call_id: call_id.clone(),
-                }
-                .into(),
+                    prompt: None,
+                    model: None,
+                    reasoning_effort: None,
+                    agents_states: Default::default(),
+                }),
            )
            .await;

@@ -129,19 +132,20 @@ impl Handler {
                    let mut statuses = HashMap::with_capacity(1);
                    statuses.insert(*id, session.services.agent_control.get_status(*id).await);
                    session
-                        .send_event(
+                        .emit_turn_item_completed(
                            &turn,
-                            CollabWaitingEndEvent {
+                            TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
+                                id: call_id.clone(),
+                                tool: CollabAgentTool::Wait,
+                                status: wait_tool_call_status(&statuses),
                                sender_thread_id: session.thread_id,
-                                call_id: call_id.clone(),
-                                completed_at_ms: now_unix_timestamp_ms(),
-                                agent_statuses: build_wait_agent_statuses(
-                                    &statuses,
-                                    &receiver_agents,
-                                ),
-                                statuses,
-                            }
-                            .into(),
+                                receiver_thread_ids: statuses.keys().copied().collect(),
+                                receiver_agents: wait_receiver_agents(&statuses, &receiver_agents),
+                                prompt: None,
+                                model: None,
+                                reasoning_effort: None,
+                                agents_states: statuses,
+                            }),
                        )
                        .await;
                    return Err(collab_agent_error(*id, err));
@@ -183,7 +187,6 @@ impl Handler {

        let timed_out = statuses.is_empty();
        let statuses_by_id = statuses.clone().into_iter().collect::<HashMap<_, _>>();
-        let agent_statuses = build_wait_agent_statuses(&statuses_by_id, &receiver_agents);
        let result = WaitAgentResult {
            status: statuses
                .into_iter()
@@ -198,16 +201,20 @@ impl Handler {
        };

        session
-            .send_event(
+            .emit_turn_item_completed(
                &turn,
-                CollabWaitingEndEvent {
+                TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
+                    id: call_id,
+                    tool: CollabAgentTool::Wait,
+                    status: wait_tool_call_status(&statuses_by_id),
                    sender_thread_id: session.thread_id,
-                    call_id,
-                    completed_at_ms: now_unix_timestamp_ms(),
-                    agent_statuses,
-                    statuses: statuses_by_id,
-                }
-                .into(),
+                    receiver_thread_ids: statuses_by_id.keys().copied().collect(),
+                    receiver_agents: wait_receiver_agents(&statuses_by_id, &receiver_agents),
+                    prompt: None,
+                    model: None,
+                    reasoning_effort: None,
+                    agents_states: statuses_by_id,
+                }),
            )
            .await;

@@ -215,6 +222,48 @@ impl Handler {
    }
 }

+fn wait_tool_call_status(statuses: &HashMap<ThreadId, AgentStatus>) -> CollabAgentToolCallStatus {
+    if statuses
+        .values()
+        .any(|status| matches!(status, AgentStatus::Errored(_) | AgentStatus::NotFound))
+    {
+        CollabAgentToolCallStatus::Failed
+    } else {
+        CollabAgentToolCallStatus::Completed
+    }
+}
+
+fn wait_receiver_agents(
+    statuses: &HashMap<ThreadId, AgentStatus>,
+    receiver_agents: &[CollabAgentRef],
+) -> Vec<CollabAgentRef> {
+    if statuses.is_empty() {
+        return Vec::new();
+    }
+
+    let mut agents = Vec::with_capacity(statuses.len());
+    let mut seen = HashMap::with_capacity(receiver_agents.len());
+    for receiver_agent in receiver_agents {
+        seen.insert(receiver_agent.thread_id, ());
+        if statuses.contains_key(&receiver_agent.thread_id) {
+            agents.push(receiver_agent.clone());
+        }
+    }
+
+    let mut extras = statuses
+        .keys()
+        .filter(|thread_id| !seen.contains_key(thread_id))
+        .map(|thread_id| CollabAgentRef {
+            thread_id: *thread_id,
+            agent_nickname: None,
+            agent_role: None,
+        })
+        .collect::<Vec<_>>();
+    extras.sort_by_key(|agent| agent.thread_id.to_string());
+    agents.extend(extras);
+    agents
+}
+
 impl CoreToolRuntime for Handler {
    fn matches_kind(&self, payload: &ToolPayload) -> bool {
        matches!(payload, ToolPayload::Function { .. })
--- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs
+++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs
@@ -1,4 +1,3 @@
-use crate::agent::AgentStatus;
 use crate::config::Config;
 use crate::config::DEFAULT_MULTI_AGENT_V2_MIN_WAIT_TIMEOUT_MS;
 use crate::config::HARD_MAX_MULTI_AGENT_V2_TIMEOUT_MS;
@@ -16,14 +15,11 @@ use codex_protocol::models::BaseInstructions;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::openai_models::ReasoningEffort;
 use codex_protocol::openai_models::ReasoningEffortPreset;
-use codex_protocol::protocol::CollabAgentRef;
-use codex_protocol::protocol::CollabAgentStatusEntry;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::SubAgentSource;
 use codex_protocol::user_input::UserInput;
 use serde::Serialize;
 use serde_json::Value as JsonValue;
-use std::collections::HashMap;

 /// Minimum wait timeout to prevent tight polling loops from burning CPU.
 pub(crate) const MIN_WAIT_TIMEOUT_MS: i64 = DEFAULT_MULTI_AGENT_V2_MIN_WAIT_TIMEOUT_MS;
@@ -71,43 +67,6 @@ where
    })
 }

-pub(crate) fn build_wait_agent_statuses(
-    statuses: &HashMap<ThreadId, AgentStatus>,
-    receiver_agents: &[CollabAgentRef],
-) -> Vec<CollabAgentStatusEntry> {
-    if statuses.is_empty() {
-        return Vec::new();
-    }
-
-    let mut entries = Vec::with_capacity(statuses.len());
-    let mut seen = HashMap::with_capacity(receiver_agents.len());
-    for receiver_agent in receiver_agents {
-        seen.insert(receiver_agent.thread_id, ());
-        if let Some(status) = statuses.get(&receiver_agent.thread_id) {
-            entries.push(CollabAgentStatusEntry {
-                thread_id: receiver_agent.thread_id,
-                agent_nickname: receiver_agent.agent_nickname.clone(),
-                agent_role: receiver_agent.agent_role.clone(),
-                status: status.clone(),
-            });
-        }
-    }
-
-    let mut extras = statuses
-        .iter()
-        .filter(|(thread_id, _)| !seen.contains_key(thread_id))
-        .map(|(thread_id, status)| CollabAgentStatusEntry {
-            thread_id: *thread_id,
-            agent_nickname: None,
-            agent_role: None,
-            status: status.clone(),
-        })
-        .collect::<Vec<_>>();
-    extras.sort_by_key(|entry| entry.thread_id.to_string());
-    entries.extend(extras);
-    entries
-}
-
 pub(crate) fn collab_spawn_error(err: CodexErr) -> FunctionCallError {
    match err {
        CodexErr::UnsupportedOperation(message) if message == "thread manager dropped" => {
--- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs
+++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs
@@ -12,12 +12,13 @@ use crate::tools::handlers::parse_arguments;
 use crate::tools::registry::CoreToolRuntime;
 use crate::tools::registry::ToolExecutor;
 use codex_protocol::AgentPath;
+use codex_protocol::items::CollabAgentTool;
+use codex_protocol::items::CollabAgentToolCallItem;
+use codex_protocol::items::CollabAgentToolCallStatus;
 use codex_protocol::items::SubAgentActivityItem;
 use codex_protocol::items::TurnItem;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::openai_models::ReasoningEffort;
-use codex_protocol::protocol::CollabWaitingBeginEvent;
-use codex_protocol::protocol::CollabWaitingEndEvent;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::SubAgentActivityKind;
 use codex_tools::ToolName;
--- a/codex-rs/core/src/tools/handlers/multi_agents_v2/wait.rs
+++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/wait.rs
@@ -2,7 +2,6 @@ use super::*;
 use crate::session::InputQueueActivity;
 use crate::tools::handlers::multi_agents_spec::WaitAgentTimeoutOptions;
 use crate::tools::handlers::multi_agents_spec::create_wait_agent_tool_v2;
-use crate::turn_timing::now_unix_timestamp_ms;
 use codex_tools::ToolSpec;
 use std::collections::HashMap;
 use std::time::Duration;
@@ -76,16 +75,20 @@ impl Handler {
            .await;

        session
-            .send_event(
+            .emit_turn_item_started(
                &turn,
-                CollabWaitingBeginEvent {
-                    started_at_ms: now_unix_timestamp_ms(),
+                &TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
+                    id: call_id.clone(),
+                    tool: CollabAgentTool::Wait,
+                    status: CollabAgentToolCallStatus::InProgress,
                    sender_thread_id: session.thread_id,
                    receiver_thread_ids: Vec::new(),
                    receiver_agents: Vec::new(),
-                    call_id: call_id.clone(),
-                }
-                .into(),
+                    prompt: None,
+                    model: None,
+                    reasoning_effort: None,
+                    agents_states: Default::default(),
+                }),
            )
            .await;

@@ -94,16 +97,20 @@ impl Handler {
        let result = WaitAgentResult::from_outcome(outcome);

        session
-            .send_event(
+            .emit_turn_item_completed(
                &turn,
-                CollabWaitingEndEvent {
+                TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
+                    id: call_id,
+                    tool: CollabAgentTool::Wait,
+                    status: CollabAgentToolCallStatus::Completed,
                    sender_thread_id: session.thread_id,
-                    call_id,
-                    completed_at_ms: now_unix_timestamp_ms(),
-                    agent_statuses: Vec::new(),
-                    statuses: HashMap::new(),
-                }
-                .into(),
+                    receiver_thread_ids: Vec::new(),
+                    receiver_agents: Vec::new(),
+                    prompt: None,
+                    model: None,
+                    reasoning_effort: None,
+                    agents_states: HashMap::new(),
+                }),
            )
            .await;