diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 2319cdaccee0..b6f47dfe820b 100644 --- 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), diff --git a/codex-rs/core/src/tools/handlers/multi_agents.rs b/codex-rs/core/src/tools/handlers/multi_agents.rs index c48c8b0b58b3..87afc2b7cffa 100644 --- 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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/wait.rs b/codex-rs/core/src/tools/handlers/multi_agents/wait.rs index 8d0e8daf0af1..75fe4f0f9fa7 100644 --- 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::>(); - 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) -> CollabAgentToolCallStatus { + if statuses + .values() + .any(|status| matches!(status, AgentStatus::Errored(_) | AgentStatus::NotFound)) + { + CollabAgentToolCallStatus::Failed + } else { + CollabAgentToolCallStatus::Completed + } +} + +fn wait_receiver_agents( + statuses: &HashMap, + receiver_agents: &[CollabAgentRef], +) -> Vec { + 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::>(); + 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 { .. }) diff --git a/codex-rs/core/src/tools/handlers/multi_agents_common.rs b/codex-rs/core/src/tools/handlers/multi_agents_common.rs index 12ec89d651c9..287e4b30d5d6 100644 --- 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, - receiver_agents: &[CollabAgentRef], -) -> Vec { - 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::>(); - 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" => { diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index 9473908a463f..d1446ec6218f 100644 --- 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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/wait.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/wait.rs index e97f47238b67..4e2eced26037 100644 --- 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;