use super::{
config::SubagentRunConfig,
dto::SubagentTask,
worker::{ChildAgentRun, subagent_prompt},
};
use crate::{
agent::AgentOutputSink,
cancellation::AgentCancellation,
output::{
ActivityEvent, ActivityId, ActivityKind, ActivityMetadata, ActivitySender, ActivityStatus,
OutputEvent, ToolStatus,
},
};
use std::{collections::BTreeSet, path::PathBuf, sync::Arc};
pub(super) fn subagent_task_activity_metadata(
id: &str,
task: &SubagentTask,
depth: usize,
run: Option<&ChildAgentRun>,
) -> ActivityMetadata {
let mut metadata = ActivityMetadata::new(format!("{} · depth {} · {}", id, depth, task.intent));
metadata.detail = Some(subagent_prompt(id, task, None));
metadata
.fields
.push(("depth".to_string(), depth.to_string()));
if let Some(identity) = &task.identity {
metadata
.fields
.push(("identity".to_string(), identity.clone()));
}
if let Some(agent) = &task.agent {
metadata.fields.push(("agent".to_string(), agent.clone()));
}
if let Some(run) = run {
metadata
.fields
.push(("provider".to_string(), run.agent.provider_id().to_string()));
metadata
.fields
.push(("model".to_string(), run.agent.model().to_string()));
metadata.fields.push((
"reasoning".to_string(),
run.agent.thinking_level().as_str().to_string(),
));
}
metadata
}
pub(super) fn emit_activity(config: &SubagentRunConfig, event: ActivityEvent) {
if let Some(sender) = &config.activity_sender {
sender(event);
}
}
fn scope_child_activity(mut event: ActivityEvent, parent: &ActivityId) -> ActivityEvent {
let scope = |id: &mut ActivityId| {
if id != parent && !id.is_path_descendant_of(parent) {
*id = parent.child(id.as_str());
}
};
match &mut event {
ActivityEvent::Started { id, parent_id, .. } => {
scope(id);
if let Some(parent_id) = parent_id {
scope(parent_id);
} else {
*parent_id = Some(parent.clone());
}
}
ActivityEvent::Delta { id, .. }
| ActivityEvent::UsageUpdate { id, .. }
| ActivityEvent::UsageSnapshot { id, .. }
| ActivityEvent::ToolStartedDetail { id, .. }
| ActivityEvent::ToolResultDetail { id, .. }
| ActivityEvent::FinalPreview { id, .. }
| ActivityEvent::Finished { id, .. } => scope(id),
ActivityEvent::FastObservation { .. } => {}
}
event
}
#[derive(Debug)]
pub(super) struct ActiveCompactionActivity {
id: ActivityId,
metadata: ActivityMetadata,
}
#[derive(Debug, Default)]
pub(super) struct UsageAccumulator {
pub(super) totals: crate::agent::provider_stream::RequestTokenTotals,
cumulative: crate::output::NormalizedUsageSnapshot,
active_sequence: Option<u64>,
active: Option<crate::output::NormalizedUsageSnapshot>,
finalized_sequence: Option<u64>,
has_usage: bool,
latest: Option<crate::output::NormalizedUsageSnapshot>,
latest_final: bool,
}
impl UsageAccumulator {
pub(super) fn request_started(&mut self, sequence: u64) -> bool {
if self
.finalized_sequence
.is_some_and(|value| sequence <= value)
|| self.active_sequence.is_some_and(|value| sequence <= value)
{
return false;
}
if let Some(active_sequence) = self.active_sequence {
self.finalize_active();
self.finalized_sequence = Some(
self.finalized_sequence
.map_or(active_sequence, |value| value.max(active_sequence)),
);
}
self.active_sequence = Some(sequence);
self.latest = None;
self.latest_final = false;
true
}
pub(super) fn observe(
&mut self,
sequence: u64,
usage: crate::output::NormalizedUsageSnapshot,
final_usage: bool,
) -> crate::output::NormalizedUsageAggregate {
if self
.finalized_sequence
.is_some_and(|value| sequence <= value)
|| self.active_sequence.is_some_and(|value| sequence < value)
{
return self.aggregate();
}
self.request_started(sequence);
self.active = Some(usage);
self.latest = Some(usage);
self.latest_final = final_usage;
if final_usage {
self.finalize_active();
self.finalized_sequence = Some(
self.finalized_sequence
.map_or(sequence, |value| value.max(sequence)),
);
}
self.aggregate()
}
pub(super) fn metrics(&self) -> Option<crate::output::NormalizedUsageAggregate> {
(self.has_usage || self.latest.is_some()).then_some(
crate::output::NormalizedUsageAggregate {
whole_run: self.current_whole_run(),
latest: self.latest,
latest_request_sequence: self.active_sequence,
latest_final: self.latest_final,
},
)
}
fn aggregate(&self) -> crate::output::NormalizedUsageAggregate {
self.metrics().unwrap_or_default()
}
fn current_whole_run(&self) -> crate::output::NormalizedUsageSnapshot {
let mut whole_run = self.cumulative;
if let Some(active) = self.active {
whole_run.effective_input = whole_run
.effective_input
.saturating_add(active.effective_input);
whole_run.output = whole_run.output.saturating_add(active.output);
whole_run.cache_read = whole_run.cache_read.saturating_add(active.cache_read);
whole_run.cache_known = if self.has_usage {
whole_run.cache_known && active.cache_known
} else {
active.cache_known
};
}
whole_run
}
fn finalize_active(&mut self) {
if let Some(usage) = self.active.take() {
self.cumulative.effective_input = self
.cumulative
.effective_input
.saturating_add(usage.effective_input);
self.cumulative.output = self.cumulative.output.saturating_add(usage.output);
self.cumulative.cache_read =
self.cumulative.cache_read.saturating_add(usage.cache_read);
self.cumulative.cache_known = if self.has_usage {
self.cumulative.cache_known && usage.cache_known
} else {
usage.cache_known
};
self.has_usage = true;
}
}
#[cfg(test)]
pub(super) fn snapshot(&self) -> crate::output::NormalizedUsageSnapshot {
self.current_whole_run()
}
}
pub(super) struct SubagentActivitySink {
pub(super) parent_id: ActivityId,
pub(super) activity_sender: Option<ActivitySender>,
pub(super) assistant_id: ActivityId,
pub(super) assistant_started: bool,
pub(super) reasoning_summary_group_count: usize,
pub(super) reasoning_summary_group_lines: Vec<String>,
pub(super) progress_reporter: Option<Arc<dyn Fn() + Send + Sync>>,
pub(super) cancellation: AgentCancellation,
pub(super) changed_files: BTreeSet<PathBuf>,
pub(super) compaction_sequence: usize,
pub(super) active_compaction: Option<ActiveCompactionActivity>,
pub(super) usage_by_request: UsageAccumulator,
}
impl SubagentActivitySink {
pub(super) fn has_filesystem_changes(&self) -> bool {
!self.changed_files.is_empty()
}
pub(super) fn usage_metrics(&self) -> Option<crate::output::NormalizedUsageAggregate> {
self.usage_by_request.metrics()
}
pub(super) fn finish_assistant(&mut self, status: ActivityStatus) {
if !self.assistant_started {
return;
}
self.assistant_started = false;
if self.cancellation.is_canceled()
&& matches!(status, ActivityStatus::Canceled | ActivityStatus::Failed)
{
if let Some(sender) = &self.activity_sender {
sender(ActivityEvent::Finished {
id: self.assistant_id.clone(),
status: ActivityStatus::Canceled,
metadata: None,
});
}
return;
}
let _ = self.emit_activity_event_raw(ActivityEvent::Finished {
id: self.assistant_id.clone(),
status,
metadata: None,
});
}
fn start_compaction(
&mut self,
current_tokens: Option<usize>,
max_tokens: Option<usize>,
threshold: Option<String>,
) -> anyhow::Result<()> {
if self.active_compaction.is_some() {
self.finish_compaction(ActivityStatus::Failed)?;
}
self.close_reasoning_summary_group();
self.compaction_sequence = self.compaction_sequence.saturating_add(1);
let id = self
.parent_id
.child(format!("compaction/{}", self.compaction_sequence));
let mut metadata = ActivityMetadata::new("automatic");
if let (Some(current_tokens), Some(max_tokens)) = (current_tokens, max_tokens) {
metadata.label = format!("automatic · {current_tokens}/{max_tokens}");
metadata
.fields
.push(("before_tokens".to_string(), current_tokens.to_string()));
metadata
.fields
.push(("max_tokens".to_string(), max_tokens.to_string()));
}
if let Some(threshold) = threshold {
metadata.fields.push(("threshold".to_string(), threshold));
}
self.emit_activity_event_raw(ActivityEvent::Started {
id: id.clone(),
parent_id: Some(self.parent_id.clone()),
kind: ActivityKind::Compaction,
status: ActivityStatus::Running,
metadata: metadata.clone(),
})?;
self.active_compaction = Some(ActiveCompactionActivity { id, metadata });
Ok(())
}
fn complete_compaction(
&mut self,
current_tokens: usize,
max_tokens: usize,
summary: String,
) -> anyhow::Result<()> {
if self.active_compaction.is_none() {
if self.cancellation.is_canceled() {
return Ok(());
}
self.start_compaction(None, None, None)?;
}
let Some(active) = self.active_compaction.as_ref() else {
return Ok(());
};
let mut active = ActiveCompactionActivity {
id: active.id.clone(),
metadata: active.metadata.clone(),
};
active.metadata.label = format!("automatic · {current_tokens}/{max_tokens}");
active
.metadata
.fields
.push(("after_tokens".to_string(), current_tokens.to_string()));
if !active
.metadata
.fields
.iter()
.any(|(name, _)| name == "max_tokens")
{
active
.metadata
.fields
.push(("max_tokens".to_string(), max_tokens.to_string()));
}
self.emit_activity_event_raw(ActivityEvent::FinalPreview {
id: active.id.clone(),
preview: summary,
metadata: Some(active.metadata.clone()),
status: Some(ActivityStatus::Success),
})?;
self.emit_activity_event_raw(ActivityEvent::Finished {
id: active.id.clone(),
status: ActivityStatus::Success,
metadata: Some(active.metadata.clone()),
})?;
self.active_compaction = None;
Ok(())
}
pub(super) fn finish_compaction(&mut self, status: ActivityStatus) -> anyhow::Result<()> {
let Some(active) = self.active_compaction.take() else {
return Ok(());
};
self.emit_activity_event_raw(ActivityEvent::Finished {
id: active.id,
status,
metadata: Some(active.metadata),
})
}
fn emit_reasoning_summary_lines(&mut self, text: &str) -> anyhow::Result<()> {
let lines = text
.lines()
.map(str::trim)
.filter(|line| !line.is_empty())
.map(str::to_string)
.collect::<Vec<_>>();
if lines.is_empty() {
return Ok(());
}
if self.reasoning_summary_group_lines.is_empty() {
self.reasoning_summary_group_count += 1;
}
self.reasoning_summary_group_lines.extend(lines);
let metadata = ActivityMetadata::reasoning(&self.reasoning_summary_group_lines);
self.emit_activity_event_raw(ActivityEvent::Started {
id: self
.parent_id
.child(format!("reasoning/{}", self.reasoning_summary_group_count)),
parent_id: Some(self.parent_id.clone()),
kind: ActivityKind::Assistant,
status: ActivityStatus::Success,
metadata,
})?;
Ok(())
}
fn close_reasoning_summary_group(&mut self) {
self.reasoning_summary_group_lines.clear();
}
pub(super) fn emit_activity_event_raw(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
let allowed_after_cancel = match &event {
ActivityEvent::Started {
kind: ActivityKind::Tool,
..
}
| ActivityEvent::ToolStartedDetail { .. }
| ActivityEvent::ToolResultDetail { .. }
| ActivityEvent::UsageSnapshot { .. }
| ActivityEvent::Finished { .. } => true,
ActivityEvent::FinalPreview {
id,
status: Some(ActivityStatus::Success),
..
} => self
.active_compaction
.as_ref()
.is_some_and(|active| active.id == *id),
_ => false,
};
if self.cancellation.is_canceled() && !allowed_after_cancel {
return Ok(());
}
if let Some(progress) = &self.progress_reporter {
progress();
}
if let Some(sender) = &self.activity_sender {
sender(scope_child_activity(event, &self.parent_id));
}
Ok(())
}
}
impl AgentOutputSink for SubagentActivitySink {
fn request_token_total(&mut self, sequence: u64, total: u64) -> anyhow::Result<()> {
self.usage_by_request.totals.observe(sequence, total);
Ok(())
}
fn assistant_delta(&mut self, text: &str) -> anyhow::Result<()> {
if self.cancellation.is_canceled() {
return Ok(());
}
self.close_reasoning_summary_group();
if !self.assistant_started {
self.assistant_started = true;
self.activity_event(ActivityEvent::Started {
id: self.assistant_id.clone(),
parent_id: Some(self.parent_id.clone()),
kind: ActivityKind::Assistant,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("assistant"),
})?;
}
self.activity_event(ActivityEvent::Delta {
id: self.assistant_id.clone(),
preview: text.to_string(),
})
}
fn output_event(&mut self, event: OutputEvent) -> anyhow::Result<()> {
if self.cancellation.is_canceled()
&& !matches!(
event,
OutputEvent::ToolStarted { .. }
| OutputEvent::ToolResult { .. }
| OutputEvent::UsageSnapshot { .. }
| OutputEvent::CompactionCompleted { .. }
| OutputEvent::CompactionFailed { .. }
)
{
return Ok(());
}
if let OutputEvent::UsageSnapshot {
usage,
request_sequence,
final_usage,
} = event
{
let cumulative = self
.usage_by_request
.observe(request_sequence, usage, final_usage);
return self.activity_event(ActivityEvent::UsageSnapshot {
id: self.parent_id.clone(),
usage: cumulative,
request_sequence,
final_usage,
});
}
match event {
OutputEvent::ToolStarted { call, label } => {
self.close_reasoning_summary_group();
let status = crate::output::pending_activity_status(&call);
self.activity_event(ActivityEvent::ToolStartedDetail {
id: ActivityId::new(call.id.clone()),
detail: crate::tool_display::tool_activity_detail_pending(&call, label, status),
})
}
OutputEvent::ToolResult {
call,
result,
summary,
changed_paths,
} => {
self.changed_files.extend(changed_paths);
let status = match summary.status {
ToolStatus::Running => ActivityStatus::Running,
ToolStatus::Writing => ActivityStatus::Writing,
ToolStatus::Success => ActivityStatus::Success,
ToolStatus::Failure => ActivityStatus::Failed,
};
self.activity_event(ActivityEvent::ToolResultDetail {
id: ActivityId::new(call.id.clone()),
detail: crate::tool_display::tool_activity_detail(
&call,
&result,
summary.label.clone(),
status,
),
})
}
OutputEvent::AssistantDelta { text } => self.assistant_delta(&text),
OutputEvent::ThinkingSummaryComplete { text } => {
self.emit_reasoning_summary_lines(&text)
}
OutputEvent::ThinkingSummaryCompleteIdentified { text, .. } => {
self.emit_reasoning_summary_lines(&text)
}
OutputEvent::ContextUsage {
current_tokens,
max_tokens,
reasoning_tokens,
source,
request_sequence,
} => {
let request_started = self.usage_by_request.request_started(request_sequence);
self.activity_event(ActivityEvent::UsageUpdate {
id: self.parent_id.clone(),
current_tokens,
max_tokens,
reasoning_tokens,
source,
request_sequence,
})?;
if request_started && let Some(usage) = self.usage_by_request.metrics() {
self.activity_event(ActivityEvent::UsageSnapshot {
id: self.parent_id.clone(),
usage,
request_sequence,
final_usage: false,
})?;
}
Ok(())
}
OutputEvent::FastObservation {
provider_id,
model,
requested_service_tier,
outcome,
request_sequence,
run_order,
} => self.activity_event(ActivityEvent::FastObservation {
provider_id,
model,
requested_service_tier,
outcome,
request_sequence,
run_order,
}),
OutputEvent::CompactionTriggered {
current_tokens,
max_tokens,
threshold,
} => self.start_compaction(Some(current_tokens), Some(max_tokens), Some(threshold)),
OutputEvent::CompactionStarted => {
if self.active_compaction.is_none() {
self.start_compaction(None, None, None)?;
}
Ok(())
}
OutputEvent::CompactionCompleted {
current_tokens,
max_tokens,
summary,
} => self.complete_compaction(current_tokens, max_tokens, summary),
OutputEvent::CompactionFailed { .. } => {
self.close_reasoning_summary_group();
Ok(())
}
OutputEvent::BashCommand { .. }
| OutputEvent::ThinkingSummaryDelta { .. }
| OutputEvent::Diagnostic { .. }
| OutputEvent::SessionHeader { .. }
| OutputEvent::UserPrompt { .. }
| OutputEvent::AutomaticUserPrompt { .. }
| OutputEvent::AssistantComplete { .. }
| OutputEvent::HookDiagnostic { .. }
| OutputEvent::ProviderContextInjection { .. }
| OutputEvent::SubdirInstructionInjection { .. }
| OutputEvent::UsageSnapshot { .. } => {
self.close_reasoning_summary_group();
Ok(())
}
}
}
fn activity_event(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
if !matches!(
event,
ActivityEvent::UsageUpdate { .. } | ActivityEvent::UsageSnapshot { .. }
) {
self.close_reasoning_summary_group();
}
self.emit_activity_event_raw(event)
}
fn activity_sender(&self) -> Option<ActivitySender> {
let sender = self.activity_sender.clone()?;
let parent = self.parent_id.clone();
Some(Arc::new(move |event| {
sender(scope_child_activity(event, &parent))
}))
}
fn current_parent_activity_id(&self) -> Option<ActivityId> {
Some(self.parent_id.clone())
}
fn tool_block(&mut self, _block: &str) -> anyhow::Result<()> {
Ok(())
}
}