diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 863aeac79bf8..831715523b2e 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -852,23 +852,10 @@ pub(crate) async fn apply_bespoke_event_handling( ); outgoing.send_server_notification(notification).await; } - EventMsg::SubAgentActivity(activity) => { - if activity.kind == SubAgentActivityKind::Interrupted - && thread_manager - .get_thread(activity.agent_thread_id) - .await - .is_err() - { - thread_watch_manager - .remove_thread(&activity.agent_thread_id.to_string()) - .await; - } - let notification = item_event_to_server_notification( - EventMsg::SubAgentActivity(activity), - &conversation_id.to_string(), - &event_turn_id, - ); - 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::CollabCloseEnd(end_event) => { if thread_manager @@ -1033,7 +1020,13 @@ pub(crate) async fn apply_bespoke_event_handling( } } EventMsg::ItemCompleted(event) => { - apply_canonical_item_completed_side_effects(&thread_state, &event.item).await; + apply_canonical_item_completed_side_effects( + &thread_manager, + &thread_watch_manager, + &thread_state, + &event.item, + ) + .await; let notification = item_event_to_server_notification( EventMsg::ItemCompleted(event), &conversation_id.to_string(), @@ -1334,16 +1327,43 @@ async fn emit_turn_completed_with_status( } async fn apply_canonical_item_completed_side_effects( + thread_manager: &Arc, + thread_watch_manager: &ThreadWatchManager, thread_state: &Arc>, item: &CoreTurnItem, ) { - if let CoreTurnItem::CommandExecution(item) = item { - thread_state - .lock() - .await - .turn_summary - .command_execution_started - .remove(&item.id); + match item { + CoreTurnItem::CommandExecution(item) => { + thread_state + .lock() + .await + .turn_summary + .command_execution_started + .remove(&item.id); + } + CoreTurnItem::SubAgentActivity(activity) + if activity.kind == SubAgentActivityKind::Interrupted => + { + remove_missing_thread_watch( + thread_manager, + thread_watch_manager, + activity.agent_thread_id, + ) + .await; + } + _ => {} + } +} + +async fn remove_missing_thread_watch( + thread_manager: &Arc, + thread_watch_manager: &ThreadWatchManager, + thread_id: ThreadId, +) { + if thread_manager.get_thread(thread_id).await.is_err() { + thread_watch_manager + .remove_thread(&thread_id.to_string()) + .await; } } @@ -2175,6 +2195,7 @@ mod tests { use codex_protocol::items::DynamicToolCallItem; use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus; use codex_protocol::items::HookPromptFragment; + use codex_protocol::items::SubAgentActivityItem; use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::items::build_hook_prompt_message; use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions; @@ -2192,12 +2213,12 @@ mod tests { use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; + use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionSource; - use codex_protocol::protocol::SubAgentActivityEvent; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::UserMessageEvent; @@ -3420,13 +3441,17 @@ mod tests { apply_bespoke_event_handling( Event { id: "turn-1".to_string(), - msg: EventMsg::SubAgentActivity(SubAgentActivityEvent { - event_id: "activity-1".to_string(), - occurred_at_ms: 42, - agent_thread_id: child_thread_id, - agent_path: AgentPath::try_from("/root/worker") - .expect("agent path should parse"), - kind: SubAgentActivityKind::Interrupted, + msg: EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: conversation_id, + turn_id: "turn-1".to_string(), + item: CoreTurnItem::SubAgentActivity(SubAgentActivityItem { + id: "activity-1".to_string(), + kind: SubAgentActivityKind::Interrupted, + agent_thread_id: child_thread_id, + agent_path: AgentPath::try_from("/root/worker") + .expect("agent path should parse"), + }), + completed_at_ms: 42, }), }, conversation_id, 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 8e7f73a7622f..9473908a463f 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::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::SubAgentActivityEvent; use codex_protocol::protocol::SubAgentActivityKind; use codex_tools::ToolName; use serde::Deserialize; @@ -39,6 +40,16 @@ mod send_message; mod spawn; pub(crate) mod wait; +pub(crate) async fn emit_sub_agent_activity( + session: &crate::session::session::Session, + turn: &crate::session::turn_context::TurnContext, + item: SubAgentActivityItem, +) { + session + .emit_turn_item_completed(turn, TurnItem::SubAgentActivity(item)) + .await; +} + pub(super) fn communication_from_tool_message( author: AgentPath, recipient: AgentPath, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/interrupt_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/interrupt_agent.rs index aeed055b4eff..00eb149ca166 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/interrupt_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/interrupt_agent.rs @@ -1,6 +1,5 @@ use super::*; use crate::tools::handlers::multi_agents_spec::create_interrupt_agent_tool_v2; -use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::error::CodexErr; use codex_tools::ToolSpec; @@ -71,19 +70,17 @@ async fn handle_interrupt_agent( Err(err) => Err(collab_agent_error(agent_id, err)), }; result?; - session - .send_event( - &turn, - SubAgentActivityEvent { - event_id: call_id, - occurred_at_ms: now_unix_timestamp_ms(), - agent_thread_id: agent_id, - agent_path: receiver_agent_path, - kind: SubAgentActivityKind::Interrupted, - } - .into(), - ) - .await; + emit_sub_agent_activity( + &session, + &turn, + SubAgentActivityItem { + id: call_id, + agent_thread_id: agent_id, + agent_path: receiver_agent_path, + kind: SubAgentActivityKind::Interrupted, + }, + ) + .await; Ok(InterruptAgentResult { previous_status: status, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 9501b4355ac9..62defe65c411 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -7,7 +7,6 @@ use super::*; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; use crate::tools::context::FunctionToolOutput; -use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::protocol::InterAgentCommunication; #[derive(Clone, Copy, PartialEq, Eq)] @@ -115,19 +114,17 @@ pub(crate) async fn handle_message_string_tool( .await .map_err(|err| collab_agent_error(receiver_thread_id, err)); result?; - session - .send_event( - &turn, - SubAgentActivityEvent { - event_id: call_id, - occurred_at_ms: now_unix_timestamp_ms(), - agent_thread_id: receiver_thread_id, - agent_path: receiver_agent_path, - kind: SubAgentActivityKind::Interacted, - } - .into(), - ) - .await; + emit_sub_agent_activity( + &session, + &turn, + SubAgentActivityItem { + id: call_id, + agent_thread_id: receiver_thread_id, + agent_path: receiver_agent_path, + kind: SubAgentActivityKind::Interacted, + }, + ) + .await; Ok(FunctionToolOutput::from_text(String::new(), Some(true))) } diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index 8eeb6ddd1b78..1c68849c5e06 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -9,7 +9,6 @@ use crate::agent_communication::AgentCommunicationKind; use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions; use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2; use crate::tools::handlers::multi_agents_v2::message_tool::message_content; -use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::AgentPath; use codex_tools::ToolSpec; @@ -140,19 +139,17 @@ async fn handle_spawn_agent( .as_ref() .and_then(|snapshot| snapshot.session_source.get_nickname()) .or(spawned_agent.metadata.agent_nickname); - session - .send_event( - &turn, - SubAgentActivityEvent { - event_id: call_id, - occurred_at_ms: now_unix_timestamp_ms(), - agent_thread_id: new_thread_id, - agent_path: new_agent_path.clone(), - kind: SubAgentActivityKind::Started, - } - .into(), - ) - .await; + emit_sub_agent_activity( + &session, + &turn, + SubAgentActivityItem { + id: call_id, + agent_thread_id: new_thread_id, + agent_path: new_agent_path.clone(), + kind: SubAgentActivityKind::Started, + }, + ) + .await; let role_tag = role_name.unwrap_or(DEFAULT_ROLE_NAME); turn.session_telemetry.counter( "codex.multi_agent.spawn",