Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
owenlin0 merged 1 commit into from
Jul 7, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 23 additions & 29 deletions codex-rs/app-server/src/bespoke_event_handling.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(_)
Expand All @@ -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) => {
Expand Down Expand Up @@ -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),
Expand Down
2 changes: 0 additions & 2 deletions codex-rs/core/src/tools/handlers/multi_agents.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
103 changes: 76 additions & 27 deletions codex-rs/core/src/tools/handlers/multi_agents/wait.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand All @@ -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));
Expand Down Expand Up @@ -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()
Expand All @@ -198,23 +201,69 @@ 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;

Ok(boxed_tool_output(result))
}
}

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 { .. })
Expand Down
41 changes: 0 additions & 41 deletions codex-rs/core/src/tools/handlers/multi_agents_common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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" => {
Expand Down
5 changes: 3 additions & 2 deletions codex-rs/core/src/tools/handlers/multi_agents_v2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
37 changes: 22 additions & 15 deletions codex-rs/core/src/tools/handlers/multi_agents_v2/wait.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand All @@ -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;

Expand Down
Loading