use super::{config::SubagentRunConfig, dto::SubagentTask, worker::subagent_prompt};
use crate::{
agent::AgentOutputSink,
agent::cancellation::AgentCancellation,
output::{
ActivityEvent, ActivityId, ActivityKind, ActivityMetadata, ActivitySender, ActivityStatus,
OutputEvent, ToolStatus,
},
tool_display::format_tool_activity_detail_block,
};
use std::sync::Arc;
pub(super) fn subagent_task_activity_metadata(
id: &str,
task: &SubagentTask,
depth: usize,
) -> 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()));
}
metadata
}
pub(super) fn emit_activity(config: &SubagentRunConfig, event: ActivityEvent) {
if let Some(sender) = &config.activity_sender {
sender(event);
}
}
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,
}
impl SubagentActivitySink {
pub(super) fn finish_assistant(&mut self, status: ActivityStatus) {
if self.assistant_started {
let _ = self.activity_event(ActivityEvent::Finished {
id: self.assistant_id.clone(),
status,
metadata: None,
});
}
}
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 line_count = self.reasoning_summary_group_lines.len();
let mut metadata = ActivityMetadata::new(format!("reasoning summaries ×{line_count}"));
metadata.detail = Some(numbered_reasoning_lines(
&self.reasoning_summary_group_lines,
));
self.emit_activity_event_raw(ActivityEvent::Started {
id: ActivityId::new(format!(
"{}/reasoning/{}",
self.parent_id.as_str(),
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<()> {
if self.cancellation.is_canceled() {
return Ok(());
}
if let Some(progress) = &self.progress_reporter {
progress();
}
if let Some(sender) = &self.activity_sender {
sender(event);
}
Ok(())
}
}
impl AgentOutputSink for SubagentActivitySink {
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() {
return Ok(());
}
match event {
OutputEvent::ToolStarted { .. } => {
self.close_reasoning_summary_group();
Ok(())
}
OutputEvent::ToolResult {
call,
result,
summary,
} => {
self.close_reasoning_summary_group();
let preview = format_tool_activity_detail_block(&call, &result);
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::FinalPreview {
id: ActivityId::new(call.id.clone()),
preview,
metadata: Some(ActivityMetadata {
label: summary.label.clone(),
detail: Some(final_preview_detail(&summary.tool_name)),
fields: summary.metadata.clone(),
}),
status: Some(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,
} => self.activity_event(ActivityEvent::UsageUpdate {
id: self.parent_id.clone(),
current_tokens,
max_tokens,
reasoning_tokens,
source,
request_sequence,
}),
OutputEvent::ThinkingSummaryDelta { .. } => Ok(()),
OutputEvent::SessionHeader { .. }
| OutputEvent::UserPrompt { .. }
| OutputEvent::AutomaticUserPrompt { .. }
| OutputEvent::CompactionTriggered { .. }
| OutputEvent::CompactionStarted
| OutputEvent::CompactionCompleted { .. }
| OutputEvent::BashCommand { .. }
| OutputEvent::Diagnostic { .. }
| OutputEvent::HookDiagnostic { .. }
| OutputEvent::ProviderContextInjection { .. }
| OutputEvent::SubdirInstructionInjection { .. }
| OutputEvent::AssistantComplete { .. } => {
self.close_reasoning_summary_group();
Ok(())
}
}
}
fn activity_event(&mut self, event: ActivityEvent) -> anyhow::Result<()> {
if !matches!(event, ActivityEvent::UsageUpdate { .. }) {
self.close_reasoning_summary_group();
}
self.emit_activity_event_raw(event)
}
fn activity_sender(&self) -> Option<ActivitySender> {
self.activity_sender.clone()
}
fn current_parent_activity_id(&self) -> Option<ActivityId> {
Some(self.parent_id.clone())
}
fn tool_block(&mut self, _block: &str) -> anyhow::Result<()> {
Ok(())
}
}
pub(super) fn numbered_reasoning_lines(lines: &[String]) -> String {
lines
.iter()
.enumerate()
.map(|(index, line)| format!("{}. {line}", index + 1))
.collect::<Vec<_>>()
.join("\n")
}
pub(super) fn final_preview_detail(tool_name: &str) -> String {
match tool_name {
"bash" | "shell" => "bash preview: final output only".to_string(),
_ => "final tool output preview".to_string(),
}
}