use super::*;
use crate::output::tool_card::{TaskBatchTelemetry, TaskTelemetry};
use std::{
collections::HashMap,
sync::{Arc, Mutex},
};
pub(super) struct TaskTelemetryCapture {
pub(super) report: TaskBatchTelemetry,
task_ids: HashMap<String, usize>,
calls: HashMap<String, (usize, bool)>,
closed: bool,
}
impl TaskTelemetryCapture {
pub(super) fn install(
args: &SubagentsArgs,
config: &mut SubagentRunConfig,
) -> Arc<Mutex<Self>> {
let batch = config
.parent_activity_id
.clone()
.unwrap_or_else(|| ActivityId::new("subagents"));
let tasks: Vec<_> = args
.tasks
.iter()
.enumerate()
.map(|(index, task)| {
let mut telemetry = TaskTelemetry::new(index);
telemetry.scoped_id = Some(batch.child(format!("g{}", index + 1)).0);
telemetry.set_intent(&task.intent);
let mut metadata = ActivityMetadata::new("");
if let Some(profile) = &task.identity {
metadata.fields.push(("identity".into(), profile.clone()));
}
if let Some(agent) = &task.agent {
metadata.fields.push(("agent".into(), agent.clone()));
}
telemetry.metadata(&metadata);
telemetry.calls_unknown = Some(0);
telemetry.calls_failed = Some(0);
telemetry.calls_changed = Some(0);
telemetry.activity_omitted = Some(0);
telemetry
})
.collect();
let task_ids = tasks
.iter()
.enumerate()
.filter_map(|(i, t)| Some((t.scoped_id.clone()?, i)))
.collect();
let capture = Arc::new(Mutex::new(Self {
report: TaskBatchTelemetry {
schema_version: 1,
tasks,
},
task_ids,
calls: HashMap::new(),
closed: false,
}));
let prior = config.activity_sender.take();
let observer = capture.clone();
config.activity_sender = Some(Arc::new(move |event| {
let mut capture = observer.lock().unwrap_or_else(|e| e.into_inner());
capture.observe(&event);
let publish = !capture.closed
&& match &event {
ActivityEvent::Started {
id,
kind: ActivityKind::SubagentTask,
..
}
| ActivityEvent::Finished { id, .. } => {
capture.task_ids.contains_key(id.as_str())
}
_ => false,
};
if let Some(sender) = &prior {
sender(event);
if publish {
sender(ActivityEvent::ToolMetadata {
id: batch.clone(),
metadata: json!({"subagent_telemetry": capture.report}),
});
}
}
}));
capture
}
fn observe(&mut self, event: &ActivityEvent) {
if self.closed {
return;
}
match event {
ActivityEvent::Started {
id,
kind: ActivityKind::SubagentTask,
metadata,
..
} => {
if let Some(&index) = self.task_ids.get(id.as_str()) {
self.report.tasks[index].metadata(metadata);
self.report.tasks[index].outcome = Some("running".into());
}
}
ActivityEvent::Started {
id,
parent_id: Some(parent),
kind: ActivityKind::Tool,
..
} => {
if let Some(&index) = self.task_ids.get(parent.as_str()) {
if self.calls.contains_key(id.as_str()) {
return;
}
if self.calls.len() < 4096 {
self.calls.insert(id.0.clone(), (index, false));
let count = &mut self.report.tasks[index].calls_unknown;
*count = count.map(|n| n.saturating_add(1));
} else {
self.report.tasks[index].calls_unknown = None;
self.report.tasks[index].calls_failed = None;
self.report.tasks[index].calls_changed = None;
self.report.tasks[index].activity_omitted = None;
self.report.tasks[index].text_limited = true;
}
}
}
ActivityEvent::UsageSnapshot { id, usage, .. } => {
if let Some(&index) = self.task_ids.get(id.as_str()) {
self.report.tasks[index].usage = Some(*usage);
}
}
ActivityEvent::ToolResultDetail { id, detail } => {
if let Some((index, finished)) = self.calls.get_mut(id.as_str()) {
if *finished {
return;
}
*finished = true;
let task = &mut self.report.tasks[*index];
if matches!(
detail.status,
ActivityStatus::Success | ActivityStatus::Failed
) {
task.calls_unknown = task.calls_unknown.map(|n| n.saturating_sub(1));
}
task.calls_failed = task
.calls_failed
.map(|n| n + u64::from(detail.status == ActivityStatus::Failed));
let card = crate::output::tool_card::project_tool_card(
&crate::output::tool_card::ToolCardInput {
tool_name: &detail.tool_name,
detail: Some(detail),
status: Some(detail.status),
interrupted: false,
bash_command: None,
bash_mode_response: false,
preview: "",
body: "",
children: &[],
children_complete: false,
},
);
task.calls_changed = task.calls_changed.map(|n| {
n + u64::from(matches!(
card.changes,
crate::output::tool_card::ChangeState::Applied
| crate::output::tool_card::ChangeState::Partial
))
});
task.activity(&format!(
"{}: {}",
detail.tool_name,
match detail.status {
ActivityStatus::Success => "success",
ActivityStatus::Failed => "failure",
ActivityStatus::Canceled => "cancelled",
_ => "unknown",
}
));
}
}
ActivityEvent::Finished {
id,
status,
metadata,
} => {
if let Some(&index) = self.task_ids.get(id.as_str()) {
let task = &mut self.report.tasks[index];
task.outcome = Some(
match status {
ActivityStatus::Success => "success",
ActivityStatus::Failed => "failure",
ActivityStatus::Canceled => "cancelled",
_ => "unknown",
}
.into(),
);
if let Some(metadata) = metadata {
task.metadata(metadata);
}
}
}
_ => {}
}
}
pub(super) fn finish(&mut self, output: Option<&SubagentsOutput>) -> Value {
self.closed = true;
for task in &mut self.report.tasks {
if task
.outcome
.as_deref()
.is_none_or(|outcome| outcome == "running")
{
task.outcome = Some("unknown".into());
}
if let Some(result) = output.and_then(|output| output.results.get(task.index))
&& result.id == format!("g{}", task.index + 1)
{
task.usage = result.usage.or(task.usage);
if task.outcome.as_deref() != Some("unknown") {
task.changed_paths = Some(result.changed_files.len() as u64);
}
}
}
while serde_json::to_vec(&self.report)
.expect("task telemetry")
.len()
> 64 * 1024
{
let Some(task) = self
.report
.tasks
.iter_mut()
.rev()
.find(|task| !task.activity.is_empty())
else {
return Value::Null;
};
task.activity.pop();
task.activity_omitted = task.activity_omitted.map(|count| count.saturating_add(1));
task.text_limited = true;
}
serde_json::to_value(&self.report).expect("bounded task telemetry")
}
}