| 1 | //! Receipts-only projection of every agent that ran this session (#5479). |
| 2 | //! |
| 3 | //! Separated from roster glyphs and terminal rendering so engine events, |
| 4 | //! protocol parity, and headless audit records can track worker receipts |
| 5 | //! without depending on TUI layout modules. |
| 6 | //! |
| 7 | //! ## The truth rule |
| 8 | //! |
| 9 | //! Every number here is an `Option`, and `None` renders as `—`, never as `0`. |
| 10 | //! The distinction is the whole point: "this worker reported 96,300 input |
| 11 | //! tokens" and "no usage receipt exists for this worker" are different facts, |
| 12 | //! and a rail that prints `0` for the second one is lying in the direction that |
| 13 | //! makes Codewhale look cheap. Nothing here estimates, derives a token count |
| 14 | //! from text, or back-fills a missing receipt — values come from |
| 15 | //! `AgentRunUsage`, which is populated from immutable per-response route |
| 16 | //! audits, or they are absent. |
| 17 | //! |
| 18 | //! A finished agent keeps the numbers it finished with: rows are built from the |
| 19 | //! retained worker record, never recomputed from live state. |
| 20 | |
| 21 | use serde::{Deserialize, Serialize}; |
| 22 | |
| 23 | use crate::tools::subagent::{AgentWorkerRecord, AgentWorkerStatus}; |
| 24 | |
| 25 | /// What a row is doing, collapsed to one glanceable state. |
| 26 | /// |
| 27 | /// Deliberately coarser than `AgentWorkerStatus`: the rail needs a glyph and |
| 28 | /// a sort rank, not the full lifecycle. The precise status stays on the row. |
| 29 | #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] |
| 30 | #[serde(rename_all = "snake_case")] |
| 31 | pub enum RosterState { |
| 32 | Running, |
| 33 | Waiting, |
| 34 | /// Settled because the parent's turn ended before this child did (#5906). |
| 35 | /// |
| 36 | /// Distinct from `Waiting`, which means a person can answer it. Nothing |
| 37 | /// will answer a parked husk; it is continued with `resume_from` or |
| 38 | /// dismissed with `cancel`. |
| 39 | Parked, |
| 40 | Done, |
| 41 | Failed, |
| 42 | Cancelled, |
| 43 | } |
| 44 | |
| 45 | impl RosterState { |
| 46 | #[must_use] |
| 47 | pub const fn is_terminal(self) -> bool { |
| 48 | matches!(self, Self::Done | Self::Failed | Self::Cancelled) |
| 49 | } |
| 50 | |
| 51 | #[must_use] |
| 52 | pub const fn as_str(self) -> &'static str { |
| 53 | match self { |
| 54 | Self::Running => "running", |
| 55 | Self::Waiting => "waiting", |
| 56 | Self::Parked => "parked", |
| 57 | Self::Done => "done", |
| 58 | Self::Failed => "failed", |
| 59 | Self::Cancelled => "cancelled", |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | /// Single-width glyph. Filled = attention, hollow = at rest. |
| 64 | #[must_use] |
| 65 | pub const fn glyph(self) -> &'static str { |
| 66 | match self { |
| 67 | Self::Running => "●", |
| 68 | Self::Waiting => "◐", |
| 69 | // Hollow, dotted: at rest but not finished. |
| 70 | Self::Parked => "◌", |
| 71 | Self::Done => "○", |
| 72 | Self::Failed => "✗", |
| 73 | Self::Cancelled => "⊘", |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | /// Row state for one retained record. |
| 78 | /// |
| 79 | /// `parked_at_turn_end` is checked first and outranks the worker status: |
| 80 | /// a parked child settles as `WaitingForUser` or `Interrupted` like any |
| 81 | /// other, and only this flag separates it from a child that really asked |
| 82 | /// (#5906). |
| 83 | #[must_use] |
| 84 | pub const fn from_record(record: &AgentWorkerRecord) -> Self { |
| 85 | if record.parked_at_turn_end { |
| 86 | return Self::Parked; |
| 87 | } |
| 88 | Self::from_worker(record.status) |
| 89 | } |
| 90 | |
| 91 | #[must_use] |
| 92 | pub const fn from_worker(status: AgentWorkerStatus) -> Self { |
| 93 | match status { |
| 94 | AgentWorkerStatus::Queued |
| 95 | | AgentWorkerStatus::Starting |
| 96 | | AgentWorkerStatus::Running |
| 97 | | AgentWorkerStatus::ModelWait |
| 98 | | AgentWorkerStatus::RunningTool => Self::Running, |
| 99 | AgentWorkerStatus::WaitingForUser => Self::Waiting, |
| 100 | AgentWorkerStatus::Completed => Self::Done, |
| 101 | AgentWorkerStatus::Failed => Self::Failed, |
| 102 | AgentWorkerStatus::Cancelled | AgentWorkerStatus::Interrupted => Self::Cancelled, |
| 103 | } |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | /// One agent's row. Every optional field means "no receipt", not "zero". |
| 108 | #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] |
| 109 | pub struct AgentRosterRow { |
| 110 | pub worker_id: String, |
| 111 | /// Session name, else role, else the fleet type — whichever the user named. |
| 112 | pub display_name: String, |
| 113 | pub model: String, |
| 114 | pub state: RosterState, |
| 115 | pub status: AgentWorkerStatus, |
| 116 | /// The agent's current step or last tool, in one line. `None` when the |
| 117 | /// worker has not reported an event yet. |
| 118 | pub activity: Option<String>, |
| 119 | /// Wall time: elapsed for a live agent, final duration for a finished one. |
| 120 | pub millis: Option<u64>, |
| 121 | pub input_tokens: Option<u64>, |
| 122 | pub output_tokens: Option<u64>, |
| 123 | pub cost_microusd: Option<u64>, |
| 124 | pub steps_taken: u32, |
| 125 | /// Set when this agent was spawned by another; the parent aggregates it. |
| 126 | pub parent_run_id: Option<String>, |
| 127 | pub run_id: String, |
| 128 | } |
| 129 | |
| 130 | impl AgentRosterRow { |
| 131 | /// `n/m done` for a workflow parent, from its children's terminal states. |
| 132 | #[must_use] |
| 133 | pub fn workflow_progress(&self, rows: &[Self]) -> Option<(usize, usize)> { |
| 134 | let children = rows |
| 135 | .iter() |
| 136 | .filter(|row| row.parent_run_id.as_deref() == Some(self.run_id.as_str())) |
| 137 | .collect::<Vec<_>>(); |
| 138 | if children.is_empty() { |
| 139 | return None; |
| 140 | } |
| 141 | let done = children |
| 142 | .iter() |
| 143 | .filter(|row| row.state.is_terminal()) |
| 144 | .count(); |
| 145 | Some((done, children.len())) |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | /// Build the roster from retained worker records. |
| 150 | /// |
| 151 | /// `now_ms` is passed in rather than read from the clock so the projection is a |
| 152 | /// pure function — the caller supplies the same instant it renders with, and |
| 153 | /// tests get deterministic elapsed values. |
| 154 | #[must_use] |
| 155 | pub fn build_agent_roster(records: &[AgentWorkerRecord], now_ms: u64) -> Vec<AgentRosterRow> { |
| 156 | let mut rows: Vec<AgentRosterRow> = records |
| 157 | .iter() |
| 158 | .map(|record| row_from_record(record, now_ms)) |
| 159 | .collect(); |
| 160 | // Oldest first: the rail is a history of the session, and a list that |
| 161 | // reorders itself as agents finish is unreadable while you are watching it. |
| 162 | rows.sort_by(|a, b| a.worker_id.cmp(&b.worker_id)); |
| 163 | rows.sort_by_key(|row| creation_key(records, &row.worker_id)); |
| 164 | // ...except parked husks, which sink below everything still live or |
| 165 | // answerable (#5906). They are the one class of row the operator is not |
| 166 | // meant to scan past to find real work, and the sort is stable so the |
| 167 | // history order survives inside each group. |
| 168 | rows.sort_by_key(|row| row.state == RosterState::Parked); |
| 169 | rows |
| 170 | } |
| 171 | |
| 172 | fn creation_key(records: &[AgentWorkerRecord], worker_id: &str) -> u64 { |
| 173 | records |
| 174 | .iter() |
| 175 | .find(|record| record.spec.worker_id == worker_id) |
| 176 | .map_or(u64::MAX, |record| record.created_at_ms) |
| 177 | } |
| 178 | |
| 179 | #[must_use] |
| 180 | pub fn row_from_record(record: &AgentWorkerRecord, now_ms: u64) -> AgentRosterRow { |
| 181 | let state = RosterState::from_record(record); |
| 182 | AgentRosterRow { |
| 183 | worker_id: record.spec.worker_id.clone(), |
| 184 | display_name: display_name(record), |
| 185 | model: record.spec.model.clone(), |
| 186 | state, |
| 187 | status: record.status, |
| 188 | activity: activity_line(record), |
| 189 | millis: wall_millis(record, now_ms), |
| 190 | input_tokens: record.usage.input_tokens, |
| 191 | output_tokens: record.usage.output_tokens, |
| 192 | cost_microusd: record.usage.cost_microusd, |
| 193 | steps_taken: record.steps_taken, |
| 194 | parent_run_id: record.parent_run_id.clone(), |
| 195 | run_id: record.spec.run_id.clone(), |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | #[must_use] |
| 200 | pub fn display_name(record: &AgentWorkerRecord) -> String { |
| 201 | record |
| 202 | .spec |
| 203 | .session_name |
| 204 | .clone() |
| 205 | .or_else(|| { |
| 206 | record |
| 207 | .spec |
| 208 | .child_route |
| 209 | .as_ref() |
| 210 | .and_then(|route| route.resolved_profile_id.clone()) |
| 211 | .filter(|profile| !profile.trim().is_empty()) |
| 212 | }) |
| 213 | .or_else(|| record.spec.role.clone()) |
| 214 | .unwrap_or_else(|| record.spec.agent_type.as_str().to_string()) |
| 215 | } |
| 216 | |
| 217 | /// The agent's current step or last tool, in one line. |
| 218 | /// |
| 219 | /// Preference order is most-specific-first: the newest event naming a tool, then |
| 220 | /// the newest event carrying a message, then the worker's latest message. A |
| 221 | /// finished worker shows what it finished doing, not a stale "running" line. |
| 222 | #[must_use] |
| 223 | pub fn activity_line(record: &AgentWorkerRecord) -> Option<String> { |
| 224 | let from_events = record.events.iter().rev().find_map(|event| { |
| 225 | event |
| 226 | .tool_name |
| 227 | .as_ref() |
| 228 | .map(|tool| match event.step { |
| 229 | Some(step) => format!("step {step} · {tool}"), |
| 230 | None => tool.clone(), |
| 231 | }) |
| 232 | .or_else(|| event.message.clone()) |
| 233 | }); |
| 234 | from_events |
| 235 | .or_else(|| record.latest_message.clone()) |
| 236 | .or_else(|| record.result_summary.clone()) |
| 237 | .map(|line| one_line(&line)) |
| 238 | } |
| 239 | |
| 240 | /// Collapse to a single line and bound it. Rail rows are one row. |
| 241 | #[must_use] |
| 242 | pub fn one_line(text: &str) -> String { |
| 243 | const MAX_CHARS: usize = 72; |
| 244 | let flattened = text.split_whitespace().collect::<Vec<_>>().join(" "); |
| 245 | if flattened.chars().count() <= MAX_CHARS { |
| 246 | return flattened; |
| 247 | } |
| 248 | let kept = flattened.chars().take(MAX_CHARS - 1).collect::<String>(); |
| 249 | format!("{kept}…") |
| 250 | } |
| 251 | |
| 252 | /// Elapsed for a live agent, final duration for a finished one. |
| 253 | /// |
| 254 | /// `None` when the worker has no start timestamp — a queued worker has not |
| 255 | /// started, and reporting `0s` would imply it had. |
| 256 | #[must_use] |
| 257 | pub fn wall_millis(record: &AgentWorkerRecord, now_ms: u64) -> Option<u64> { |
| 258 | let started = record.started_at_ms?; |
| 259 | let end = record.completed_at_ms.unwrap_or(now_ms); |
| 260 | Some(end.saturating_sub(started)) |
| 261 | } |
| 262 | |
| 263 | /// `3m 29s`, `12s`, `450ms`. Compact because it shares a row. |
| 264 | #[must_use] |
| 265 | pub fn format_duration(millis: u64) -> String { |
| 266 | if millis < 1_000 { |
| 267 | return format!("{millis}ms"); |
| 268 | } |
| 269 | let seconds = millis / 1_000; |
| 270 | if seconds < 60 { |
| 271 | return format!("{seconds}s"); |
| 272 | } |
| 273 | let minutes = seconds / 60; |
| 274 | let rest = seconds % 60; |
| 275 | if minutes < 60 { |
| 276 | return format!("{minutes}m {rest}s"); |
| 277 | } |
| 278 | format!("{}h {}m", minutes / 60, minutes % 60) |
| 279 | } |
| 280 | |
| 281 | /// `96.3k`, `1.2M`, `812`. Never rounds a real count to zero. |
| 282 | #[must_use] |
| 283 | pub fn format_tokens(tokens: u64) -> String { |
| 284 | if tokens < 1_000 { |
| 285 | return tokens.to_string(); |
| 286 | } |
| 287 | if tokens < 1_000_000 { |
| 288 | return format!("{:.1}k", tokens as f64 / 1_000.0); |
| 289 | } |
| 290 | format!("{:.1}M", tokens as f64 / 1_000_000.0) |
| 291 | } |
| 292 | |
| 293 | /// Usage totals across the roster, for a footer line. |
| 294 | /// |
| 295 | /// Returns `None` for a field when *no* row reported it — summing absent |
| 296 | /// receipts into `0` would restate the same lie the per-row rule forbids. |
| 297 | #[must_use] |
| 298 | pub fn roster_totals(rows: &[AgentRosterRow]) -> (Option<u64>, Option<u64>) { |
| 299 | fn total(values: impl Iterator<Item = Option<u64>>) -> Option<u64> { |
| 300 | let reported: Vec<u64> = values.flatten().collect(); |
| 301 | (!reported.is_empty()).then(|| reported.into_iter().fold(0u64, u64::saturating_add)) |
| 302 | } |
| 303 | ( |
| 304 | total(rows.iter().map(|row| row.input_tokens)), |
| 305 | total(rows.iter().map(|row| row.output_tokens)), |
| 306 | ) |
| 307 | } |
| 308 | |
| 309 | /// True when at least one row reported a usage receipt. Callers use this to |
| 310 | /// label the totals line honestly ("partial receipts") instead of implying the |
| 311 | /// number covers every agent. |
| 312 | #[must_use] |
| 313 | pub fn all_rows_have_usage(rows: &[AgentRosterRow]) -> bool { |
| 314 | !rows.is_empty() |
| 315 | && rows |
| 316 | .iter() |
| 317 | .all(|row| row.input_tokens.is_some() || row.output_tokens.is_some()) |
| 318 | } |
| 319 |