feat(core): emit canonical sub-agent activity items (#31299)

This PR depends on [#31296](https://github.com/openai/codex/pull/31296) for the canonical-to-legacy event mappings. ## Description This PR makes the MultiAgentV2 spawn, message/follow-up, and interrupt paths emit completed canonical `TurnItem::SubAgentActivity` items instead of `SubAgentActivityEvent` directly. App-server v2 now applies interrupted-agent thread-watch cleanup from the canonical completed item and ignores the mapped legacy activity event. ## Why Sub-agent activity is separate from the v1 collab tool begin/end lifecycle. Keeping it separate makes the v2 watcher side effect reviewable without mixing in the larger collab tool-call migration. ## What changed - Emit canonical sub-agent activity items from v2 spawn, message/follow-up, and interrupt paths. - Move missing-thread watcher cleanup onto canonical completed activity items. - Update focused app-server coverage to exercise canonical interrupted activity.
2026-10-18 17:18:52 +00:00 · 2026-07-07 12:57:42 -07:00
parent 1f710c973b
commit 1bd9d841ca
5 changed files with 103 additions and 76 deletions
@@ -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<ThreadManager>,
+    thread_watch_manager: &ThreadWatchManager,
    thread_state: &Arc<Mutex<ThreadState>>,
    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<ThreadManager>,
+    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,
@@ -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,
@@ -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,
@@ -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)))
 }
@@ -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",