#[derive(Debug, Clone, Default)]
struct TaskActivity {
driver: Option<&'static str>,
invocations: u32,
last_duration_ms: u64,
accounting: Option<rhei_tui::AccountingRunSummary>,
missing_outputs: Option<(String, Vec<String>)>,
}
#[derive(Debug, Clone)]
struct LedgerRecord {
task: String,
from: String,
to: String,
driver: &'static str,
log_path: std::path::PathBuf,
exit_code: Option<i32>,
duration_ms: u64,
outcome: LedgerOutcome,
}
#[derive(Debug, Clone)]
enum LedgerOutcome {
Completed,
Failed(String),
Cancelled,
TimedOut,
}
pub struct SummarySink {
inner: Mutex<SummaryState>,
}
#[derive(Default)]
struct SummaryState {
inflight: HashMap<u16, &'static str>,
tasks: HashMap<String, TaskActivity>,
ledger: Vec<LedgerRecord>,
usages: Vec<rhei_tui::UsageSummary>,
usage_by_task: HashMap<String, Vec<rhei_tui::UsageSummary>>,
accounting: Option<rhei_tui::AccountingRunSummary>,
}
impl SummarySink {
pub fn new() -> Self {
Self { inner: Mutex::new(SummaryState::default()) }
}
fn snapshot(&self) -> HashMap<String, TaskActivity> {
self.inner.lock().map(|state| state.tasks.clone()).unwrap_or_default()
}
fn ledger(&self) -> Vec<LedgerRecord> {
self.inner.lock().map(|state| state.ledger.clone()).unwrap_or_default()
}
fn accounting(&self) -> Option<rhei_tui::AccountingRunSummary> {
self.inner
.lock()
.ok()
.and_then(|state| {
state
.accounting
.clone()
.or_else(|| rhei_tui::summarize_usage_summaries(state.usages.iter()))
})
}
}
impl Default for SummarySink {
fn default() -> Self {
Self::new()
}
}
impl rhei_tui::EventSink for SummarySink {
fn emit(&self, event: rhei_tui::RunEvent) {
let mut state = match self.inner.lock() {
Ok(state) => state,
Err(_) => return,
};
match event {
rhei_tui::RunEvent::SlotAssigned { slot, task, agent, .. } => {
let driver = if agent.is_some() { "agent" } else { "program" };
state.inflight.insert(slot, driver);
state.tasks.entry(task).or_default().missing_outputs = None;
}
rhei_tui::RunEvent::SlotReleased {
slot,
task,
from,
to,
log_path,
outcome,
exit_code,
duration_ms,
..
} => {
let driver = state.inflight.remove(&slot).unwrap_or("program");
let entry = state.tasks.entry(task.clone()).or_default();
entry.driver = Some(driver);
entry.invocations += 1;
entry.last_duration_ms = duration_ms;
let outcome = match outcome {
rhei_tui::TaskOutcome::Completed => LedgerOutcome::Completed,
rhei_tui::TaskOutcome::Failed(msg) => LedgerOutcome::Failed(msg),
rhei_tui::TaskOutcome::Cancelled => LedgerOutcome::Cancelled,
rhei_tui::TaskOutcome::TimedOut => LedgerOutcome::TimedOut,
};
state.ledger.push(LedgerRecord {
task,
from,
to,
driver,
log_path,
exit_code,
duration_ms,
outcome,
});
}
rhei_tui::RunEvent::UsageReported { task, usage, .. } => {
state.usages.push(usage.clone());
state.usage_by_task.entry(task.clone()).or_default().push(usage);
let accounting = state
.usage_by_task
.get(&task)
.and_then(|usages| rhei_tui::summarize_usage_summaries(usages.iter()));
if let Some(accounting) = accounting {
state.tasks.entry(task).or_default().accounting = Some(accounting);
}
}
rhei_tui::RunEvent::TaskOutputsMissing { task, state: stalled_in, entries } => {
state.tasks.entry(task).or_default().missing_outputs =
Some((stalled_in, entries));
}
rhei_tui::RunEvent::RunFinished { summary } => {
state.accounting = summary.accounting.clone().or_else(|| {
rhei_tui::summarize_usage_summaries(state.usages.iter())
});
}
_ => {}
}
}
}
fn emit_run_report(
input: &std::path::Path,
machines: &rhei_validator::MachineSet,
summary: &SummarySink,
runtime_dir: &std::path::Path,
stats: RunStats,
) {
use std::io::IsTerminal;
let Ok(loaded) = load_plan(input) else {
return;
};
let dry_run = stats.dry_run;
let plan_arg = plan_arg_for_help(input);
let mut report = RunSummaryReport::build(&loaded.rhei, machines, summary, stats, &plan_arg);
if !dry_run {
if let Err(err) = report.write_to_runtime(runtime_dir) {
eprintln!("warning: could not write run report: {err}");
}
}
if std::io::stdout().is_terminal() {
let color = std::env::var_os("NO_COLOR").is_none();
print!("{}", report.render_tty(color));
} else if let Some(report_path) = &report.report_path {
println!("Report: {report_path}");
}
}
fn short_run_id(started_at: std::time::SystemTime) -> String {
let nanos =
started_at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0);
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for b in nanos.to_le_bytes() {
hash ^= b as u64;
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
format!("{:06x}", hash & 0xff_ffff)
}
fn frozen_dashboard_relative_path(
enabled_this_run: bool,
runtime_dir: &std::path::Path,
workspace_root: &std::path::Path,
) -> Option<String> {
if !enabled_this_run {
return None;
}
let path = runtime_dir.join("dashboard.html");
path.exists().then(|| relativize(&path, workspace_root))
}
fn current_command_line() -> String {
let mut args: Vec<String> = std::env::args().collect();
if let Some(first) = args.first_mut() {
*first = "rhei".to_string();
}
args.join(" ")
}
fn collect_initial_states(
rhei: &rhei_core::ast::Rhei,
machines: &rhei_validator::MachineSet,
) -> HashMap<String, String> {
fn walk(
tasks: &[rhei_core::ast::Task],
machines: &rhei_validator::MachineSet,
out: &mut HashMap<String, String>,
) {
for task in tasks {
out.insert(
task.id.to_string(),
normalized_state_name(task.state.as_str(), machines.for_task(&task.id)),
);
walk(&task.children, machines, out);
}
}
let mut out = HashMap::new();
walk(&rhei.tasks, machines, &mut out);
out
}
struct RunReportGuard<'a> {
input: &'a std::path::Path,
machines: &'a rhei_validator::MachineSet,
runtime_dir: std::path::PathBuf,
run_started: std::time::Instant,
run_started_wall: std::time::SystemTime,
run_id: String,
workspace_root: std::path::PathBuf,
command: String,
parallel: usize,
mode: &'static str,
initial_states: HashMap<String, String>,
dry_run: bool,
summary: Option<std::sync::Arc<SummarySink>>,
armed: bool,
}
impl RunReportGuard<'_> {
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for RunReportGuard<'_> {
fn drop(&mut self) {
if !self.armed || self.dry_run {
return;
}
let Some(summary) = self.summary.clone() else {
return;
};
let ledger = summary.ledger();
let agents = ledger.iter().filter(|r| r.driver == "agent").count() as u32;
let programs = ledger.iter().filter(|r| r.driver == "program").count() as u32;
emit_run_report(
self.input,
self.machines,
&summary,
&self.runtime_dir,
RunStats {
agents_spawned: agents,
programs_spawned: programs,
callback_only: 0,
duration: Some(self.run_started.elapsed()),
dashboard: None,
run_id: self.run_id.clone(),
started_at: Some(self.run_started_wall),
workspace_root: self.workspace_root.clone(),
command: self.command.clone(),
parallel: self.parallel,
mode: self.mode,
initial_states: self.initial_states.clone(),
dry_run: false,
},
);
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Marker {
Done,
Gate,
Attention,
Cancelled,
TerminalAtStart,
}
impl Marker {
fn glyph(self) -> char {
match self {
Marker::Done => '✓',
Marker::Gate => '⏸',
Marker::Attention => '!',
Marker::Cancelled => '⊘',
Marker::TerminalAtStart => '·',
}
}
fn color(self) -> &'static str {
match self {
Marker::Done => GREEN,
Marker::Gate => YELLOW,
Marker::Attention => RED,
Marker::Cancelled => DIM,
Marker::TerminalAtStart => DIM,
}
}
fn needs_attention(self) -> bool {
matches!(self, Marker::Gate | Marker::Attention)
}
}
fn state_is_failure(state: &str) -> bool {
matches!(state, "blocked" | "failed")
}
fn classify_marker(state: &str, machine: &rhei_validator::StateMachine) -> Marker {
match state {
"cancelled" | "canceled" => return Marker::Cancelled,
_ if state_is_failure(state) => return Marker::Attention,
_ => {}
}
let def = machine.states.get(state);
if def.map(|d| d.gating).unwrap_or(false) {
Marker::Gate
} else if def.map(|d| d.terminal).unwrap_or(false) {
Marker::Done
} else {
Marker::Attention
}
}
fn marker_for_task(
id: &str,
state: &str,
machine: &rhei_validator::StateMachine,
halt_causes: &HashMap<String, HaltCause>,
) -> Marker {
if is_calm_parent(id, state, machine, halt_causes) {
return Marker::Gate;
}
classify_marker(state, machine)
}
fn is_calm_parent(
id: &str,
state: &str,
machine: &rhei_validator::StateMachine,
halt_causes: &HashMap<String, HaltCause>,
) -> bool {
classify_marker(state, machine) == Marker::Attention
&& !state_is_failure(state)
&& matches!(halt_causes.get(id), Some(HaltCause::WaitingOnDescendants { .. }))
}
struct TaskRow {
depth: usize,
id: String,
state: String,
marker: Marker,
detail: Option<String>,
}
struct AttentionRow {
id: String,
state: String,
reason: String,
next: String,
is_gate: bool,
}
pub struct RunStats {
pub agents_spawned: u32,
pub programs_spawned: u32,
pub callback_only: u32,
pub duration: Option<std::time::Duration>,
pub dashboard: Option<String>,
pub run_id: String,
pub started_at: Option<std::time::SystemTime>,
pub workspace_root: std::path::PathBuf,
pub command: String,
pub parallel: usize,
pub mode: &'static str,
pub initial_states: HashMap<String, String>,
pub dry_run: bool,
}
struct LedgerEntry {
task: String,
from: String,
to: String,
driver: &'static str,
invocation: String,
reason: String,
}
struct InvocationRow {
driver: &'static str,
task: String,
exit: String,
duration_ms: u64,
log: String,
}
struct TaskAccountingRow {
task: String,
cost: String,
total: String,
input: String,
input_cached: String,
output: String,
output_cached: String,
coverage: String,
}
pub struct RunSummaryReport {
title: String,
result: String,
duration: Option<std::time::Duration>,
state_counts: Vec<(String, usize, Marker)>,
total_tasks: usize,
work: String,
accounting: Option<rhei_tui::AccountingRunSummary>,
attention: Vec<AttentionRow>,
rows: Vec<TaskRow>,
dashboard: Option<String>,
run_id: String,
started_at: Option<std::time::SystemTime>,
workspace: String,
command: String,
parallel: usize,
mode: &'static str,
agents_spawned: u32,
programs_spawned: u32,
callback_only: u32,
terminal_at_start: usize,
ledger: Vec<LedgerEntry>,
invocations: Vec<InvocationRow>,
task_accounting: Vec<TaskAccountingRow>,
report_path: Option<String>,
history_path: Option<String>,
}
const RESET: &str = "\x1b[0m";
const BOLD: &str = "\x1b[1m";
const DIM: &str = "\x1b[2m";
const RED: &str = "\x1b[31m";
const GREEN: &str = "\x1b[32m";
const YELLOW: &str = "\x1b[33m";
const BAR_WIDTH: usize = 24;
const MAX_TASK_ROWS: usize = 40;
const MAX_ATTENTION_ROWS: usize = 5;
impl RunSummaryReport {
pub fn build(
rhei: &rhei_core::ast::Rhei,
machines: &rhei_validator::MachineSet,
summary: &SummarySink,
stats: RunStats,
plan_arg: &str,
) -> Self {
let activity = summary.snapshot();
let halt_causes: HashMap<String, HaltCause> = classify_halted_tasks(
rhei,
machines,
&None,
&|id| activity.contains_key(id),
&|id, state| {
activity
.get(id)
.and_then(|entry| entry.missing_outputs.as_ref())
.filter(|(stalled_in, entries)| stalled_in == state && !entries.is_empty())
.map(|(_, entries)| entries.clone())
},
plan_arg,
)
.into_iter()
.map(|(task, cause)| (task.id.to_string(), cause))
.collect();
let mut rows = Vec::new();
let mut attention = Vec::new();
let mut counts: std::collections::BTreeMap<String, (usize, Marker)> =
std::collections::BTreeMap::new();
collect_rows(
&rhei.tasks,
0,
machines,
&activity,
&halt_causes,
&mut rows,
&mut attention,
&mut counts,
);
let mut terminal_at_start = 0usize;
for row in &mut rows {
let was = stats.initial_states.get(&row.id).map(String::as_str);
let unchanged_terminal = was == Some(row.state.as_str())
&& is_terminal_state(&row.state, machines.for_task(&parse_task_id(&row.id)));
if unchanged_terminal {
terminal_at_start += 1;
if row.marker == Marker::Done {
row.marker = Marker::TerminalAtStart;
row.detail = Some("terminal at start".to_string());
}
}
}
let total_tasks = rows.len();
let mut state_counts: Vec<(String, usize, Marker)> =
counts.into_iter().map(|(state, (n, marker))| (state, n, marker)).collect();
state_counts.sort_by_key(|(_, _, marker)| marker_order(*marker));
let no_work = stats.agents_spawned == 0 && stats.programs_spawned == 0;
let advanced_without_work = rows.iter().any(|r| {
r.marker == Marker::Done
&& stats.initial_states.get(&r.id).map(String::as_str) != Some(r.state.as_str())
});
let result = if stats.dry_run {
"dry run — no changes applied".to_string()
} else {
result_phrase(&attention, &rows, no_work, advanced_without_work)
};
let work = format_work(stats.agents_spawned, stats.programs_spawned, stats.callback_only);
let accounting = summary.accounting();
let task_accounting = build_task_accounting_rows(&rows, &activity);
let ledger = build_ledger(
&rows,
&attention,
&halt_causes,
&summary.ledger(),
&stats.initial_states,
machines,
&stats.workspace_root,
);
let invocations = build_invocations(&summary.ledger(), &stats.workspace_root);
Self {
title: rhei.title.clone(),
result,
duration: stats.duration,
state_counts,
total_tasks,
work,
accounting,
attention,
rows,
dashboard: stats.dashboard,
run_id: stats.run_id,
started_at: stats.started_at,
workspace: stats.workspace_root.display().to_string(),
command: stats.command,
parallel: stats.parallel,
mode: stats.mode,
agents_spawned: stats.agents_spawned,
programs_spawned: stats.programs_spawned,
callback_only: stats.callback_only,
terminal_at_start,
ledger,
invocations,
task_accounting,
report_path: None,
history_path: None,
}
}
pub fn render_tty(&self, color: bool) -> String {
let c = Palette::new(color);
let mut out = String::new();
let dur = self.duration.map(format_duration_long).unwrap_or_default();
out.push_str(&format!(
"\n{}Run Report{} {}{}{}",
c.bold, c.reset, c.bold, self.title, c.reset
));
if !dur.is_empty() {
out.push_str(&format!(" {}{}{}", c.dim, dur, c.reset));
}
out.push('\n');
out.push_str(&format!(" {}{}{}\n\n", c.result_color(&self.result), self.result, c.reset));
out.push_str(" States ");
out.push_str(&self.render_bar(&c));
out.push_str(" ");
out.push_str(&self.render_state_labels(&c));
out.push('\n');
out.push_str(&format!(" Work {}\n", self.work));
if let Some(accounting) = &self.accounting {
out.push_str(&format!(
" Cost {} · Total {} · In {} · In cached {} · Out {} · Out cached {} · Coverage {:?}\n",
format_summary_cost(accounting),
format_dimension_value(&accounting.total),
format_dimension_value(&accounting.input_total),
format_dimension_value(&accounting.input_cached_read),
format_dimension_value(&accounting.output_total),
format_dimension_value(&accounting.output_cached_read),
accounting.coverage,
));
}
if !self.attention.is_empty() {
let gated = self.attention.iter().filter(|a| a.is_gate).count();
let blocked = self.attention.len() - gated;
out.push_str(&format!(
"\n{}Attention{} {} gated · {} blocked\n",
c.bold, c.reset, gated, blocked
));
for row in self.attention.iter().take(MAX_ATTENTION_ROWS) {
out.push_str(&format!(
" {}!{} {:<26} {}{:<11}{} {}\n",
c.red, c.reset, row.id, c.dim, row.state, c.reset, row.reason
));
out.push_str(&format!(" {}→ {}{}\n", c.dim, row.next, c.reset));
}
if self.attention.len() > MAX_ATTENTION_ROWS {
out.push_str(&format!(
" {}… {} more in the report{}\n",
c.dim,
self.attention.len() - MAX_ATTENTION_ROWS,
c.reset
));
}
}
out.push_str(&format!(
"\n{}Tasks{} {} tasks · source order\n",
c.bold, c.reset, self.total_tasks
));
out.push_str(&self.render_tree(&c));
out.push('\n');
if let Some(report) = &self.report_path {
out.push_str(&format!("Report {report}\n"));
}
if let Some(history) = &self.history_path {
out.push_str(&format!("History {history}\n"));
}
if let Some(dashboard) = &self.dashboard {
out.push_str(&format!("Dashboard {dashboard}\n"));
}
let trailing_newline = out.ends_with('\n');
let mut trimmed = out.lines().map(str::trim_end).collect::<Vec<_>>().join("\n");
if trailing_newline {
trimmed.push('\n');
}
trimmed
}
pub fn render_markdown(&self) -> String {
let mut out = String::new();
out.push_str(&format!("# Run Report: {}\n\n", self.title));
let when = self
.started_at
.map(format_iso8601_utc)
.map(|ts| format!("{ts} / {}", self.run_id))
.unwrap_or_else(|| self.run_id.clone());
out.push_str(&format!("Run: {when}\n"));
out.push_str(&format!("Workspace: {}\n", self.workspace));
out.push_str(&format!("Command: {}\n", self.command));
out.push_str(&format!("Mode: {} · parallel {}\n", self.mode, self.parallel));
if let Some(dur) = self.duration {
out.push_str(&format!("Duration: {}\n", format_duration_long(dur)));
}
out.push_str(&format!("Result: {}\n", self.result));
if let Some(dashboard) = &self.dashboard {
out.push_str(&format!("Dashboard: {dashboard}\n"));
}
out.push('\n');
out.push_str("| Final states | Count |\n| --- | ---: |\n");
for (state, n, _) in &self.state_counts {
out.push_str(&format!("| {state} | {n} |\n"));
}
out.push('\n');
let could_not_advance = self.attention.len();
out.push_str("| Activity | Count |\n| --- | ---: |\n");
out.push_str(&format!("| agent invocations | {} |\n", self.agents_spawned));
out.push_str(&format!("| program invocations | {} |\n", self.programs_spawned));
out.push_str(&format!("| callback-only transitions | {} |\n", self.callback_only));
out.push_str(&format!("| terminal at start | {} |\n", self.terminal_at_start));
out.push_str(&format!("| could not advance | {could_not_advance} |\n"));
out.push('\n');
if let Some(accounting) = &self.accounting {
out.push_str("| Accounting | Value |\n| --- | ---: |\n");
out.push_str(&format!("| cost | {} |\n", format_summary_cost(accounting)));
out.push_str(&format!(
"| total tokens | {} |\n",
format_dimension_value(&accounting.total)
));
out.push_str(&format!(
"| input tokens | {} |\n",
format_dimension_value(&accounting.input_total)
));
out.push_str(&format!(
"| input cached | {} |\n",
format_dimension_value(&accounting.input_cached_read)
));
out.push_str(&format!(
"| output tokens | {} |\n",
format_dimension_value(&accounting.output_total)
));
out.push_str(&format!(
"| output cached | {} |\n",
format_dimension_value(&accounting.output_cached_read)
));
out.push_str(&format!("| coverage | {:?} |\n", accounting.coverage));
out.push('\n');
}
if self.agents_spawned == 0 && self.programs_spawned == 0 {
out.push_str(
"> No agent or program ran this run. Any task that advanced did so through \
callbacks, transition rules, or outputs that already existed — inspect the \
ledger below before assuming work was performed.\n\n",
);
}
if !self.attention.is_empty() {
out.push_str("## Attention\n\n");
out.push_str("| Task | State | Reason | Next action |\n| --- | --- | --- | --- |\n");
for a in &self.attention {
out.push_str(&format!(
"| {} | {} | {} | {} |\n",
md_cell(&a.id),
md_cell(&a.state),
md_cell(&a.reason),
md_cell(&a.next),
));
}
out.push('\n');
}
out.push_str("## Transition Ledger\n\n");
out.push_str(
"| Task | From | To | Driver | Invocation | Reason |\n\
| --- | --- | --- | --- | --- | --- |\n",
);
for e in &self.ledger {
out.push_str(&format!(
"| {} | {} | {} | {} | {} | {} |\n",
e.task,
md_cell(&e.from),
md_cell(&e.to),
e.driver,
md_link_or_text(&e.invocation),
md_cell(&e.reason),
));
}
out.push('\n');
out.push_str("## Task Final States\n\n");
for row in &self.rows {
let indent = " ".repeat(row.depth);
let detail = row.detail.as_deref().unwrap_or("");
let detail = if detail.is_empty() {
String::new()
} else {
format!(" — {detail}")
};
out.push_str(&format!(
"{indent}- {} `{}` ({}){detail}\n",
row.marker.glyph(),
row.id,
row.state,
));
}
out.push('\n');
if !self.task_accounting.is_empty() {
out.push_str("## Task Costs\n\n");
out.push_str(
"| Task | Cost | Total | Input | Input cached | Output | Output cached | Coverage |\n\
| --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |\n",
);
for row in &self.task_accounting {
out.push_str(&format!(
"| {} | {} | {} | {} | {} | {} | {} | {} |\n",
md_cell(&row.task),
row.cost,
row.total,
row.input,
row.input_cached,
row.output,
row.output_cached,
row.coverage,
));
}
out.push('\n');
}
if !self.invocations.is_empty() {
out.push_str("## Invocations\n\n");
out.push_str(
"| Task | Driver | Exit | Duration | Log |\n| --- | --- | --- | --- | --- |\n",
);
for inv in &self.invocations {
out.push_str(&format!(
"| {} | {} | {} | {} | [{}]({}) |\n",
inv.task,
inv.driver,
inv.exit,
format_duration_short(inv.duration_ms),
inv.log,
inv.log,
));
}
out.push('\n');
}
out
}
pub fn write_to_runtime(&mut self, runtime_dir: &std::path::Path) -> std::io::Result<()> {
let body = self.render_markdown();
let latest = runtime_dir.join("run-report.md");
let history_dir = runtime_dir.join("run-reports");
std::fs::create_dir_all(&history_dir)?;
let stamp = self
.started_at
.map(format_iso8601_utc)
.map(|ts| ts.replace(':', "-"))
.unwrap_or_else(|| "unknown".to_string());
let history = history_dir.join(format!("{stamp}-{}.md", self.run_id));
std::fs::write(&latest, &body)?;
std::fs::write(&history, &body)?;
self.report_path = Some(relativize(&latest, &self.workspace_root_path()));
self.history_path = Some(relativize(&history, &self.workspace_root_path()));
Ok(())
}
fn workspace_root_path(&self) -> std::path::PathBuf {
std::path::PathBuf::from(&self.workspace)
}
fn render_bar(&self, c: &Palette) -> String {
if self.total_tasks == 0 {
return String::new();
}
let mut widths: Vec<usize> = self
.state_counts
.iter()
.map(|(_, n, _)| {
let w = (*n * BAR_WIDTH) / self.total_tasks;
if *n > 0 {
w.max(1)
} else {
0
}
})
.collect();
let mut total: usize = widths.iter().sum();
while total > BAR_WIDTH {
if let Some((idx, _)) =
widths.iter().enumerate().filter(|(_, w)| **w > 1).max_by_key(|(_, w)| **w)
{
widths[idx] -= 1;
total -= 1;
} else {
break;
}
}
let mut bar = String::new();
for ((_, _, marker), w) in self.state_counts.iter().zip(widths) {
if w == 0 {
continue;
}
bar.push_str(c.color(marker.color()));
bar.push_str(&"█".repeat(w));
bar.push_str(c.reset);
}
bar
}
fn render_state_labels(&self, c: &Palette) -> String {
self.state_counts
.iter()
.map(|(state, n, marker)| {
format!("{}{} {}{}", c.color(marker.color()), n, state, c.reset)
})
.collect::<Vec<_>>()
.join(" · ")
}
fn render_tree(&self, c: &Palette) -> String {
let mut out = String::new();
let mut collapsed = 0usize;
let mut shown = 0usize;
for row in &self.rows {
if shown >= MAX_TASK_ROWS && row.marker == Marker::Done {
collapsed += 1;
continue;
}
shown += 1;
let gutter = if row.depth > 0 { "│ ".repeat(row.depth) } else { String::new() };
let detail = row.detail.as_deref().unwrap_or("");
let state_cell = c.colored(row.marker.color(), &row.state);
let state_pad = " ".repeat(11usize.saturating_sub(row.state.chars().count()));
out.push_str(&format!(
" {}{}{}{} {:<width$} {}{} {}\n",
c.dim,
gutter,
c.reset,
c.colored(row.marker.color(), &row.marker.glyph().to_string()),
row.id,
state_cell,
state_pad,
detail,
width = 26usize.saturating_sub(row.depth * 2),
));
}
if collapsed > 0 {
out.push_str(&format!(
" {}… {collapsed} completed tasks collapsed{}\n",
c.dim, c.reset
));
}
out
}
}
#[allow(clippy::too_many_arguments)]
fn collect_rows(
tasks: &[rhei_core::ast::Task],
depth: usize,
machines: &rhei_validator::MachineSet,
activity: &HashMap<String, TaskActivity>,
halt_causes: &HashMap<String, HaltCause>,
rows: &mut Vec<TaskRow>,
attention: &mut Vec<AttentionRow>,
counts: &mut std::collections::BTreeMap<String, (usize, Marker)>,
) {
for task in tasks {
let machine = machines.for_task(&task.id);
let state = normalized_state_name(task.state.as_str(), machine);
let id = task.id.to_string();
let marker = marker_for_task(&id, &state, machine, halt_causes);
let entry = counts.entry(state.clone()).or_insert((0, marker));
entry.0 += 1;
let detail = task_detail(&id, &state, marker, halt_causes, activity);
if marker.needs_attention() && !is_calm_parent(&id, &state, machine, halt_causes) {
let (reason, next) = attention_reason(marker, &id, &state, halt_causes);
attention.push(AttentionRow {
id: id.clone(),
state: state.clone(),
reason,
next,
is_gate: marker == Marker::Gate,
});
}
rows.push(TaskRow { depth, id, state, marker, detail });
collect_rows(
&task.children,
depth + 1,
machines,
activity,
halt_causes,
rows,
attention,
counts,
);
}
}
fn task_detail(
id: &str,
state: &str,
marker: Marker,
halt_causes: &HashMap<String, HaltCause>,
activity: &HashMap<String, TaskActivity>,
) -> Option<String> {
if let Some(act) = activity.get(id) {
let cost = act
.accounting
.as_ref()
.map(|accounting| format!(" · {}", format_summary_cost(accounting)))
.unwrap_or_default();
if let Some(driver) = act.driver {
let label = if act.invocations > 1 {
format!("{driver}×{}", act.invocations)
} else {
driver.to_string()
};
return Some(format!(
"{label} {}{}",
format_duration_short(act.last_duration_ms),
cost
));
}
if !cost.is_empty() {
return Some(cost.trim_start_matches(" · ").to_string());
}
}
match marker {
Marker::Gate | Marker::Attention => {
Some(attention_reason(marker, id, state, halt_causes).0)
}
_ => None,
}
}
fn build_task_accounting_rows(
rows: &[TaskRow],
activity: &HashMap<String, TaskActivity>,
) -> Vec<TaskAccountingRow> {
rows.iter()
.filter_map(|row| {
let accounting = activity.get(&row.id)?.accounting.as_ref()?;
Some(TaskAccountingRow {
task: row.id.clone(),
cost: format_summary_cost(accounting),
total: format_dimension_value(&accounting.total),
input: format_dimension_value(&accounting.input_total),
input_cached: format_dimension_value(&accounting.input_cached_read),
output: format_dimension_value(&accounting.output_total),
output_cached: format_dimension_value(&accounting.output_cached_read),
coverage: format!("{:?}", accounting.coverage),
})
})
.collect()
}
fn attention_reason(
marker: Marker,
id: &str,
state: &str,
halt_causes: &HashMap<String, HaltCause>,
) -> (String, String) {
if let Some(cause) = halt_causes.get(id) {
return cause.describe(id, state);
}
match marker {
Marker::Gate => HaltCause::Gate.describe(id, state),
_ => HaltCause::Stalled.describe(id, state),
}
}
fn result_phrase(
attention: &[AttentionRow],
rows: &[TaskRow],
no_work: bool,
advanced_without_work: bool,
) -> String {
let all_terminal_success =
rows.iter().all(|r| matches!(r.marker, Marker::Done | Marker::TerminalAtStart));
if !attention.is_empty() {
"stopped for human attention".to_string()
} else if all_terminal_success && no_work && advanced_without_work {
"completed — no work spawned".to_string()
} else if all_terminal_success {
"completed".to_string()
} else {
"finished".to_string()
}
}
fn md_cell(value: &str) -> String {
value.replace('|', "\\|").replace('\n', " ")
}
fn md_link_or_text(value: &str) -> String {
match value.split_once(" / ") {
Some((label, path)) => format!("{} / [{}]({})", md_cell(label), path, path),
None => md_cell(value),
}
}
fn relativize(path: &std::path::Path, root: &std::path::Path) -> String {
let rel = path.strip_prefix(root).unwrap_or(path);
rel.components()
.map(|c| c.as_os_str().to_string_lossy())
.collect::<Vec<_>>()
.join("/")
}
fn ledger_outcome_reason(outcome: &LedgerOutcome, exit_code: Option<i32>) -> String {
match outcome {
LedgerOutcome::Completed => match exit_code {
Some(0) | None => "exit 0".to_string(),
Some(code) => format!("exit {code}"),
},
LedgerOutcome::Failed(msg) => {
let msg = msg.lines().next().unwrap_or("").trim();
match exit_code {
Some(code) if msg.is_empty() => format!("failed, exit {code}"),
Some(code) => format!("exit {code}: {msg}"),
None if msg.is_empty() => "failed".to_string(),
None => format!("failed: {msg}"),
}
}
LedgerOutcome::Cancelled => "cancelled".to_string(),
LedgerOutcome::TimedOut => "timed out".to_string(),
}
}
#[allow(clippy::too_many_arguments)]
fn build_ledger(
rows: &[TaskRow],
attention: &[AttentionRow],
halt_causes: &HashMap<String, HaltCause>,
records: &[LedgerRecord],
initial_states: &HashMap<String, String>,
machines: &rhei_validator::MachineSet,
workspace_root: &std::path::Path,
) -> Vec<LedgerEntry> {
let attention_by_id: HashMap<&str, &AttentionRow> =
attention.iter().map(|a| (a.id.as_str(), a)).collect();
let mut ledger = Vec::new();
for row in rows {
let task_records: Vec<&LedgerRecord> =
records.iter().filter(|r| r.task == row.id).collect();
if !task_records.is_empty() {
for rec in &task_records {
let log = relativize(&rec.log_path, workspace_root);
ledger.push(LedgerEntry {
task: row.id.clone(),
from: rec.from.clone(),
to: rec.to.clone(),
driver: rec.driver,
invocation: format!("{} / {}", rec.driver, log),
reason: ledger_outcome_reason(&rec.outcome, rec.exit_code),
});
}
let last_to = task_records.last().map(|r| r.to.as_str());
if matches!(row.marker, Marker::Done | Marker::TerminalAtStart)
&& last_to != Some(row.state.as_str())
{
ledger.push(LedgerEntry {
task: row.id.clone(),
from: last_to.unwrap_or("").to_string(),
to: row.state.clone(),
driver: "callback-only",
invocation: "none".to_string(),
reason: "advanced without spawning work".to_string(),
});
}
continue;
}
let initial = initial_states.get(&row.id).map(String::as_str);
if row.marker == Marker::TerminalAtStart {
ledger.push(LedgerEntry {
task: row.id.clone(),
from: row.state.clone(),
to: "-".to_string(),
driver: "terminal-at-start",
invocation: "none".to_string(),
reason: "already terminal".to_string(),
});
} else if matches!(row.marker, Marker::Attention | Marker::Gate)
&& !is_calm_parent(
&row.id,
&row.state,
machines.for_task(&parse_task_id(&row.id)),
halt_causes,
)
{
let reason = attention_by_id
.get(row.id.as_str())
.map(|a| a.reason.clone())
.unwrap_or_else(|| format!("stalled in non-terminal state {}", row.state));
ledger.push(LedgerEntry {
task: row.id.clone(),
from: row.state.clone(),
to: "-".to_string(),
driver: "blocked",
invocation: "none".to_string(),
reason,
});
} else if initial != Some(row.state.as_str()) {
ledger.push(LedgerEntry {
task: row.id.clone(),
from: initial.unwrap_or("").to_string(),
to: row.state.clone(),
driver: "callback-only",
invocation: "none".to_string(),
reason: "advanced without spawning work".to_string(),
});
} else if is_terminal_state(&row.state, machines.for_task(&parse_task_id(&row.id))) {
ledger.push(LedgerEntry {
task: row.id.clone(),
from: row.state.clone(),
to: "-".to_string(),
driver: "terminal-at-start",
invocation: "none".to_string(),
reason: "already terminal".to_string(),
});
}
}
ledger
}
fn build_invocations(
records: &[LedgerRecord],
workspace_root: &std::path::Path,
) -> Vec<InvocationRow> {
records
.iter()
.map(|rec| InvocationRow {
driver: rec.driver,
task: rec.task.clone(),
exit: match (&rec.outcome, rec.exit_code) {
(LedgerOutcome::Cancelled, _) => "cancelled".to_string(),
(LedgerOutcome::TimedOut, _) => "timed out".to_string(),
(_, Some(code)) => format!("exit {code}"),
(_, None) => "—".to_string(),
},
duration_ms: rec.duration_ms,
log: relativize(&rec.log_path, workspace_root),
})
.collect()
}
fn format_work(agents: u32, programs: u32, callback_only: u32) -> String {
let mut parts = vec![format!("{agents} agents"), format!("{programs} programs")];
if callback_only > 0 {
parts.push(format!("{callback_only} callback-only"));
}
parts.join(" · ")
}
fn marker_order(marker: Marker) -> u8 {
match marker {
Marker::Done => 0,
Marker::Gate => 1,
Marker::Attention => 2,
Marker::Cancelled => 3,
Marker::TerminalAtStart => 4,
}
}
fn format_duration_short(ms: u64) -> String {
if ms < 60_000 {
format!("{:.1}s", ms as f64 / 1000.0)
} else {
format!("{}m{:02}s", ms / 60_000, (ms % 60_000) / 1000)
}
}
fn format_duration_long(d: std::time::Duration) -> String {
let secs = d.as_secs();
if secs < 60 {
format!("{:.1}s", d.as_secs_f64())
} else {
format!("{}m{:02}s", secs / 60, secs % 60)
}
}
struct Palette {
color: bool,
reset: &'static str,
bold: &'static str,
dim: &'static str,
red: &'static str,
}
impl Palette {
fn new(color: bool) -> Self {
Self {
color,
reset: if color { RESET } else { "" },
bold: if color { BOLD } else { "" },
dim: if color { DIM } else { "" },
red: if color { RED } else { "" },
}
}
fn color(&self, code: &'static str) -> &'static str {
if self.color {
code
} else {
""
}
}
fn colored(&self, code: &'static str, text: &str) -> String {
if self.color {
format!("{code}{text}{RESET}")
} else {
text.to_string()
}
}
fn result_color(&self, result: &str) -> &'static str {
if !self.color {
return "";
}
if result.starts_with("stopped — ") {
RED
} else if result.starts_with("stopped") {
YELLOW
} else if result == "completed" {
GREEN
} else {
""
}
}
}
#[cfg(test)]
mod run_summary_tests {
use super::*;
fn machine() -> rhei_validator::StateMachine {
rhei_validator::StateMachine::builtin_default()
}
fn report(tasks: &[(&str, &str)]) -> RunSummaryReport {
let mut md = String::from("# Rhei: Test Plan\n\n## Tasks\n\n");
for (id, state) in tasks {
md.push_str(&format!("### Task {id}: Task {id}\n**State:** {state}\n\n"));
}
let rhei = rhei_core::parse(&md).expect("plan parses");
RunSummaryReport::build(&rhei, &rhei_validator::MachineSet::single(machine()), &SummarySink::new(), test_stats(), "plan.rhei.md")
}
fn test_stats() -> RunStats {
RunStats {
agents_spawned: 2,
programs_spawned: 3,
callback_only: 0,
duration: Some(std::time::Duration::from_secs(5)),
dashboard: None,
run_id: "abc123".to_string(),
started_at: Some(std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_749_115_351)),
workspace_root: std::path::PathBuf::from("examples/test"),
command: "rhei run .".to_string(),
parallel: 4,
mode: "agent",
initial_states: HashMap::new(),
dry_run: false,
}
}
#[test]
fn markers_classify_by_state_class() {
let m = machine();
assert_eq!(classify_marker("completed", &m), Marker::Done);
assert_eq!(classify_marker("blocked", &m), Marker::Attention);
assert_eq!(classify_marker("cancelled", &m), Marker::Cancelled);
}
#[test]
fn a_parent_waiting_on_its_subtree_reads_as_a_calm_pause() {
let m = machine();
let mut causes: HashMap<String, HaltCause> = HashMap::new();
causes.insert(
"plan.1".to_string(),
HaltCause::WaitingOnDescendants { open: "Task plan.1.1 (human-gate)".to_string() },
);
causes.insert("plan.2".to_string(), HaltCause::Stalled);
assert_eq!(classify_marker("pending", &m), Marker::Attention);
assert_eq!(marker_for_task("plan.1", "pending", &m, &causes), Marker::Gate);
assert_eq!(marker_for_task("plan.2", "pending", &m, &causes), Marker::Attention);
assert_eq!(marker_for_task("plan.3", "pending", &m, &causes), Marker::Attention);
let (reason, _) = attention_reason(Marker::Gate, "plan.1", "pending", &causes);
assert!(
reason.contains("waiting on open descendant Task plan.1.1 (human-gate)"),
"{reason}"
);
}
#[test]
fn one_gate_under_three_ancestors_is_counted_once() {
let rhei = rhei_core::parse(
r#"# Rhei: Deep Subtree
---
structure:
maxLevels: 4
---
## Tasks
### Task 1: Top
**State:** work
#### Task 1.1: Middle
**State:** work
##### Task 1.1.1: Inner
**State:** work
###### Task 1.1.1.1: Gated leaf
**State:** human-gate
"#,
)
.expect("plan parses");
let machine = rhei_validator::StateMachine::from_yaml_str(
r#"name: t
version: 1
states:
work:
initial: true
description: work
human-gate:
description: awaiting a human
gating: true
done:
description: terminal
final: true
transitions:
- from: work
to: done
- from: human-gate
to: done
"#,
)
.expect("valid state machine");
let report = RunSummaryReport::build(
&rhei,
&rhei_validator::MachineSet::single(machine),
&SummarySink::new(),
test_stats(),
"plan.rhei.md",
);
assert_eq!(
report.attention.iter().map(|a| a.id.as_str()).collect::<Vec<_>>(),
vec!["1.1.1.1"],
"only the gate itself is halted work"
);
let tty = report.render_tty(false);
assert!(tty.contains("Attention 1 gated · 0 blocked"), "{tty}");
let markdown = report.render_markdown();
assert!(markdown.contains("| could not advance | 1 |"), "{markdown}");
assert_eq!(
report.ledger.iter().filter(|e| e.driver == "blocked").count(),
1,
"one blocked ledger row, not one per ancestor"
);
for id in ["1", "1.1", "1.1.1"] {
let row = report.rows.iter().find(|r| r.id == id).expect("row present");
assert_eq!(row.marker, Marker::Gate, "{id}");
assert!(
row.detail.as_deref().is_some_and(|d| d.contains("waiting on open descendant")),
"{id}: {:?}",
row.detail
);
}
}
#[test]
fn a_failed_parent_keeps_its_attention_marker() {
let m = machine();
let mut causes: HashMap<String, HaltCause> = HashMap::new();
causes.insert(
"plan.1".to_string(),
HaltCause::WaitingOnDescendants { open: "Task plan.1.1 (pending)".to_string() },
);
assert_eq!(marker_for_task("plan.1", "blocked", &m, &causes), Marker::Attention);
}
#[test]
fn plain_render_lists_every_task_with_state() {
let r = report(&[("1", "completed"), ("2", "blocked")]);
let out = r.render_tty(false);
assert!(out.contains("Run Report"), "{out}");
assert!(out.contains("Test Plan"), "{out}");
assert!(out.contains("completed"), "{out}");
assert!(out.contains("blocked"), "{out}");
assert!(!out.contains('\x1b'), "{out}");
}
#[test]
fn attention_block_surfaces_blocked_tasks() {
let r = report(&[("1", "completed"), ("2", "blocked")]);
let out = r.render_tty(false);
assert!(out.contains("Attention"), "{out}");
assert!(out.contains("1 blocked"), "{out}");
assert!(out.contains("stopped for human attention"), "{out}");
}
#[test]
fn all_completed_reads_as_completed() {
let r = report(&[("1", "completed"), ("2", "completed")]);
let out = r.render_tty(false);
assert!(out.contains("completed"), "{out}");
assert!(!out.contains("Attention"), "{out}");
}
#[test]
fn color_render_emits_ansi() {
let r = report(&[("1", "blocked")]);
let out = r.render_tty(true);
assert!(out.contains('\x1b'), "expected ANSI escapes");
}
#[test]
fn duration_formats_short_and_long() {
assert_eq!(format_duration_short(200), "0.2s");
assert_eq!(format_duration_short(8_100), "8.1s");
assert_eq!(format_duration_short(65_000), "1m05s");
assert_eq!(format_duration_long(std::time::Duration::from_secs(724)), "12m04s");
}
fn report_with(tasks: &[(&str, &str)], stats: RunStats) -> RunSummaryReport {
let mut md = String::from("# Rhei: Test Plan\n\n## Tasks\n\n");
for (id, state) in tasks {
md.push_str(&format!("### Task {id}: Task {id}\n**State:** {state}\n\n"));
}
let rhei = rhei_core::parse(&md).expect("plan parses");
RunSummaryReport::build(&rhei, &rhei_validator::MachineSet::single(machine()), &SummarySink::new(), stats, "plan.rhei.md")
}
#[test]
fn markdown_report_has_all_sections() {
let r = report(&[("1", "completed"), ("2", "blocked")]);
let md = r.render_markdown();
assert!(md.starts_with("# Run Report: Test Plan"), "{md}");
assert!(md.contains("Run: 2025-"), "header carries the ISO start: {md}");
assert!(md.contains("| Final states | Count |"), "{md}");
assert!(md.contains("| Activity | Count |"), "{md}");
assert!(md.contains("## Attention"), "{md}");
assert!(md.contains("## Transition Ledger"), "{md}");
assert!(md.contains("## Task Final States"), "{md}");
}
#[test]
fn run_id_is_stable_for_a_given_start() {
let t = std::time::UNIX_EPOCH + std::time::Duration::from_nanos(1_749_115_351_123_456);
assert_eq!(short_run_id(t), short_run_id(t));
assert_eq!(short_run_id(t).len(), 6);
}
#[test]
fn no_work_run_that_advanced_reads_differently() {
let mut initial = HashMap::new();
initial.insert("1".to_string(), "queued".to_string());
let stats = RunStats {
agents_spawned: 0,
programs_spawned: 0,
callback_only: 1,
initial_states: initial,
..test_stats()
};
let r = report_with(&[("1", "completed")], stats);
assert_eq!(r.result, "completed — no work spawned");
let md = r.render_markdown();
assert!(md.contains("No agent or program ran"), "{md}");
assert!(md.contains("| 1 | queued | completed | callback-only |"), "{md}");
}
#[test]
fn terminal_at_start_task_is_marked_calm() {
let mut initial = HashMap::new();
initial.insert("done".to_string(), "completed".to_string());
let stats = RunStats { initial_states: initial, ..test_stats() };
let r = report_with(&[("done", "completed")], stats);
assert_eq!(r.terminal_at_start, 1);
let md = r.render_markdown();
assert!(md.contains("terminal at start"), "{md}");
assert!(md.contains("| done | completed | - | terminal-at-start |"), "{md}");
}
#[test]
fn write_to_runtime_emits_latest_and_history() {
let dir = tempfile::tempdir().expect("tmpdir");
let runtime = dir.path().join("runtime");
let stats =
RunStats { workspace_root: dir.path().to_path_buf(), ..test_stats() };
let mut r = report_with(&[("1", "completed")], stats);
r.write_to_runtime(&runtime).expect("write report");
assert!(runtime.join("run-report.md").exists());
assert_eq!(r.report_path.as_deref(), Some("runtime/run-report.md"));
let history = std::fs::read_dir(runtime.join("run-reports"))
.expect("history dir")
.filter_map(Result::ok)
.count();
assert_eq!(history, 1, "one timestamped history entry written");
}
#[test]
fn dry_run_result_reads_as_preview() {
let stats = RunStats { dry_run: true, ..test_stats() };
let r = report_with(&[("1", "completed")], stats);
assert_eq!(r.result, "dry run — no changes applied");
assert!(r.render_markdown().contains("Result: dry run — no changes applied"));
}
#[test]
fn dashboard_pointer_gated_on_enabled_this_run() {
let dir = tempfile::tempdir().expect("tmpdir");
let runtime = dir.path().join("runtime");
std::fs::create_dir_all(&runtime).unwrap();
std::fs::write(runtime.join("dashboard.html"), "<html>").unwrap();
assert_eq!(frozen_dashboard_relative_path(false, &runtime, dir.path()), None);
assert_eq!(
frozen_dashboard_relative_path(true, &runtime, dir.path()).as_deref(),
Some("runtime/dashboard.html"),
);
}
#[test]
fn md_cell_escapes_pipes_and_newlines() {
assert_eq!(md_cell("a|b"), "a\\|b");
assert_eq!(md_cell("line1\nline2"), "line1 line2");
}
fn summary_with_spawn(task: &str, from: &str, to: &str, agent: bool) -> SummarySink {
use rhei_tui::EventSink;
let s = SummarySink::new();
let log = std::path::PathBuf::from("runtime/logs/x.log");
s.emit(rhei_tui::RunEvent::SlotAssigned {
slot: 0,
task: task.to_string(),
from: from.to_string(),
to: to.to_string(),
agent: agent.then(|| "mock".to_string()),
template_context: None,
log_path: log.clone(),
started_at: std::time::Instant::now(),
wall_clock: std::time::SystemTime::now(),
});
s.emit(rhei_tui::RunEvent::SlotReleased {
slot: 0,
task: task.to_string(),
from: from.to_string(),
to: to.to_string(),
log_path: log,
outcome: rhei_tui::TaskOutcome::Completed,
finished_at: std::time::Instant::now(),
wall_clock: std::time::SystemTime::now(),
exit_code: Some(0),
duration_ms: 1_200,
});
s
}
#[test]
fn ledger_records_trailing_callback_advance_after_spawn() {
let summary = summary_with_spawn("1", "build", "review", true);
let stats = RunStats { initial_states: HashMap::new(), ..test_stats() };
let mut md = String::from("# Rhei: Test Plan\n\n## Tasks\n\n");
md.push_str("### Task 1: Task 1\n**State:** completed\n\n");
let rhei = rhei_core::parse(&md).expect("plan parses");
let report = RunSummaryReport::build(&rhei, &rhei_validator::MachineSet::single(machine()), &summary, stats, "plan.rhei.md");
let md = report.render_markdown();
assert!(md.contains("| 1 | build | review | agent |"), "{md}");
assert!(md.contains("| 1 | review | completed | callback-only |"), "{md}");
}
}