feat(core): emit canonical collab tool call items (#31300)

This PR depends on [#31296](https://github.com/openai/codex/pull/31296)
for the canonical-to-legacy event mappings.

## Description

This PR makes the non-wait v1 collaboration tools—spawn, send input,
resume, and close—emit canonical `TurnItem::CollabAgentToolCall`
lifecycle instead of their legacy begin/end events directly.

App-server v2 consumes the canonical collab items directly, ignores the
mapped legacy events, and applies close-agent thread-watch cleanup from
the completed item.

## Why

These four tools share the same single-target lifecycle shape. Wait
stays separate because it carries multi-target status snapshots and has
its own status-shaping cleanup.

## What changed

- Add shared helpers for emitting canonical collab tool-call lifecycle.
- Migrate spawn, send input, resume, and close handlers.
- Move close-agent watcher cleanup onto canonical completed collab
items.
This commit is contained in:
Owen Lin
2026-07-07 13:35:57 -07:00
committed by GitHub
parent 777f5da55c
commit 058d97c5dc
8 changed files with 174 additions and 122 deletions
@@ -89,6 +89,7 @@ use codex_core::ThreadManager;
use codex_core::review_format::format_review_findings_block;
use codex_core::review_prompts;
use codex_protocol::ThreadId;
use codex_protocol::items::CollabAgentTool as CoreCollabAgentTool;
use codex_protocol::items::TurnItem as CoreTurnItem;
use codex_protocol::items::parse_hook_prompt_message;
use codex_protocol::models::AdditionalPermissionProfile as CoreAdditionalPermissionProfile;
@@ -831,15 +832,20 @@ pub(crate) async fn apply_bespoke_event_handling(
// Deprecated MCP tool-call events are still fanned out for legacy clients.
// App-server v2 receives the canonical TurnItem::McpToolCall lifecycle instead.
}
msg @ (EventMsg::CollabAgentSpawnBegin(_)
EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_)
| 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.
}
msg @ (EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::PlanDelta(_)
| EventMsg::ReasoningContentDelta(_)
@@ -857,23 +863,6 @@ pub(crate) async fn apply_bespoke_event_handling(
// rollout compatibility consumers. App-server v2 receives the canonical
// SubAgentActivity item lifecycle instead.
}
EventMsg::CollabCloseEnd(end_event) => {
if thread_manager
.get_thread(end_event.receiver_thread_id)
.await
.is_err()
{
thread_watch_manager
.remove_thread(&end_event.receiver_thread_id.to_string())
.await;
}
let notification = item_event_to_server_notification(
EventMsg::CollabCloseEnd(end_event),
&conversation_id.to_string(),
&event_turn_id,
);
outgoing.send_server_notification(notification).await;
}
EventMsg::ContextCompacted(..) => {
// Core still fans out this deprecated event for legacy clients;
// v2 clients receive the canonical ContextCompaction item instead.
@@ -1351,6 +1340,11 @@ async fn apply_canonical_item_completed_side_effects(
)
.await;
}
CoreTurnItem::CollabAgentToolCall(item) if item.tool == CoreCollabAgentTool::CloseAgent => {
for thread_id in &item.receiver_thread_ids {
remove_missing_thread_watch(thread_manager, thread_watch_manager, *thread_id).await;
}
}
_ => {}
}
}
@@ -8,8 +8,6 @@
use crate::agent::AgentStatus;
use crate::agent::exceeds_thread_spawn_depth_limit;
use crate::function_tool::FunctionCallError;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
@@ -20,17 +18,13 @@ use crate::tools::handlers::parse_arguments;
use crate::tools::registry::CoreToolRuntime;
use crate::tools::registry::ToolExecutor;
use codex_protocol::ThreadId;
use codex_protocol::items::CollabAgentTool;
use codex_protocol::items::CollabAgentToolCallItem;
use codex_protocol::items::CollabAgentToolCallStatus;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::CollabAgentInteractionBeginEvent;
use codex_protocol::protocol::CollabAgentInteractionEndEvent;
use codex_protocol::protocol::CollabAgentRef;
use codex_protocol::protocol::CollabAgentSpawnBeginEvent;
use codex_protocol::protocol::CollabAgentSpawnEndEvent;
use codex_protocol::protocol::CollabCloseBeginEvent;
use codex_protocol::protocol::CollabCloseEndEvent;
use codex_protocol::protocol::CollabResumeBeginEvent;
use codex_protocol::protocol::CollabResumeEndEvent;
use codex_protocol::protocol::CollabWaitingBeginEvent;
use codex_protocol::protocol::CollabWaitingEndEvent;
use codex_protocol::user_input::UserInput;
@@ -91,6 +85,17 @@ mod send_input;
mod spawn;
pub(crate) mod wait;
pub(crate) fn collab_tool_call_status(
status: &AgentStatus,
receiver_thread_id: Option<ThreadId>,
) -> CollabAgentToolCallStatus {
match status {
AgentStatus::Errored(_) | AgentStatus::NotFound => CollabAgentToolCallStatus::Failed,
_ if receiver_thread_id.is_some() => CollabAgentToolCallStatus::Completed,
_ => CollabAgentToolCallStatus::Failed,
}
}
#[cfg(test)]
#[path = "multi_agents_tests.rs"]
mod tests;
@@ -1,6 +1,5 @@
use super::*;
use crate::tools::handlers::multi_agents_spec::create_close_agent_tool_v1;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::error::CodexErr;
use codex_tools::ToolSpec;
@@ -44,15 +43,20 @@ async fn handle_close_agent(
let known_agent = receiver_agent.is_some();
let receiver_agent = receiver_agent.unwrap_or_default();
session
.send_event(
.emit_turn_item_started(
&turn,
CollabCloseBeginEvent {
call_id: call_id.clone(),
started_at_ms: now_unix_timestamp_ms(),
&TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id.clone(),
tool: CollabAgentTool::CloseAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: session.thread_id,
receiver_thread_id: agent_id,
}
.into(),
receiver_thread_ids: vec![agent_id],
receiver_agents: Vec::new(),
prompt: None,
model: None,
reasoning_effort: None,
agents_states: Default::default(),
}),
)
.await;
let status = match session
@@ -68,18 +72,24 @@ async fn handle_close_agent(
Err(err) => {
let status = session.services.agent_control.get_status(agent_id).await;
session
.send_event(
.emit_turn_item_completed(
&turn,
CollabCloseEndEvent {
call_id: call_id.clone(),
completed_at_ms: now_unix_timestamp_ms(),
TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id.clone(),
tool: CollabAgentTool::CloseAgent,
status: collab_tool_call_status(&status, Some(agent_id)),
sender_thread_id: session.thread_id(),
receiver_thread_id: agent_id,
receiver_agent_nickname: receiver_agent.agent_nickname.clone(),
receiver_agent_role: receiver_agent.agent_role.clone(),
status,
}
.into(),
receiver_thread_ids: vec![agent_id],
receiver_agents: vec![CollabAgentRef {
thread_id: agent_id,
agent_nickname: receiver_agent.agent_nickname.clone(),
agent_role: receiver_agent.agent_role.clone(),
}],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: [(agent_id, status)].into_iter().collect(),
}),
)
.await;
return Err(collab_agent_error(agent_id, err));
@@ -90,18 +100,24 @@ async fn handle_close_agent(
.map_err(|err| collab_agent_error(agent_id, err))
.map(|_| ());
session
.send_event(
.emit_turn_item_completed(
&turn,
CollabCloseEndEvent {
call_id,
completed_at_ms: now_unix_timestamp_ms(),
TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id,
tool: CollabAgentTool::CloseAgent,
status: collab_tool_call_status(&status, Some(agent_id)),
sender_thread_id: session.thread_id,
receiver_thread_id: agent_id,
receiver_agent_nickname: receiver_agent.agent_nickname,
receiver_agent_role: receiver_agent.agent_role,
status: status.clone(),
}
.into(),
receiver_thread_ids: vec![agent_id],
receiver_agents: vec![CollabAgentRef {
thread_id: agent_id,
agent_nickname: receiver_agent.agent_nickname,
agent_role: receiver_agent.agent_role,
}],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: [(agent_id, status.clone())].into_iter().collect(),
}),
)
.await;
result?;
@@ -1,7 +1,8 @@
use super::*;
use crate::agent::next_thread_spawn_depth;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::tools::handlers::multi_agents_spec::create_resume_agent_tool;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_tools::ToolSpec;
use std::sync::Arc;
@@ -57,17 +58,24 @@ async fn handle_resume_agent(
}
session
.send_event(
.emit_turn_item_started(
&turn,
CollabResumeBeginEvent {
call_id: call_id.clone(),
started_at_ms: now_unix_timestamp_ms(),
&TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id.clone(),
tool: CollabAgentTool::ResumeAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: session.thread_id,
receiver_thread_id,
receiver_agent_nickname: receiver_agent.agent_nickname.clone(),
receiver_agent_role: receiver_agent.agent_role.clone(),
}
.into(),
receiver_thread_ids: vec![receiver_thread_id],
receiver_agents: vec![CollabAgentRef {
thread_id: receiver_thread_id,
agent_nickname: receiver_agent.agent_nickname.clone(),
agent_role: receiver_agent.agent_role.clone(),
}],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: Default::default(),
}),
)
.await;
@@ -113,18 +121,24 @@ async fn handle_resume_agent(
(receiver_agent, None)
};
session
.send_event(
.emit_turn_item_completed(
&turn,
CollabResumeEndEvent {
call_id,
completed_at_ms: now_unix_timestamp_ms(),
TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id,
tool: CollabAgentTool::ResumeAgent,
status: collab_tool_call_status(&status, Some(receiver_thread_id)),
sender_thread_id: session.thread_id(),
receiver_thread_id,
receiver_agent_nickname: receiver_agent.agent_nickname,
receiver_agent_role: receiver_agent.agent_role,
status: status.clone(),
}
.into(),
receiver_thread_ids: vec![receiver_thread_id],
receiver_agents: vec![CollabAgentRef {
thread_id: receiver_thread_id,
agent_nickname: receiver_agent.agent_nickname,
agent_role: receiver_agent.agent_role,
}],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: [(receiver_thread_id, status.clone())].into_iter().collect(),
}),
)
.await;
@@ -1,7 +1,6 @@
use super::*;
use crate::agent::control::render_input_preview;
use crate::tools::handlers::multi_agents_spec::create_send_input_tool_v1;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_tools::ToolSpec;
pub(crate) struct Handler;
@@ -67,16 +66,20 @@ impl Handler {
.map_err(|err| collab_agent_error(receiver_thread_id, err))?;
}
session
.send_event(
.emit_turn_item_started(
&turn,
CollabAgentInteractionBeginEvent {
call_id: call_id.clone(),
started_at_ms: now_unix_timestamp_ms(),
&TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id.clone(),
tool: CollabAgentTool::SendInput,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: session.thread_id,
receiver_thread_id,
prompt: prompt.clone(),
}
.into(),
receiver_thread_ids: vec![receiver_thread_id],
receiver_agents: Vec::new(),
prompt: Some(prompt.clone()),
model: None,
reasoning_effort: None,
agents_states: Default::default(),
}),
)
.await;
let agent_control = session.services.agent_control.clone();
@@ -90,19 +93,24 @@ impl Handler {
.get_status(receiver_thread_id)
.await;
session
.send_event(
.emit_turn_item_completed(
&turn,
CollabAgentInteractionEndEvent {
call_id,
completed_at_ms: now_unix_timestamp_ms(),
TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id,
tool: CollabAgentTool::SendInput,
status: collab_tool_call_status(&status, Some(receiver_thread_id)),
sender_thread_id: session.thread_id,
receiver_thread_id,
receiver_agent_nickname: receiver_agent.agent_nickname,
receiver_agent_role: receiver_agent.agent_role,
prompt,
status,
}
.into(),
receiver_thread_ids: vec![receiver_thread_id],
receiver_agents: vec![CollabAgentRef {
thread_id: receiver_thread_id,
agent_nickname: receiver_agent.agent_nickname,
agent_role: receiver_agent.agent_role,
}],
prompt: Some(prompt),
model: None,
reasoning_effort: None,
agents_states: [(receiver_thread_id, status)].into_iter().collect(),
}),
)
.await;
let submission_id = result?;
@@ -8,7 +8,6 @@ use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::apply_role_to_config;
use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions;
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v1;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_tools::ToolSpec;
#[derive(Default)]
@@ -71,17 +70,20 @@ async fn handle_spawn_agent(
));
}
session
.send_event(
.emit_turn_item_started(
&turn,
CollabAgentSpawnBeginEvent {
call_id: call_id.clone(),
started_at_ms: now_unix_timestamp_ms(),
&TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id.clone(),
tool: CollabAgentTool::SpawnAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: session.thread_id,
prompt: prompt.clone(),
model: args.model.clone().unwrap_or_default(),
reasoning_effort: args.reasoning_effort.clone().unwrap_or_default(),
}
.into(),
receiver_thread_ids: Vec::new(),
receiver_agents: Vec::new(),
prompt: Some(prompt.clone()),
model: Some(args.model.clone().unwrap_or_default()),
reasoning_effort: Some(args.reasoning_effort.clone().unwrap_or_default()),
agents_states: Default::default(),
}),
)
.await;
let mut config =
@@ -177,22 +179,33 @@ async fn handle_spawn_agent(
.and_then(|snapshot| snapshot.reasoning_effort.clone())
.unwrap_or(args.reasoning_effort.unwrap_or_default());
let nickname = new_agent_nickname.clone();
let receiver_thread_ids = new_thread_id.into_iter().collect();
let receiver_agents = new_thread_id
.map(|thread_id| CollabAgentRef {
thread_id,
agent_nickname: new_agent_nickname,
agent_role: new_agent_role,
})
.into_iter()
.collect();
let agents_states = new_thread_id
.map(|thread_id| [(thread_id, status.clone())].into_iter().collect())
.unwrap_or_default();
session
.send_event(
.emit_turn_item_completed(
&turn,
CollabAgentSpawnEndEvent {
call_id,
completed_at_ms: now_unix_timestamp_ms(),
TurnItem::CollabAgentToolCall(CollabAgentToolCallItem {
id: call_id,
tool: CollabAgentTool::SpawnAgent,
status: collab_tool_call_status(&status, new_thread_id),
sender_thread_id: session.thread_id,
new_thread_id,
new_agent_nickname,
new_agent_role,
prompt,
model: effective_model,
reasoning_effort: effective_reasoning_effort,
status,
}
.into(),
receiver_thread_ids,
receiver_agents,
prompt: Some(prompt),
model: Some(effective_model),
reasoning_effort: Some(effective_reasoning_effort),
agents_states,
}),
)
.await;
let new_thread_id = result?.thread_id;
@@ -1,5 +1,6 @@
use super::*;
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;
@@ -7,6 +7,7 @@ use crate::init_state_db;
use crate::local_agent_graph_store_from_state_db;
use crate::session::step_context::StepContext;
use crate::session::tests::make_session_and_context;
use crate::session::turn_context::TurnContext;
use crate::session_prefix::format_inter_agent_completion_message;
use crate::thread_manager::thread_store_from_config;
use crate::tools::context::ToolOutput;