diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 831715523b2e..2319cdaccee0 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -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; + } + } _ => {} } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents.rs b/codex-rs/core/src/tools/handlers/multi_agents.rs index 8bc96d19c2a1..c48c8b0b58b3 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents.rs @@ -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, +) -> 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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs index d1017adb65c9..d905cf4b9426 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs @@ -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?; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs index fa464ddbbb26..8e97b2afbd8c 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs @@ -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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index 375bb50ac7cb..649eaf1ce0ef 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -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?; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 204385934e2b..e5923bab7958 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -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; 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 be22b369082f..8d0e8daf0af1 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/wait.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/wait.rs @@ -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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index f313fcf0d9a7..c369fccbc770 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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;