mirror of
https://github.com/openai/codex.git
synced 2026-11-08 16:38:30 +00:00
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
This commit is contained in:
1 parent
1674b0a130
commit
9894a14c81
4 files changed
+182
-11
No files matched your search
@@ -41,6 +41,8 @@ use crate::reducer::valid_plugin_measurement_row;
|
|||||||
use codex_app_server_protocol::ClientRequest;
|
use codex_app_server_protocol::ClientRequest;
|
||||||
use codex_app_server_protocol::ClientResponsePayload;
|
use codex_app_server_protocol::ClientResponsePayload;
|
||||||
use codex_app_server_protocol::InitializeParams;
|
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::JSONRPCErrorError;
|
||||||
use codex_app_server_protocol::RequestId;
|
use codex_app_server_protocol::RequestId;
|
||||||
use codex_app_server_protocol::ServerNotification;
|
use codex_app_server_protocol::ServerNotification;
|
||||||
@@ -51,6 +53,9 @@ use codex_login::CodexAuth;
|
|||||||
use codex_login::default_client::create_client;
|
use codex_login::default_client::create_client;
|
||||||
use codex_plugin::PluginId;
|
use codex_plugin::PluginId;
|
||||||
use codex_plugin::PluginTelemetryMetadata;
|
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 codex_protocol::request_permissions::RequestPermissionsResponse;
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::path::PathBuf;
|
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) {
|
pub fn track_code_mode_tool_call(&self, input: crate::facts::CodeModeToolCallFact) {
|
||||||
self.record_fact(AnalyticsFact::Custom(
|
self.record_fact(AnalyticsFact::Custom(
|
||||||
CustomAnalyticsFact::CodeModeToolCall(input),
|
CustomAnalyticsFact::CodeModeToolCall(input),
|
||||||
|
|||||||
@@ -464,6 +464,7 @@ struct TurnToolCounts {
|
|||||||
mcp_tool_call: usize,
|
mcp_tool_call: usize,
|
||||||
dynamic_tool_call: usize,
|
dynamic_tool_call: usize,
|
||||||
subagent_tool_call: usize,
|
subagent_tool_call: usize,
|
||||||
|
subagent_tool_call_ids: HashSet<String>,
|
||||||
web_search: usize,
|
web_search: usize,
|
||||||
image_generation: usize,
|
image_generation: usize,
|
||||||
}
|
}
|
||||||
@@ -475,7 +476,11 @@ impl TurnToolCounts {
|
|||||||
ThreadItem::FileChange { .. } => self.file_change += 1,
|
ThreadItem::FileChange { .. } => self.file_change += 1,
|
||||||
ThreadItem::McpToolCall { .. } => self.mcp_tool_call += 1,
|
ThreadItem::McpToolCall { .. } => self.mcp_tool_call += 1,
|
||||||
ThreadItem::DynamicToolCall { .. } => self.dynamic_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;
|
self.subagent_tool_call += 1;
|
||||||
}
|
}
|
||||||
ThreadItem::WebSearch(_) => self.web_search += 1,
|
ThreadItem::WebSearch(_) => self.web_search += 1,
|
||||||
@@ -2672,7 +2677,7 @@ fn tool_item_event(input: ToolItemEventInput<'_>) -> Option<TrackEventRequest> {
|
|||||||
ToolItemOutcome {
|
ToolItemOutcome {
|
||||||
terminal_status,
|
terminal_status,
|
||||||
failure_kind,
|
failure_kind,
|
||||||
execution_duration_ms: None,
|
execution_duration_ms: observed_duration_ms(started_at_ms, completed_at_ms),
|
||||||
},
|
},
|
||||||
ToolItemContext {
|
ToolItemContext {
|
||||||
started_at_ms,
|
started_at_ms,
|
||||||
|
|||||||
@@ -3693,6 +3693,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> {
|
|||||||
const CHILD_PROMPT: &str = "child: do work";
|
const CHILD_PROMPT: &str = "child: do work";
|
||||||
const PARENT_PROMPT: &str = "spawn a child and continue";
|
const PARENT_PROMPT: &str = "spawn a child and continue";
|
||||||
const SPAWN_CALL_ID: &str = "spawn-call-direct-input-rejection";
|
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 =
|
const ERROR_MESSAGE: &str =
|
||||||
"direct app-server input is not allowed for multi-agent v2 sub-agents";
|
"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,
|
"message": CHILD_PROMPT,
|
||||||
"task_name": "worker",
|
"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(
|
let _parent_turn = responses::mount_sse_once_match(
|
||||||
&server,
|
&server,
|
||||||
|req: &wiremock::Request| body_contains(req, PARENT_PROMPT),
|
|req: &wiremock::Request| body_contains(req, PARENT_PROMPT),
|
||||||
responses::sse(vec![
|
responses::sse(vec![
|
||||||
responses::ev_response_created("resp-parent-1"),
|
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(
|
responses::ev_function_call_with_namespace(
|
||||||
SPAWN_CALL_ID,
|
SPAWN_CALL_ID,
|
||||||
MULTI_AGENT_V2_NAMESPACE,
|
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()?;
|
let codex_home = TempDir::new()?;
|
||||||
MockResponsesConfig::new(&server.uri())
|
MockResponsesConfig::new(&server.uri())
|
||||||
.enable_feature(Feature::MultiAgentV2)
|
.enable_feature(Feature::MultiAgentV2)
|
||||||
|
.with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri()))
|
||||||
.write(codex_home.path())?;
|
.write(codex_home.path())?;
|
||||||
write_models_cache(codex_home.path())?;
|
write_models_cache(codex_home.path())?;
|
||||||
|
mount_analytics_capture(&server, codex_home.path()).await?;
|
||||||
|
|
||||||
let mut mcp = TestAppServer::builder()
|
let mut mcp = TestAppServer::builder()
|
||||||
.with_codex_home(codex_home.path())
|
.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(|request_id| ClientRequest::TurnStart {
|
||||||
request_id,
|
request_id,
|
||||||
params: TurnStartParams {
|
params: TurnStartParams {
|
||||||
thread_id: thread.id,
|
thread_id: thread.id.clone(),
|
||||||
input: vec![V2UserInput::Text {
|
input: vec![V2UserInput::Text {
|
||||||
text: PARENT_PROMPT.to_string(),
|
text: PARENT_PROMPT.to_string(),
|
||||||
text_elements: Vec::new(),
|
text_elements: Vec::new(),
|
||||||
@@ -3752,6 +3766,10 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> {
|
|||||||
loop {
|
loop {
|
||||||
let completed: ItemCompletedNotification =
|
let completed: ItemCompletedNotification =
|
||||||
mcp.read_notification("item/completed").await?;
|
mcp.read_notification("item/completed").await?;
|
||||||
|
assert!(!matches!(
|
||||||
|
&completed.item,
|
||||||
|
ThreadItem::CollabAgentToolCall { .. }
|
||||||
|
));
|
||||||
if let ThreadItem::SubAgentActivity {
|
if let ThreadItem::SubAgentActivity {
|
||||||
id,
|
id,
|
||||||
kind: SubAgentActivityKind::Started,
|
kind: SubAgentActivityKind::Started,
|
||||||
@@ -3825,7 +3843,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> {
|
|||||||
|
|
||||||
let direct_steer_req = mcp
|
let direct_steer_req = mcp
|
||||||
.send_turn_steer_request(TurnSteerParams {
|
.send_turn_steer_request(TurnSteerParams {
|
||||||
thread_id: child_thread_id,
|
thread_id: child_thread_id.clone(),
|
||||||
client_user_message_id: None,
|
client_user_message_id: None,
|
||||||
input: vec![V2UserInput::Text {
|
input: vec![V2UserInput::Text {
|
||||||
text: "direct app-server steer".to_string(),
|
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.code, INVALID_REQUEST_ERROR_CODE);
|
||||||
assert_eq!(direct_steer_error.error.message, ERROR_MESSAGE);
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,11 +5,15 @@ use crate::agent::next_thread_spawn_depth;
|
|||||||
use crate::agent::role::DEFAULT_ROLE_NAME;
|
use crate::agent::role::DEFAULT_ROLE_NAME;
|
||||||
use crate::agent_communication::AgentCommunicationContext;
|
use crate::agent_communication::AgentCommunicationContext;
|
||||||
use crate::agent_communication::AgentCommunicationKind;
|
use crate::agent_communication::AgentCommunicationKind;
|
||||||
|
use crate::codex_thread::ThreadConfigSnapshot;
|
||||||
use crate::session::multi_agents::resolve_usage_hints;
|
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::SpawnAgentToolOptions;
|
||||||
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2;
|
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::tools::handlers::multi_agents_v2::message_tool::message_content;
|
||||||
|
use crate::turn_timing::now_unix_timestamp_ms;
|
||||||
use codex_protocol::AgentPath;
|
use codex_protocol::AgentPath;
|
||||||
|
use codex_protocol::ThreadId;
|
||||||
use codex_protocol::protocol::MultiAgentVersion;
|
use codex_protocol::protocol::MultiAgentVersion;
|
||||||
use codex_tools::ToolSpec;
|
use codex_tools::ToolSpec;
|
||||||
|
|
||||||
@@ -34,13 +38,66 @@ impl ToolExecutor<ToolInvocation> for Handler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> {
|
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(
|
async fn handle_spawn_agent(
|
||||||
invocation: ToolInvocation,
|
invocation: ToolInvocation,
|
||||||
) -> Result<SpawnAgentResult, FunctionCallError> {
|
) -> Result<
|
||||||
|
(
|
||||||
|
SpawnAgentResult,
|
||||||
|
ThreadId,
|
||||||
|
AgentStatus,
|
||||||
|
Option<ThreadConfigSnapshot>,
|
||||||
|
),
|
||||||
|
FunctionCallError,
|
||||||
|
> {
|
||||||
let ToolInvocation {
|
let ToolInvocation {
|
||||||
session,
|
session,
|
||||||
step_context,
|
step_context,
|
||||||
@@ -163,6 +220,7 @@ async fn handle_spawn_agent(
|
|||||||
.await
|
.await
|
||||||
.map_err(collab_spawn_error)?;
|
.map_err(collab_spawn_error)?;
|
||||||
let new_thread_id = spawned_agent.thread_id;
|
let new_thread_id = spawned_agent.thread_id;
|
||||||
|
let agent_status = spawned_agent.status;
|
||||||
let agent_snapshot = session
|
let agent_snapshot = session
|
||||||
.services
|
.services
|
||||||
.agent_control
|
.agent_control
|
||||||
@@ -192,14 +250,15 @@ async fn handle_spawn_agent(
|
|||||||
let task_name = String::from(new_agent_path);
|
let task_name = String::from(new_agent_path);
|
||||||
|
|
||||||
let hide_agent_metadata = turn.config.multi_agent_v2.hide_spawn_agent_metadata;
|
let hide_agent_metadata = turn.config.multi_agent_v2.hide_spawn_agent_metadata;
|
||||||
if hide_agent_metadata {
|
let output = if hide_agent_metadata {
|
||||||
Ok(SpawnAgentResult::HiddenMetadata { task_name })
|
SpawnAgentResult::HiddenMetadata { task_name }
|
||||||
} else {
|
} else {
|
||||||
Ok(SpawnAgentResult::WithNickname {
|
SpawnAgentResult::WithNickname {
|
||||||
task_name,
|
task_name,
|
||||||
nickname,
|
nickname,
|
||||||
})
|
}
|
||||||
}
|
};
|
||||||
|
Ok((output, new_thread_id, agent_status, agent_snapshot))
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CoreToolRuntime for Handler {
|
impl CoreToolRuntime for Handler {
|
||||||
|
|||||||
Reference in new issue
Block a user