use super::support::*;
#[test]
fn parallel_children_scope_reused_tool_ids_on_sink_and_worker_paths() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let barrier = Arc::new(std::sync::Barrier::new(2));
thread::scope(|threads| {
for child in ["batch/g1", "batch/g2"] {
let sender = sender.clone();
let barrier = barrier.clone();
threads.spawn(move || {
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new(child),
activity_sender: Some(sender),
assistant_id: ActivityId::new(format!("{child}/assistant")),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let worker_sender = sink.activity_sender().unwrap();
for call_id in ["reused-provider-id", "call_legacy_function_call"] {
let call = ToolCall {
id: call_id.into(),
name: "bash".into(),
arguments: json!({"command":"true"}),
};
sink.activity_event(ActivityEvent::Started {
id: ActivityId::new(call_id),
parent_id: Some(ActivityId::new(child)),
kind: ActivityKind::Tool,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("bash"),
})
.unwrap();
barrier.wait();
sink.output_event(OutputEvent::ToolStarted {
call: Box::new(call.clone()),
label: "bash".into(),
})
.unwrap();
worker_sender(ActivityEvent::Started {
id: ActivityId::new(format!("{call_id}/hook")),
parent_id: Some(ActivityId::new(call_id)),
kind: ActivityKind::Hook,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("hook"),
});
let nested = ActivityId::new(format!("{child}/{call_id}/nested/g1"));
worker_sender(ActivityEvent::Started {
id: nested.clone(),
parent_id: Some(ActivityId::new(format!("{child}/{call_id}"))),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("nested"),
});
worker_sender(ActivityEvent::Finished {
id: nested,
status: ActivityStatus::Success,
metadata: None,
});
let result = ToolResult {
tool_name: "bash".into(),
success: true,
content: "ok".into(),
metadata: json!({}),
display: Default::default(),
};
let summary = crate::output::tool_display_summary(&call, &result);
sink.output_event(OutputEvent::ToolResult {
changed_paths: Vec::new(),
call: Box::new(call),
result: Box::new(result),
summary: Box::new(summary),
})
.unwrap();
worker_sender(ActivityEvent::Finished {
id: ActivityId::new(call_id),
status: ActivityStatus::Success,
metadata: None,
});
}
});
}
});
let events = events.lock().unwrap();
let tool_ids = events
.iter()
.filter_map(|event| match event {
ActivityEvent::Started {
id,
kind: ActivityKind::Tool,
..
} => Some(id.as_str()),
_ => None,
})
.collect::<HashSet<_>>();
assert_eq!(tool_ids.len(), 4);
for child in ["batch/g1", "batch/g2"] {
for call_id in ["reused-provider-id", "call_legacy_function_call"] {
let expected = ActivityId::new(format!("{child}/{call_id}"));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Started { id, parent_id: Some(parent), kind: ActivityKind::Tool, .. } if id == &expected && parent.as_str() == child)));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::ToolStartedDetail { id, .. } if id == &expected)));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::ToolResultDetail { id, .. } if id == &expected)));
assert!(events.iter().any(
|event| matches!(event, ActivityEvent::Finished { id, .. } if id == &expected)
));
for suffix in ["hook", "nested/g1"] {
let descendant = expected.child(suffix);
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Started { id, parent_id: Some(parent), .. } if id == &descendant && parent == &expected)));
}
}
}
assert_eq!(events.len(), 28);
}
#[test]
fn subagent_activity_sink_emits_context_usage_for_task() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::ContextUsage {
current_tokens: 12_345,
max_tokens: 128_000,
reasoning_tokens: Some(99),
source: crate::output::ContextUsageSource::ProviderExact,
request_sequence: 7,
})
.unwrap();
sink.output_event(OutputEvent::UsageSnapshot {
usage: crate::output::NormalizedUsageSnapshot {
effective_input: 10,
output: 2,
cache_read: 1,
cache_known: true,
},
request_sequence: 7,
final_usage: false,
})
.unwrap();
sink.output_event(OutputEvent::ContextUsage {
current_tokens: 20_000,
max_tokens: 128_000,
reasoning_tokens: None,
source: crate::output::ContextUsageSource::TokenizerProjection,
request_sequence: 8,
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(events.len(), 4);
assert!(matches!(
&events[0],
ActivityEvent::UsageUpdate {
id,
current_tokens: 12_345,
max_tokens: 128_000,
reasoning_tokens: Some(99),
source: crate::output::ContextUsageSource::ProviderExact,
request_sequence: 7,
} if id.as_str() == "task-g1"
));
assert!(matches!(
&events[1],
ActivityEvent::UsageSnapshot {
usage,
request_sequence: 7,
final_usage: false,
..
} if usage.latest.is_some_and(|latest| latest.effective_input == 10)
));
assert!(matches!(
&events[2],
ActivityEvent::UsageUpdate {
current_tokens: 20_000,
request_sequence: 8,
..
}
));
assert!(matches!(
&events[3],
ActivityEvent::UsageSnapshot {
usage,
request_sequence: 8,
final_usage: false,
..
} if usage.latest.is_none() && usage.whole_run.effective_input == 10
));
}
#[test]
fn child_compaction_emits_one_summary_preview_before_success() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-compaction"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-compaction/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let summary = "authoritative child summary".to_string();
sink.output_event(OutputEvent::CompactionTriggered {
current_tokens: 90,
max_tokens: 100,
threshold: "90%".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::CompactionStarted).unwrap();
sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: 20,
max_tokens: 100,
summary: summary.clone(),
})
.unwrap();
let events = events.lock().unwrap();
let preview_indices = events
.iter()
.enumerate()
.filter_map(|(index, event)| {
matches!(event, ActivityEvent::FinalPreview { .. }).then_some(index)
})
.collect::<Vec<_>>();
assert_eq!(preview_indices.len(), 1);
let preview_index = preview_indices[0];
assert!(matches!(
&events[preview_index],
ActivityEvent::FinalPreview { id, preview, status: Some(ActivityStatus::Success), .. }
if id.as_str() == "task-compaction/compaction/1" && preview == &summary
));
assert!(matches!(
&events[preview_index + 1],
ActivityEvent::Finished { id, status: ActivityStatus::Success, .. }
if id.as_str() == "task-compaction/compaction/1"
));
}
#[test]
fn canceled_child_after_committed_compaction_still_finishes_compaction() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let (cancellation, handle) = AgentCancellation::default().child_token();
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-cancel-compaction"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-cancel-compaction/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation,
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::CompactionStarted).unwrap();
handle.cancel();
sink.output_event(OutputEvent::CompactionCompleted {
current_tokens: 12,
max_tokens: 100,
summary: "committed before cancellation".to_string(),
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(
events
.iter()
.filter(|event| matches!(event, ActivityEvent::FinalPreview { .. }))
.count(),
1
);
assert!(matches!(
&events[1],
ActivityEvent::FinalPreview { id, preview, .. }
if id.as_str() == "task-cancel-compaction/compaction/1"
&& preview == "committed before cancellation"
));
assert!(matches!(
&events[2],
ActivityEvent::Finished { id, status: ActivityStatus::Success, .. }
if id.as_str() == "task-cancel-compaction/compaction/1"
));
}
#[test]
fn subagent_activity_sink_forwards_fast_observation() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::FastObservation {
provider_id: "provider".to_string(),
model: "model".to_string(),
requested_service_tier: "priority".to_string(),
outcome: crate::fast::FastOutcome::Different("default".to_string()),
request_sequence: 3,
run_order: None,
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(events.len(), 1);
assert!(matches!(
&events[0],
ActivityEvent::FastObservation {
provider_id,
model,
requested_service_tier,
outcome: crate::fast::FastOutcome::Different(returned),
request_sequence: 3,
run_order: None,
} if provider_id == "provider"
&& model == "model"
&& requested_service_tier == "priority"
&& returned == "default"
));
}
#[test]
fn subagent_activity_sink_retains_terminal_and_usage_events_after_cancel() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let (cancellation, handle) = AgentCancellation::default().child_token();
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-cancel"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-cancel/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation,
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.assistant_delta("partial").unwrap();
handle.cancel();
sink.assistant_delta("late").unwrap();
sink.finish_assistant(ActivityStatus::Failed);
sink.finish_assistant(ActivityStatus::Failed);
sink.activity_event(ActivityEvent::Delta {
id: ActivityId::new("tool-cancel"),
preview: "late tool output".to_string(),
})
.unwrap();
sink.activity_event(ActivityEvent::Finished {
id: ActivityId::new("tool-cancel"),
status: ActivityStatus::Canceled,
metadata: None,
})
.unwrap();
sink.output_event(OutputEvent::UsageSnapshot {
usage: crate::output::NormalizedUsageSnapshot {
effective_input: 10,
output: 2,
cache_read: 1,
cache_known: true,
},
request_sequence: 1,
final_usage: false,
})
.unwrap();
assert_eq!(
sink.usage_metrics()
.unwrap()
.latest
.unwrap()
.effective_input,
10
);
let events = events.lock().unwrap();
assert_eq!(events.len(), 5);
assert!(matches!(
&events[0],
ActivityEvent::Started {
status: ActivityStatus::Running,
..
}
));
assert!(matches!(
&events[1],
ActivityEvent::Delta { preview, .. } if preview == "partial"
));
assert!(matches!(
&events[2],
ActivityEvent::Finished {
status: ActivityStatus::Canceled,
metadata: None,
..
}
));
assert!(matches!(
&events[3],
ActivityEvent::Finished {
id,
status: ActivityStatus::Canceled,
metadata: None,
} if id.as_str() == "task-cancel/tool-cancel"
));
assert!(matches!(
&events[4],
ActivityEvent::UsageSnapshot {
id,
usage,
request_sequence: 1,
final_usage: false,
} if id.as_str() == "task-cancel"
&& usage.latest.is_some_and(|latest| latest.effective_input == 10)
));
}
#[test]
fn display_safety_preserves_tool_start_pair_during_cancellation() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let (cancellation, handle) = AgentCancellation::default().child_token();
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-cancel-tool"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-cancel-tool/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation,
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let call = ToolCall {
id: "child-tool".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"printf retained","timeout":5}),
};
handle.cancel();
sink.activity_event(ActivityEvent::Started {
id: ActivityId::new("child-tool"),
parent_id: Some(ActivityId::new("task-cancel-tool")),
kind: ActivityKind::Tool,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("bash printf retained"),
})
.unwrap();
sink.output_event(OutputEvent::ToolStarted {
call: Box::new(call),
label: "bash printf retained".to_string(),
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(events.len(), 2);
assert!(matches!(
&events[0],
ActivityEvent::Started {
id,
parent_id: Some(parent_id),
kind: ActivityKind::Tool,
..
} if id.as_str() == "task-cancel-tool/child-tool" && parent_id.as_str() == "task-cancel-tool"
));
assert!(matches!(
&events[1],
ActivityEvent::ToolStartedDetail { id, detail }
if id.as_str() == "task-cancel-tool/child-tool"
&& detail.tool_name.as_ref() == "bash"
&& detail.params == json!({"command":"printf retained","timeout":5})
));
}
#[test]
fn round_four_reconciles_nested_tool_result_after_cancellation() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| captured.lock().unwrap().push(event));
let (cancellation, handle) = AgentCancellation::default().child_token();
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-cancel-result"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-cancel-result/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation,
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let call = ToolCall {
id: "child-result".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"printf done"}),
};
sink.activity_event(ActivityEvent::Started {
id: ActivityId::new("child-result"),
parent_id: Some(ActivityId::new("task-cancel-result")),
kind: ActivityKind::Tool,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("bash printf done"),
})
.unwrap();
handle.cancel();
sink.output_event(OutputEvent::ToolResult {
changed_paths: Vec::new(),
call: Box::new(call),
result: Box::new(ToolResult {
tool_name: "bash".to_string(),
success: true,
content: "done".to_string(),
metadata: json!({"exit_code":0}),
display: crate::tools::ToolResultDisplay::default(),
}),
summary: Box::new(crate::output::ToolDisplaySummary {
tool_name: "bash".to_string(),
label: "bash printf done".to_string(),
status: crate::output::ToolStatus::Success,
unicode_mark: "✓",
ascii_mark: "OK",
metadata: Vec::new(),
}),
})
.unwrap();
sink.activity_event(ActivityEvent::Finished {
id: ActivityId::new("child-result"),
status: ActivityStatus::Success,
metadata: None,
})
.unwrap();
let events = events.lock().unwrap();
assert!(matches!(
&events[1],
ActivityEvent::ToolResultDetail { id, detail }
if id.as_str() == "task-cancel-result/child-result"
&& detail.status == ActivityStatus::Success
&& detail.output.as_ref() == "done"
));
assert!(matches!(
&events[2],
ActivityEvent::Finished { id, status: ActivityStatus::Success, .. }
if id.as_str() == "task-cancel-result/child-result"
));
}
#[test]
fn subagent_activity_sink_groups_consecutive_reasoning_summary_lines() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::ThinkingSummaryComplete {
text: "checked inputs\nverified output".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::ContextUsage {
current_tokens: 12,
max_tokens: 100,
reasoning_tokens: None,
source: crate::output::ContextUsageSource::ProviderExact,
request_sequence: 1,
})
.unwrap();
sink.output_event(OutputEvent::ThinkingSummaryComplete {
text: "planned fix".to_string(),
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(events.len(), 3);
assert!(matches!(events[1], ActivityEvent::UsageUpdate { .. }));
assert!(matches!(
&events[0],
ActivityEvent::Started {
id,
parent_id: Some(parent_id),
kind: ActivityKind::Assistant,
status: ActivityStatus::Success,
metadata,
} if id.as_str() == "task-g1/reasoning/1"
&& parent_id.as_str() == "task-g1"
&& metadata.label == "reasoning summaries ×2"
&& metadata.detail.as_deref() == Some("1. checked inputs\n2. verified output")
));
assert!(matches!(
&events[2],
ActivityEvent::Started {
id,
parent_id: Some(parent_id),
kind: ActivityKind::Assistant,
status: ActivityStatus::Success,
metadata,
} if id.as_str() == "task-g1/reasoning/1"
&& parent_id.as_str() == "task-g1"
&& metadata.label == "reasoning summaries ×3"
&& metadata.detail.as_deref() == Some("1. checked inputs\n2. verified output\n3. planned fix")
));
}
#[test]
fn subagent_activity_sink_splits_reasoning_groups_on_activity_boundaries() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::ThinkingSummaryComplete {
text: "first".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::ToolStarted {
call: Box::new(ToolCall {
id: "child_tool".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"true","timeout":5}),
}),
label: "bash true".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::ThinkingSummaryComplete {
text: "second\nthird".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::AssistantDelta {
text: "answer".to_string(),
})
.unwrap();
sink.output_event(OutputEvent::ThinkingSummaryComplete {
text: "fourth".to_string(),
})
.unwrap();
let events = events.lock().unwrap();
let reasoning = events
.iter()
.filter_map(|event| match event {
ActivityEvent::Started { id, metadata, .. } if id.as_str().contains("/reasoning/") => {
Some((
id.as_str().to_string(),
metadata.label.clone(),
metadata.detail.clone(),
))
}
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(
reasoning,
vec![
(
"task-g1/reasoning/1".to_string(),
"reasoning • first".to_string(),
Some("first".to_string()),
),
(
"task-g1/reasoning/2".to_string(),
"reasoning summaries ×2".to_string(),
Some("1. second\n2. third".to_string()),
),
(
"task-g1/reasoning/3".to_string(),
"reasoning • fourth".to_string(),
Some("fourth".to_string()),
),
]
);
}
#[test]
fn subagent_activity_sink_tool_result_final_preview() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let call = ToolCall {
id: "child_tool".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"false","timeout":5}),
};
let result = ToolResult {
tool_name: "bash".to_string(),
success: false,
content: "stderr: boom".to_string(),
metadata: json!({"exit_code":1}),
display: crate::tools::ToolResultDisplay::default(),
};
let summary = crate::output::tool_display_summary(&call, &result);
sink.output_event(OutputEvent::ToolResult {
changed_paths: Vec::new(),
call: Box::new(call),
result: Box::new(result),
summary: Box::new(summary),
})
.unwrap();
let events = events.lock().unwrap();
assert_eq!(events.len(), 1);
assert!(matches!(
&events[0],
ActivityEvent::ToolResultDetail { id, detail }
if id.as_str() == "task-g1/child_tool"
&& detail.status == ActivityStatus::Failed
&& detail.output.as_ref() == "stderr: boom"
));
assert!(!matches!(events[0], ActivityEvent::Delta { .. }));
}
#[test]
fn subagent_activity_sink_edit_tool_result_final_preview_includes_diff() {
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
let sender: ActivitySender = Arc::new(move |event| {
captured.lock().unwrap().push(event);
});
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-g1"),
activity_sender: Some(sender),
assistant_id: ActivityId::new("task-g1/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation: AgentCancellation::default(),
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
let call = ToolCall {
id: "child_edit".to_string(),
name: "hash_edit".to_string(),
arguments: json!({"input":"[src/lib.rs#ABCD]\nSWAP 1.=1:\n+new line"}),
};
let result = ToolResult {
tool_name: "hash_edit".to_string(),
success: true,
content: "applied 1 edits".to_string(),
metadata: json!({"path":"src/lib.rs","edits":1}),
display: crate::tools::ToolResultDisplay {
edit_diff: Some(
"diff --git a/src/lib.rs b/src/lib.rs\n--- a/src/lib.rs\n+++ b/src/lib.rs\n@@ -1 +1 @@\n-old line\n+new line\n"
.to_string(),
),
},
};
let summary = crate::output::tool_display_summary(&call, &result);
sink.output_event(OutputEvent::ToolResult {
changed_paths: vec![PathBuf::from("src/lib.rs")],
call: Box::new(call),
result: Box::new(result),
summary: Box::new(summary),
})
.unwrap();
assert!(sink.has_filesystem_changes());
let events = events.lock().unwrap();
assert_eq!(events.len(), 1);
assert!(matches!(
&events[0],
ActivityEvent::ToolResultDetail { id, detail }
if id.as_str() == "task-g1/child_edit"
&& detail.status == ActivityStatus::Success
&& detail.output.as_ref() == "applied 1 edits"
&& detail.applied_diff.as_deref().is_some_and(|diff| diff.contains("diff --git"))
));
}
#[test]
fn review_fix_subagent_activity_sink_tracks_partial_hash_edit_changes() {
let (cancellation, cancel_handle) = AgentCancellation::default().child_token();
cancel_handle.cancel();
let mut sink = SubagentActivitySink {
parent_id: ActivityId::new("task-partial-edit"),
activity_sender: None,
assistant_id: ActivityId::new("task-partial-edit/assistant"),
assistant_started: false,
reasoning_summary_group_count: 0,
reasoning_summary_group_lines: Vec::new(),
progress_reporter: None,
cancellation,
changed_files: BTreeSet::new(),
compaction_sequence: 0,
usage_by_request: UsageAccumulator::default(),
active_compaction: None,
};
sink.output_event(OutputEvent::ToolResult {
changed_paths: vec![
PathBuf::from("changed.txt"),
PathBuf::from("deleted.txt"),
PathBuf::from("source.txt"),
PathBuf::from("moved.txt"),
PathBuf::from("changed.txt"),
],
call: Box::new(ToolCall {
id: "partial-edit".to_string(),
name: "hash_edit".to_string(),
arguments: json!({}),
}),
result: Box::new(ToolResult {
tool_name: "hash_edit".to_string(),
success: false,
content: "partial commit".to_string(),
metadata: json!({
"outcome": "partial",
"files": [
{"path":"forged.txt","operation":"update","status":"committed"},
{"path":"untouched.txt","operation":"update","status":"failed"}
]
}),
display: crate::tools::ToolResultDisplay::default(),
}),
summary: Box::new(crate::output::ToolDisplaySummary {
tool_name: "hash_edit".to_string(),
label: "partial hash edit".to_string(),
status: crate::output::ToolStatus::Failure,
unicode_mark: "x",
ascii_mark: "ERR",
metadata: Vec::new(),
}),
})
.unwrap();
assert!(sink.has_filesystem_changes());
assert_eq!(
sink.changed_files,
BTreeSet::from([
PathBuf::from("changed.txt"),
PathBuf::from("deleted.txt"),
PathBuf::from("source.txt"),
PathBuf::from("moved.txt"),
])
);
}
#[test]
fn failed_subagent_emits_one_failed_task_finish() {
let temp = tempfile::TempDir::new().unwrap();
let provider = Arc::new(CountingProvider::new("never"));
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(provider.clone(), temp.path());
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let output = run_subagents(
SubagentsArgs {
concurrency: Some(1),
tasks: vec![SubagentTask {
intent: "generic".into(),
agent: None,
identity: Some("missing".into()),
context: None,
cwd: None,
}],
},
cfg,
)
.unwrap();
assert_eq!(output.summary.failed, 1);
assert!(provider.requests.lock().unwrap().is_empty());
let events = events.lock().unwrap();
assert_eq!(
events
.iter()
.filter(|event| matches!(event, ActivityEvent::Finished { id, status: ActivityStatus::Failed, .. } if id.as_str() == "subagents/g1"))
.count(),
1
);
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Started { id, status: ActivityStatus::Running, .. } if id.as_str() == "subagents/g1")));
}
#[test]
fn subagent_task_activity_metadata_includes_prompt_detail() {
let task = SubagentTask {
intent: "inspect parent detail payload".into(),
agent: Some("issue-43-plan".into()),
identity: Some("frontend-dev".into()),
context: Some("first context line\nsecond context line".into()),
cwd: None,
};
let metadata = subagent_task_activity_metadata("g1", &task, 2, None);
let expected_prompt = subagent_prompt("g1", &task, None);
assert_eq!(
metadata.label,
"g1 · depth 2 · inspect parent detail payload"
);
assert_eq!(
metadata.fields,
vec![
("depth".to_string(), "2".to_string()),
("identity".to_string(), "frontend-dev".to_string()),
("agent".to_string(), "issue-43-plan".to_string()),
]
);
assert_eq!(metadata.detail.as_deref(), Some(expected_prompt.as_str()));
assert!(expected_prompt.contains("Subagent g1 task intent:\ninspect parent detail payload"));
assert!(expected_prompt.contains("Agent label/persona: issue-43-plan"));
assert!(expected_prompt.contains("Task context:\nfirst context line\nsecond context line"));
}
#[test]
fn subagent_task_activity_metadata_includes_identity_and_agent_when_present() {
let temp = tempfile::TempDir::new().unwrap();
let provider = Arc::new(CountingProvider::new("never"));
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(provider, temp.path());
cfg.profiles.insert(
"frontend-dev".to_string(),
profile("frontend-dev", "frontend prompt"),
);
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
run_subagents(
SubagentsArgs {
concurrency: Some(1),
tasks: vec![SubagentTask {
intent: "one".into(),
agent: Some("reviewer".into()),
identity: Some("frontend-dev".into()),
context: None,
cwd: None,
}],
},
cfg,
)
.unwrap();
let events = events.lock().unwrap();
let task_metadata = events
.iter()
.rev()
.find_map(|event| match event {
ActivityEvent::Started {
kind: ActivityKind::SubagentTask,
metadata,
..
} => Some(metadata),
_ => None,
})
.expect("subagent task start event");
assert_eq!(task_metadata.label, "g1 · depth 1 · one");
assert_eq!(
task_metadata.fields,
vec![
("depth".to_string(), "1".to_string()),
("identity".to_string(), "frontend-dev".to_string()),
("agent".to_string(), "reviewer".to_string()),
("provider".to_string(), "openai-codex".to_string()),
("model".to_string(), "model".to_string()),
("reasoning".to_string(), "default".to_string()),
]
);
}
#[test]
fn task_activity_finisher_suppresses_started_after_terminal_finish() {
let temp = tempfile::TempDir::new().unwrap();
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(Arc::new(CountingProvider::new("never")), temp.path());
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let finisher = TaskActivityFinisher::new();
let id = ActivityId::new("subagents/g1");
finisher.finish(&cfg, id.clone(), ActivityStatus::Failed);
finisher.try_start(&cfg, || ActivityEvent::Started {
id: id.clone(),
parent_id: Some(ActivityId::new("subagents")),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("late start"),
});
finisher.finish(&cfg, id, ActivityStatus::Success);
let events = events.lock().unwrap();
assert_eq!(events.len(), 1, "{events:?}");
assert!(matches!(
&events[0],
ActivityEvent::Finished {
id,
status: ActivityStatus::Failed,
..
} if id.as_str() == "subagents/g1"
));
}
#[test]
fn task_activity_finisher_enriches_only_before_finished() {
let temp = tempfile::TempDir::new().unwrap();
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(Arc::new(CountingProvider::new("never")), temp.path());
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let finisher = TaskActivityFinisher::new();
let id = ActivityId::new("subagents/g1");
finisher.try_start(&cfg, || ActivityEvent::Started {
id: id.clone(),
parent_id: Some(ActivityId::new("subagents")),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("start"),
});
finisher.try_enrich(&cfg, || ActivityEvent::Started {
id: id.clone(),
parent_id: Some(ActivityId::new("subagents")),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("enriched start"),
});
finisher.finish(&cfg, id.clone(), ActivityStatus::Success);
finisher.try_enrich(&cfg, || ActivityEvent::Started {
id: id.clone(),
parent_id: Some(ActivityId::new("subagents")),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Running,
metadata: ActivityMetadata::new("late enrichment"),
});
finisher.finish(&cfg, id, ActivityStatus::Failed);
let events = events.lock().unwrap();
assert_eq!(events.len(), 3, "{events:?}");
assert!(matches!(
&events[0],
ActivityEvent::Started {
id,
status: ActivityStatus::Running,
..
} if id.as_str() == "subagents/g1"
));
assert!(matches!(
&events[1],
ActivityEvent::Started { metadata, .. } if metadata.label == "enriched start"
));
assert!(matches!(
&events[2],
ActivityEvent::Finished {
id,
status: ActivityStatus::Success,
..
} if id.as_str() == "subagents/g1"
));
}
#[test]
fn prefinished_subagent_does_not_emit_late_started_event() {
let temp = tempfile::TempDir::new().unwrap();
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(Arc::new(CountingProvider::new("never")), temp.path());
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let batch_id = ActivityId::new("subagents");
let task_activity_id = ActivityId::new("subagents/g1");
let finisher = TaskActivityFinisher::new();
finisher.finish(&cfg, task_activity_id, ActivityStatus::Failed);
let result = run_one_subagent(SubagentRunInput {
id: "g1".to_string(),
task: SubagentTask {
intent: "resolve missing identity after stall".to_string(),
agent: None,
identity: Some("missing-profile".to_string()),
context: None,
cwd: None,
},
cwd: temp.path().to_path_buf(),
config: &cfg,
cancellation: AgentCancellation::default(),
batch_id: &batch_id,
reporter: None,
finisher,
});
assert_eq!(result.status, SubagentStatus::Failed);
let events = events.lock().unwrap();
assert_eq!(events.len(), 1, "{events:?}");
assert!(matches!(
&events[0],
ActivityEvent::Finished {
id,
status: ActivityStatus::Failed,
..
} if id.as_str() == "subagents/g1"
));
}
#[test]
fn child_agent_resolution_failure_emits_started_before_finished_without_model_fields() {
let temp = tempfile::TempDir::new().unwrap();
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(Arc::new(CountingProvider::new("never")), temp.path());
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let output = run_subagents(
SubagentsArgs {
concurrency: Some(1),
tasks: vec![SubagentTask {
intent: "resolve missing identity".into(),
agent: None,
identity: Some("missing-profile".into()),
context: None,
cwd: None,
}],
},
cfg,
)
.unwrap();
assert_eq!(output.summary.failed, 1);
let events = events.lock().unwrap();
let task_events = events
.iter()
.filter(|event| match event {
ActivityEvent::Started { id, .. } | ActivityEvent::Finished { id, .. } => {
id.as_str() == "subagents/g1"
}
_ => false,
})
.collect::<Vec<_>>();
assert_eq!(task_events.len(), 2, "{task_events:?}");
match task_events[0] {
ActivityEvent::Started { metadata, .. } => {
assert_eq!(
metadata.fields,
vec![
("depth".to_string(), "1".to_string()),
("identity".to_string(), "missing-profile".to_string()),
]
);
}
event => panic!("expected task Started event, got {event:?}"),
}
assert!(matches!(
task_events[1],
ActivityEvent::Finished {
status: ActivityStatus::Failed,
..
}
));
}
#[test]
fn child_tasks_attach_directly_to_parent_activity_without_synthetic_batch() {
let temp = tempfile::TempDir::new().unwrap();
let provider = Arc::new(CountingProvider::new("never"));
let events: Arc<Mutex<Vec<ActivityEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
let mut cfg = config(provider, temp.path());
cfg.parent_activity_id = Some(ActivityId::new("tool-parent"));
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
run_subagents(
SubagentsArgs {
concurrency: Some(1),
tasks: vec![SubagentTask {
intent: "one".into(),
agent: None,
identity: None,
context: None,
cwd: None,
}],
},
cfg,
)
.unwrap();
let events = events.lock().unwrap();
assert!(!events.iter().any(|event| matches!(
event,
ActivityEvent::Started {
kind: ActivityKind::SubagentBatch,
..
}
)));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Started { id, parent_id: Some(parent), kind: ActivityKind::SubagentTask, .. } if id.as_str() == "tool-parent/g1" && parent.as_str() == "tool-parent")));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Finished { id, status: ActivityStatus::Success, .. } if id.as_str() == "tool-parent/g1")));
assert!(events.iter().any(|event| matches!(event, ActivityEvent::Finished { id, status: ActivityStatus::Success, .. } if id.as_str() == "tool-parent/g1/assistant")));
}
#[test]
fn usage_accumulator_replaces_partials_and_finalizes_each_sequence_once() {
let mut accumulator = UsageAccumulator::default();
let usage = |input, output| crate::output::NormalizedUsageSnapshot {
effective_input: input,
output,
cache_read: 0,
cache_known: true,
};
assert_eq!(
accumulator
.observe(1, usage(10, 1), false)
.whole_run
.effective_input,
10
);
assert_eq!(
accumulator.observe(1, usage(20, 2), true),
crate::output::NormalizedUsageAggregate {
whole_run: usage(20, 2),
latest: Some(usage(20, 2)),
latest_request_sequence: Some(1),
latest_final: true,
}
);
assert_eq!(
accumulator.observe(1, usage(99, 9), true),
crate::output::NormalizedUsageAggregate {
whole_run: usage(20, 2),
latest: Some(usage(20, 2)),
latest_request_sequence: Some(1),
latest_final: true,
}
);
assert_eq!(
accumulator.observe(2, usage(30, 3), false),
crate::output::NormalizedUsageAggregate {
whole_run: usage(50, 5),
latest: Some(usage(30, 3)),
latest_request_sequence: Some(2),
latest_final: false,
}
);
assert_eq!(
accumulator.observe(1, usage(500, 50), true),
crate::output::NormalizedUsageAggregate {
whole_run: usage(50, 5),
latest: Some(usage(30, 3)),
latest_request_sequence: Some(2),
latest_final: false,
}
);
assert_eq!(
accumulator.observe(2, usage(40, 4), true),
crate::output::NormalizedUsageAggregate {
whole_run: usage(60, 6),
latest: Some(usage(40, 4)),
latest_request_sequence: Some(2),
latest_final: true,
}
);
assert_eq!(accumulator.snapshot().output, 6);
}
#[test]
fn usage_accumulator_retains_unfinalized_request_for_terminal_metrics() {
let mut accumulator = UsageAccumulator::default();
let usage = crate::output::NormalizedUsageSnapshot {
effective_input: 24,
output: 6,
cache_read: 12,
cache_known: true,
};
let expected = crate::output::NormalizedUsageAggregate {
whole_run: usage,
latest: Some(usage),
latest_request_sequence: Some(7),
latest_final: false,
};
assert_eq!(accumulator.observe(7, usage, false), expected);
assert_eq!(accumulator.metrics(), Some(expected));
assert_eq!(accumulator.snapshot(), usage);
}
#[test]
fn usage_accumulator_clears_latest_only_for_new_requests() {
let mut accumulator = UsageAccumulator::default();
let mut fresh = UsageAccumulator::default();
assert!(fresh.request_started(99));
assert_eq!(fresh.metrics(), None);
let first = crate::output::NormalizedUsageSnapshot {
effective_input: 10,
output: 1,
cache_read: 2,
cache_known: true,
};
let second = crate::output::NormalizedUsageSnapshot {
effective_input: 20,
output: 3,
cache_read: 4,
cache_known: true,
};
assert_eq!(
accumulator.observe(1, first, true),
crate::output::NormalizedUsageAggregate {
whole_run: first,
latest: Some(first),
latest_request_sequence: Some(1),
latest_final: true,
}
);
assert!(accumulator.request_started(2));
assert_eq!(
accumulator.metrics(),
Some(crate::output::NormalizedUsageAggregate {
whole_run: first,
latest: None,
latest_request_sequence: Some(2),
latest_final: false,
})
);
assert!(!accumulator.request_started(2));
assert!(!accumulator.request_started(1));
assert_eq!(
accumulator.observe(2, second, false),
crate::output::NormalizedUsageAggregate {
whole_run: crate::output::NormalizedUsageSnapshot {
effective_input: 30,
output: 4,
cache_read: 6,
cache_known: true,
},
latest: Some(second),
latest_request_sequence: Some(2),
latest_final: false,
}
);
assert!(accumulator.request_started(3));
assert_eq!(
accumulator.metrics(),
Some(crate::output::NormalizedUsageAggregate {
whole_run: crate::output::NormalizedUsageSnapshot {
effective_input: 30,
output: 4,
cache_read: 6,
cache_known: true,
},
latest: None,
latest_request_sequence: Some(3),
latest_final: false,
})
);
}