use super::{RunFailure, sink::Events};
use crate::{
local_agent::tool_cards::{
self, CARDS_PER_RUN, CORE_BYTES, OPTIONAL_BYTES, OPTIONAL_SNAPSHOTS, RETAINED_BYTES,
},
output::{ActivityEvent, ActivityKind, ActivityStatus, ToolActivityDetail},
};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
struct Entry {
id: String,
kind: &'static str,
name: String,
parent: Option<String>,
task: bool,
task_total: Option<usize>,
finished: bool,
result_outcome: Option<String>,
card: Value,
last: Option<[u8; 32]>,
terminal_sent: bool,
awaiting_detail: bool,
protection_tool_name: Option<&'static str>,
bash_notice_state: tool_cards::BashNoticeState,
}
#[derive(Default)]
pub(super) struct Cards {
entries: Vec<Entry>,
indices: HashMap<String, usize>,
excluded: HashSet<String>,
optional_bytes: usize,
optional_snapshots: usize,
retained_bytes: usize,
closed: bool,
}
impl Cards {
pub(super) fn project(&mut self, events: &Events, event: ActivityEvent) -> anyhow::Result<()> {
if self.closed {
return Ok(());
}
let (index, authoritative) = match event {
ActivityEvent::Started {
id,
parent_id,
kind,
status,
metadata,
..
} => {
if self.indices.contains_key(id.as_str()) || self.excluded.contains(id.as_str()) {
return Ok(());
}
if id.as_str().len() > 1024
|| parent_id
.as_ref()
.is_some_and(|id| id.as_str().len() > 1024)
{
return Err(RunFailure("content_limit").into());
}
let kind_name = match kind {
ActivityKind::Tool => "tool",
ActivityKind::SubagentBatch | ActivityKind::SubagentTask => "subagent",
ActivityKind::Compaction => "compaction",
_ => {
if self.excluded.len() == CARDS_PER_RUN {
return Err(RunFailure("content_limit").into());
}
self.excluded.insert(id.0);
let related = self.refresh_hierarchy(None);
return self.emit_related(events, self.entries.len(), related);
}
};
if self.entries.len() == CARDS_PER_RUN {
return Err(RunFailure("content_limit").into());
}
let index = self.entries.len();
let public_id = uuid::Uuid::new_v4().to_string();
self.indices.insert(id.0, index);
let mut card = tool_cards::initial_card(&public_id, kind_name == "subagent");
if kind_name == "subagent" {
let task = crate::output::tool_card::project_task(
&crate::output::tool_card::TaskFacts {
label: &metadata.label,
intent: None,
status: Some(status),
fields: &metadata.fields,
failure: None,
usage: None,
live_tail: &[],
},
);
card["title"] = json!(task.label);
tool_cards::bound_card(&mut card);
}
self.retained_bytes += tool_cards::bytes(&card);
self.entries.push(Entry {
id: public_id,
kind: kind_name,
name: super::sink::summary_text(&metadata.label),
parent: parent_id.map(|id| id.0),
task: kind == ActivityKind::SubagentTask,
task_total: None,
finished: false,
result_outcome: None,
card,
last: None,
terminal_sent: false,
awaiting_detail: false,
protection_tool_name: None,
bash_notice_state: tool_cards::BashNoticeState::default(),
});
self.base(events, index, "started")?;
let related = self.refresh_hierarchy(None);
self.emit(events, index, true)?;
self.emit_related(events, index, related)?;
return Ok(());
}
ActivityEvent::Finished { id, status, .. } => {
let Some(&index) = self.indices.get(id.as_str()) else {
return Ok(());
};
if self.entries[index].finished {
return Ok(());
}
let entry = &mut self.entries[index];
let previous = tool_cards::bytes(&entry.card);
entry.finished = true;
entry.awaiting_detail = entry.kind == "tool"
&& status == ActivityStatus::Failed
&& entry.result_outcome.is_none();
entry.card["state"] = json!("finished");
entry.card["outcome"] = json!(
entry
.result_outcome
.as_deref()
.or_else(|| tool_cards::outcome(status))
.unwrap_or("unknown")
);
self.retained_bytes =
self.retained_bytes - previous + tool_cards::bytes(&entry.card);
self.base(
events,
index,
if status == ActivityStatus::Success {
"completed"
} else {
"failed"
},
)?;
(index, true)
}
ActivityEvent::ToolStartedDetail { id, detail }
| ActivityEvent::ToolResultDetail { id, detail } => {
let Some(&index) = self.indices.get(id.as_str()) else {
return Ok(());
};
if (self.entries[index].finished || self.entries[index].result_outcome.is_some())
&& tool_cards::outcome(detail.status).is_none()
{
return Ok(());
}
let entry = &mut self.entries[index];
let prior_bytes = tool_cards::bytes(&entry.card);
let previous_changes = entry.card["changes"].clone();
let previous_protection = entry.card["metadata"]["public_protection"].clone();
let previous_execution = tool_cards::execution_core_facts(&entry.card);
if entry.awaiting_detail {
entry.card["outcome"] = Value::Null;
}
entry.protection_tool_name =
crate::tools::ToolCapability::from_dispatch_name(&detail.tool_name)
.map(crate::tools::ToolCapability::canonical_name)
.or_else(|| {
(detail.tool_name.as_ref() == crate::code_mode::TOOL_NAME)
.then_some(crate::code_mode::TOOL_NAME)
});
if entry.protection_tool_name == Some("subagents") {
entry.task_total = detail.params["tasks"]
.as_array()
.filter(|tasks| tasks.len() <= crate::subagents::MAX_SUBAGENT_TASKS)
.map(Vec::len);
}
tool_cards::project_detail(
&mut entry.card,
&detail,
!entry.terminal_sent,
&mut entry.bash_notice_state,
);
if !entry.finished {
if let Some(outcome) = entry.card["outcome"].as_str() {
entry.result_outcome = Some(outcome.to_owned());
}
entry.card["outcome"] = Value::Null;
}
entry.awaiting_detail = false;
let authoritative = entry.card["changes"] != previous_changes
|| entry.card["metadata"]["public_protection"]["decision"]
!= previous_protection["decision"]
|| tool_cards::execution_core_facts(&entry.card) != previous_execution;
self.retained_bytes =
self.retained_bytes - prior_bytes + tool_cards::bytes(&entry.card);
(index, authoritative)
}
ActivityEvent::ToolMetadata { id, metadata } => {
let Some(&index) = self.indices.get(id.as_str()) else {
return Ok(());
};
let entry = &mut self.entries[index];
let detail = ToolActivityDetail {
tool_name: entry.protection_tool_name.unwrap_or("tool").into(),
label: "Tool".into(),
params: json!({}),
metadata,
status: ActivityStatus::Running,
output: "".into(),
applied_diff: None,
};
let mut selected = tool_cards::initial_card(&entry.id, false);
tool_cards::project_detail(
&mut selected,
&detail,
false,
&mut tool_cards::BashNoticeState::default(),
);
let protection = &selected["metadata"]["public_protection"];
let notices = crate::output::tool_card::bash_notices(Some(&detail));
let has_notices = entry.protection_tool_name == Some("bash")
&& (notices.rows.iter().any(|notice| notice.value.is_some())
|| notices.unsupported_notices > 0
|| protection["decision"] != "not_reported");
if protection["decision"] == "not_reported" && !has_notices {
return Ok(());
}
let previous = tool_cards::bytes(&entry.card);
let previous_protection = entry.card["metadata"]["public_protection"].clone();
if protection["decision"] != "not_reported"
&& (entry.protection_tool_name != Some("bash")
|| protection["decision"] == "blocked")
{
entry.card["metadata"]["public_protection"] = protection.clone();
}
if has_notices {
tool_cards::merge_bash_notices(
&mut entry.card,
&detail,
&mut entry.bash_notice_state,
);
}
let decision_changed = entry.card["metadata"]["public_protection"]["decision"]
!= previous_protection["decision"];
if let Some(omission) = selected["omissions"]
.as_array()
.unwrap()
.iter()
.find(|omission| omission["subject"] == "metadata")
{
tool_cards::omit(
&mut entry.card,
"metadata",
omission["reason"].as_str().unwrap(),
None,
);
}
tool_cards::bound_card(&mut entry.card);
self.retained_bytes =
self.retained_bytes - previous + tool_cards::bytes(&entry.card);
(index, decision_changed)
}
ActivityEvent::Delta { .. }
| ActivityEvent::FinalPreview { .. }
| ActivityEvent::UsageSnapshot { .. }
| ActivityEvent::UsageUpdate { .. }
| ActivityEvent::FastObservation { .. } => return Ok(()),
};
let related = self.refresh_hierarchy(Some(index));
self.emit(events, index, authoritative)?;
self.emit_related(events, index, related)
}
fn base(&self, events: &Events, index: usize, state: &str) -> anyhow::Result<()> {
let entry = &self.entries[index];
events.emit("activity", json!({"activity_id":entry.id,"kind":entry.kind,"state":state,"name":entry.name,"summary":state}), true)
}
fn refresh_hierarchy(&mut self, changed: Option<usize>) -> Vec<usize> {
let mut parents = Vec::with_capacity(self.entries.len());
let mut tasks = Vec::with_capacity(self.entries.len());
let mut ancestry_reasons = Vec::with_capacity(self.entries.len());
for (index, entry) in self.entries.iter().enumerate() {
let mut cursor = index;
let mut path = vec![index];
let mut task = entry.task.then(|| entry.id.clone());
let mut valid = true;
while let Some(parent) = self.entries[cursor]
.parent
.as_ref()
.and_then(|parent| self.indices.get(parent))
.copied()
{
if path.contains(&parent) || path.len() > 8 {
valid = false;
break;
}
path.push(parent);
if task.is_none() && self.entries[parent].task {
task = Some(self.entries[parent].id.clone());
}
cursor = parent;
}
parents.push(if valid {
entry
.parent
.as_ref()
.and_then(|parent| self.indices.get(parent))
.copied()
} else {
None
});
tasks.push(if valid { task } else { None });
ancestry_reasons.push(if !valid {
Some("size_limit")
} else if entry
.parent
.as_ref()
.is_some_and(|parent| self.excluded.contains(parent))
{
Some("unsupported_source")
} else if entry.parent.is_some() && parents.last() == Some(&None) {
Some("missing_source")
} else {
None
});
}
let mut children = vec![Vec::new(); self.entries.len()];
for (child, parent) in parents.iter().enumerate() {
if let Some(parent) = parent {
children[*parent].push(child);
}
}
let mut related = Vec::new();
for index in 0..self.entries.len() {
if let Some(changed) = changed
&& index != changed
&& parents[changed] != Some(index)
{
continue;
}
related.push(index);
let child_rows = children[index].iter().take(32).map(|&child_index| {
let child = &self.entries[child_index];
json!({"activity_id":child.id,"task_id":tasks[child_index],"title":child.card["title"],"state":child.card["state"],"outcome":child.card["outcome"],"changes":child.card["changes"]})
}).collect::<Vec<_>>();
let task_children = children[index]
.iter()
.filter(|&&child| self.entries[child].task)
.copied()
.collect::<Vec<_>>();
let tasks_finished = task_children
.iter()
.filter(|&&child| self.entries[child].finished)
.count();
let parent_id = parents[index].map(|parent| self.entries[parent].id.clone());
let unused_cores = (CARDS_PER_RUN - self.entries.len()) * CORE_BYTES;
let entry = &mut self.entries[index];
let previous = tool_cards::bytes(&entry.card);
entry.card["parent_activity_id"] = json!(parent_id);
entry.card["parent_state"] = json!(if entry.parent.is_none() {
"root"
} else if parents[index].is_some() {
"available"
} else if ancestry_reasons[index].is_some_and(|reason| reason != "missing_source") {
"omitted"
} else {
"missing"
});
entry.card["task_id"] = json!(tasks[index]);
entry.card["blocks"]
.as_array_mut()
.unwrap()
.retain(|block| block["type"] != "children");
if !child_rows.is_empty() {
if entry.card["blocks"].as_array().unwrap().len() == 16 {
entry.card["blocks"].as_array_mut().unwrap().pop();
tool_cards::omit(&mut entry.card, "blocks", "size_limit", None);
}
entry.card["blocks"].as_array_mut().unwrap().push(
json!({"type":"children","children":child_rows,"total":children[index].len()}),
);
if children[index].len() > 32 {
tool_cards::omit(
&mut entry.card,
"children",
"size_limit",
Some(children[index].len() - 32),
);
}
}
if entry.card["tool_kind"] == "subagent"
&& (entry.task_total.is_some() || !task_children.is_empty())
{
entry.card["blocks"]
.as_array_mut()
.unwrap()
.retain(|block| block["type"] != "progress");
if entry.card["blocks"].as_array().unwrap().len() < 16 {
let total = entry.task_total.filter(|total| tasks_finished <= *total);
entry.card["blocks"].as_array_mut().unwrap().push(json!({"type":"progress","label":"Tasks finished","completed":tasks_finished,"total":total,"unit":"items"}));
} else {
tool_cards::omit(&mut entry.card, "progress", "size_limit", None);
}
}
entry.card["omissions"]
.as_array_mut()
.unwrap()
.retain(|omission| omission["subject"] != "ancestry");
if let Some(reason) = ancestry_reasons[index] {
tool_cards::omit(&mut entry.card, "ancestry", reason, None);
}
tool_cards::bound_card(&mut entry.card);
if self.retained_bytes - previous + tool_cards::bytes(&entry.card) + unused_cores
> RETAINED_BYTES
{
tool_cards::make_core(&mut entry.card, "budget_limit");
}
self.retained_bytes = self.retained_bytes - previous + tool_cards::bytes(&entry.card);
}
related
}
fn emit_related(
&mut self,
events: &Events,
changed: usize,
related: Vec<usize>,
) -> anyhow::Result<()> {
for index in related {
if index != changed {
self.emit(events, index, false)?;
}
}
Ok(())
}
fn emit(&mut self, events: &Events, index: usize, authoritative: bool) -> anyhow::Result<()> {
let unused_cores = (CARDS_PER_RUN - self.entries.len()) * CORE_BYTES;
let entry = &mut self.entries[index];
if entry.awaiting_detail {
return Ok(());
}
let previous = tool_cards::bytes(&entry.card);
let reserved = entry.last.is_none() || (entry.finished && !entry.terminal_sent);
let mut candidate = entry.card.clone();
if entry.last.is_some() {
candidate["revision"] = json!(candidate["revision"].as_u64().unwrap() + 1);
}
tool_cards::bound_card(&mut candidate);
let cost = |size: usize| {
if reserved {
size.saturating_sub(CORE_BYTES)
} else {
size
}
};
if self.retained_bytes - previous + tool_cards::bytes(&candidate) + unused_cores
> RETAINED_BYTES
|| self.optional_bytes + cost(tool_cards::bytes(&candidate)) > OPTIONAL_BYTES
|| (!reserved && self.optional_snapshots == OPTIONAL_SNAPSHOTS)
{
tool_cards::make_core(&mut candidate, "budget_limit");
tool_cards::omit(&mut candidate, "updates", "budget_limit", None);
}
let signature = snapshot_hash(&candidate);
if entry.last == Some(signature) {
return Ok(());
}
let size = tool_cards::bytes(&candidate);
if !reserved
&& (self.optional_bytes + cost(size) > OPTIONAL_BYTES
|| self.optional_snapshots == OPTIONAL_SNAPSHOTS)
{
if authoritative {
return Err(RunFailure("content_limit").into());
}
candidate["revision"] = entry.card["revision"].clone();
self.retained_bytes = self.retained_bytes - previous + tool_cards::bytes(&candidate);
entry.card = candidate;
return Ok(());
}
let mut core = candidate.clone();
tool_cards::make_core(&mut core, "size_limit");
if size > tool_cards::CARD_BYTES
|| tool_cards::bytes(&core) > CORE_BYTES
|| self.optional_bytes + cost(size) > OPTIONAL_BYTES
|| self.retained_bytes - previous + size + unused_cores > RETAINED_BYTES
{
return Err(RunFailure("content_limit").into());
}
events.emit(
"tool.card",
json!({"activity_id":entry.id,"part":0,"last_part":true,"card":candidate}),
true,
)?;
self.retained_bytes = self.retained_bytes - previous + size;
entry.card = candidate;
entry.last = Some(signature);
entry.terminal_sent |= entry.finished;
self.optional_bytes += cost(size);
if !reserved {
self.optional_snapshots += 1;
}
Ok(())
}
pub(super) fn finish(&mut self, events: &Events) -> anyhow::Result<()> {
self.closed = true;
for index in 0..self.entries.len() {
self.entries[index].awaiting_detail = false;
if !self.entries[index].finished {
let previous = tool_cards::bytes(&self.entries[index].card);
self.entries[index].finished = true;
self.entries[index].card["state"] = json!("finished");
self.entries[index].card["outcome"] = json!(
self.entries[index]
.result_outcome
.as_deref()
.unwrap_or("unknown")
);
self.retained_bytes =
self.retained_bytes - previous + tool_cards::bytes(&self.entries[index].card);
}
}
self.refresh_hierarchy(None);
for index in 0..self.entries.len() {
self.emit(events, index, true)?;
}
Ok(())
}
}
fn snapshot_hash(card: &Value) -> [u8; 32] {
let mut normalized = card.clone();
normalized["revision"] = json!(0);
Sha256::digest(serde_json::to_vec(&normalized).expect("wire card")).into()
}