From 9894a14c81e50bbd845a337e4f77293f1cbc2633 Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 20 Aug 2026 14:28:15 +0000 Subject: [PATCH] Track multi-agent v2 spawn calls in analytics (#39722) ## What changed - Emit started and completed collaboration tool events for multi-agent v2 `spawn_agent` calls, including failed invocations. - Record execution duration and successful child-agent configuration metadata without including the spawn prompt. - Deduplicate collaboration and subagent activity items by call ID when calculating per-turn subagent tool counts. ## Testing - Extend the app-server multi-agent v2 integration test to cover successful and failed spawn telemetry, duration, prompt omission, and turn counts. GitOrigin-RevId: c1c2fb9cc2a1a9b0f63ef68509ad36d11a92d1e5 --- codex-rs/analytics/src/client.rs | 32 ++++++++ codex-rs/analytics/src/reducer.rs | 9 ++- .../app-server/tests/suite/v2/turn_start.rs | 79 ++++++++++++++++++- .../tools/handlers/multi_agents_v2/spawn.rs | 73 +++++++++++++++-- 4 files changed, 182 insertions(+), 11 deletions(-) diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 00ec2d8d4c..42d8f0e7c3 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -41,6 +41,8 @@ use crate::reducer::valid_plugin_measurement_row; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::ItemCompletedNotification; +use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; @@ -51,6 +53,9 @@ use codex_login::CodexAuth; use codex_login::default_client::create_client; use codex_plugin::PluginId; use codex_plugin::PluginTelemetryMetadata; +use codex_protocol::items::CollabAgentToolCallItem; +use codex_protocol::items::CollabAgentToolCallStatus; +use codex_protocol::items::TurnItem; use codex_protocol::request_permissions::RequestPermissionsResponse; use std::collections::HashSet; use std::path::PathBuf; @@ -330,6 +335,33 @@ impl AnalyticsEventsClient { )); } + pub fn track_collab_tool_call( + &self, + turn_id: String, + mut item: CollabAgentToolCallItem, + started_at_ms: i64, + completed_at_ms: i64, + ) { + let thread_id = item.sender_thread_id.to_string(); + let completed_item = TurnItem::CollabAgentToolCall(item.clone()).into(); + item.status = CollabAgentToolCallStatus::InProgress; + + self.track_notification(&ServerNotification::ItemStarted(ItemStartedNotification { + item: TurnItem::CollabAgentToolCall(item).into(), + thread_id: thread_id.clone(), + turn_id: turn_id.clone(), + started_at_ms, + })); + self.track_notification(&ServerNotification::ItemCompleted( + ItemCompletedNotification { + item: completed_item, + thread_id, + turn_id, + completed_at_ms, + }, + )); + } + pub fn track_code_mode_tool_call(&self, input: crate::facts::CodeModeToolCallFact) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::CodeModeToolCall(input), diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 974658ec20..ee3248484a 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -464,6 +464,7 @@ struct TurnToolCounts { mcp_tool_call: usize, dynamic_tool_call: usize, subagent_tool_call: usize, + subagent_tool_call_ids: HashSet, web_search: usize, image_generation: usize, } @@ -475,7 +476,11 @@ impl TurnToolCounts { ThreadItem::FileChange { .. } => self.file_change += 1, ThreadItem::McpToolCall { .. } => self.mcp_tool_call += 1, ThreadItem::DynamicToolCall { .. } => self.dynamic_tool_call += 1, - ThreadItem::CollabAgentToolCall { .. } | ThreadItem::SubAgentActivity { .. } => { + ThreadItem::CollabAgentToolCall { id, .. } + | ThreadItem::SubAgentActivity { id, .. } => { + if !self.subagent_tool_call_ids.insert(id.clone()) { + return; + } self.subagent_tool_call += 1; } ThreadItem::WebSearch(_) => self.web_search += 1, @@ -2672,7 +2677,7 @@ fn tool_item_event(input: ToolItemEventInput<'_>) -> Option { ToolItemOutcome { terminal_status, failure_kind, - execution_duration_ms: None, + execution_duration_ms: observed_duration_ms(started_at_ms, completed_at_ms), }, ToolItemContext { started_at_ms, diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index c9feab115a..d46ab18128 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -3693,6 +3693,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { const CHILD_PROMPT: &str = "child: do work"; const PARENT_PROMPT: &str = "spawn a child and continue"; const SPAWN_CALL_ID: &str = "spawn-call-direct-input-rejection"; + const FAILED_SPAWN_CALL_ID: &str = "spawn-call-invalid-fork-turns"; const ERROR_MESSAGE: &str = "direct app-server input is not allowed for multi-agent v2 sub-agents"; @@ -3701,11 +3702,22 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { "message": CHILD_PROMPT, "task_name": "worker", }))?; + let failed_spawn_args = serde_json::to_string(&json!({ + "message": CHILD_PROMPT, + "task_name": "invalid-worker", + "fork_turns": "0", + }))?; let _parent_turn = responses::mount_sse_once_match( &server, |req: &wiremock::Request| body_contains(req, PARENT_PROMPT), responses::sse(vec![ responses::ev_response_created("resp-parent-1"), + responses::ev_function_call_with_namespace( + FAILED_SPAWN_CALL_ID, + MULTI_AGENT_V2_NAMESPACE, + "spawn_agent", + &failed_spawn_args, + ), responses::ev_function_call_with_namespace( SPAWN_CALL_ID, MULTI_AGENT_V2_NAMESPACE, @@ -3719,8 +3731,10 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) .enable_feature(Feature::MultiAgentV2) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) .write(codex_home.path())?; write_models_cache(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -3738,7 +3752,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { .request(|request_id| ClientRequest::TurnStart { request_id, params: TurnStartParams { - thread_id: thread.id, + thread_id: thread.id.clone(), input: vec![V2UserInput::Text { text: PARENT_PROMPT.to_string(), text_elements: Vec::new(), @@ -3752,6 +3766,10 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { loop { let completed: ItemCompletedNotification = mcp.read_notification("item/completed").await?; + assert!(!matches!( + &completed.item, + ThreadItem::CollabAgentToolCall { .. } + )); if let ThreadItem::SubAgentActivity { id, kind: SubAgentActivityKind::Started, @@ -3825,7 +3843,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { let direct_steer_req = mcp .send_turn_steer_request(TurnSteerParams { - thread_id: child_thread_id, + thread_id: child_thread_id.clone(), client_user_message_id: None, input: vec![V2UserInput::Text { text: "direct app-server steer".to_string(), @@ -3844,6 +3862,63 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> { assert_eq!(direct_steer_error.error.code, INVALID_REQUEST_ERROR_CODE); assert_eq!(direct_steer_error.error.message, ERROR_MESSAGE); + let event = wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_collab_agent_tool_call_event" + && event["event_params"]["item_id"] == SPAWN_CALL_ID + }) + .await?; + assert_eq!( + json!({ + "tool_name": event["event_params"]["tool_name"], + "sender_thread_id": event["event_params"]["sender_thread_id"], + "receiver_thread_ids": event["event_params"]["receiver_thread_ids"], + "agent_state_count": event["event_params"]["agent_state_count"], + "terminal_status": event["event_params"]["terminal_status"], + }), + json!({ + "tool_name": "spawn_agent", + "sender_thread_id": thread.id, + "receiver_thread_ids": [child_thread_id], + "agent_state_count": 1, + "terminal_status": "completed", + }) + ); + let duration_ms = event["event_params"]["duration_ms"] + .as_u64() + .context("successful spawn should report its execution duration")?; + assert_eq!(event["event_params"]["execution_duration_ms"], duration_ms); + assert!(!event.to_string().contains(CHILD_PROMPT)); + + let failed_event = wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_collab_agent_tool_call_event" + && event["event_params"]["item_id"] == FAILED_SPAWN_CALL_ID + }) + .await?; + assert_eq!( + json!({ + "tool_name": failed_event["event_params"]["tool_name"], + "receiver_thread_count": failed_event["event_params"]["receiver_thread_count"], + "receiver_thread_ids": failed_event["event_params"]["receiver_thread_ids"], + "terminal_status": failed_event["event_params"]["terminal_status"], + "failure_kind": failed_event["event_params"]["failure_kind"], + }), + json!({ + "tool_name": "spawn_agent", + "receiver_thread_count": 0, + "receiver_thread_ids": [], + "terminal_status": "failed", + "failure_kind": "tool_error", + }) + ); + assert!(!failed_event.to_string().contains(CHILD_PROMPT)); + + let turn_event = wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_turn_event" && event["event_params"]["thread_id"] == thread.id + }) + .await?; + assert_eq!(turn_event["event_params"]["subagent_tool_call_count"], 2); + assert_eq!(turn_event["event_params"]["total_tool_call_count"], 2); + Ok(()) } 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 898064d2e0..9d93d8d35f 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 @@ -5,11 +5,15 @@ use crate::agent::next_thread_spawn_depth; use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; +use crate::codex_thread::ThreadConfigSnapshot; use crate::session::multi_agents::resolve_usage_hints; +use crate::tools::handlers::multi_agents::collab_tool_call_status; 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_protocol::ThreadId; use codex_protocol::protocol::MultiAgentVersion; use codex_tools::ToolSpec; @@ -34,13 +38,66 @@ impl ToolExecutor for Handler { } fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> { - Box::pin(async move { handle_spawn_agent(invocation).await.map(boxed_tool_output) }) + Box::pin(async move { + let analytics = invocation.session.services.analytics_events_client.clone(); + let sender_thread_id = invocation.session.thread_id; + let turn_id = invocation.step_context.turn.sub_id.clone(); + let call_id = invocation.call_id.clone(); + let started_at_ms = now_unix_timestamp_ms(); + let result = handle_spawn_agent(invocation).await; + let completed_at_ms = now_unix_timestamp_ms(); + let (status, receiver_thread_ids, agents_states) = match &result { + Ok((_, thread_id, agent_status, _)) => ( + collab_tool_call_status(agent_status, Some(*thread_id)), + vec![*thread_id], + [(*thread_id, agent_status.clone())].into_iter().collect(), + ), + Err(_) => ( + CollabAgentToolCallStatus::Failed, + Vec::new(), + Default::default(), + ), + }; + let agent_snapshot = result + .as_ref() + .ok() + .and_then(|(_, _, _, snapshot)| snapshot.as_ref()); + + analytics.track_collab_tool_call( + turn_id, + CollabAgentToolCallItem { + id: call_id, + tool: CollabAgentTool::SpawnAgent, + status, + sender_thread_id, + receiver_thread_ids, + receiver_agents: Vec::new(), + prompt: None, + model: agent_snapshot.map(|snapshot| snapshot.model.clone()), + reasoning_effort: agent_snapshot + .and_then(|snapshot| snapshot.reasoning_effort.clone()), + agents_states, + }, + started_at_ms, + completed_at_ms, + ); + + result.map(|(output, _, _, _)| boxed_tool_output(output)) + }) } } async fn handle_spawn_agent( invocation: ToolInvocation, -) -> Result { +) -> Result< + ( + SpawnAgentResult, + ThreadId, + AgentStatus, + Option, + ), + FunctionCallError, +> { let ToolInvocation { session, step_context, @@ -163,6 +220,7 @@ async fn handle_spawn_agent( .await .map_err(collab_spawn_error)?; let new_thread_id = spawned_agent.thread_id; + let agent_status = spawned_agent.status; let agent_snapshot = session .services .agent_control @@ -192,14 +250,15 @@ async fn handle_spawn_agent( let task_name = String::from(new_agent_path); let hide_agent_metadata = turn.config.multi_agent_v2.hide_spawn_agent_metadata; - if hide_agent_metadata { - Ok(SpawnAgentResult::HiddenMetadata { task_name }) + let output = if hide_agent_metadata { + SpawnAgentResult::HiddenMetadata { task_name } } else { - Ok(SpawnAgentResult::WithNickname { + SpawnAgentResult::WithNickname { task_name, nickname, - }) - } + } + }; + Ok((output, new_thread_id, agent_status, agent_snapshot)) } impl CoreToolRuntime for Handler {