skiagram_core/analysis/
aggregate.rs

1//! Rollups of DEDUPLICATED usage by session / model / day / tool, with cost.
2//!
3//! Inputs are the per-request records from [`super::dedup`] — never raw events
4//! (CLAUDE.md §8.1). Sub-agent transcripts are folded into their parent
5//! session's row so spawned work is attributed, not dropped (§8.3). Day
6//! bucketing is UTC (see [`super::utc_date`]).
7
8use std::collections::{BTreeMap, BTreeSet};
9
10use jiff::civil::Date;
11use jiff::Timestamp;
12use serde::Serialize;
13
14use crate::analysis::dedup::{dedup_session, DedupStats, UsageRecord};
15use crate::analysis::utc_date;
16use crate::model::{EventKind, Session};
17use crate::pricing::PricingTable;
18
19/// Aggregation filters.
20#[derive(Debug, Default, Clone, Copy)]
21pub struct Filter {
22    /// Only count events on/after this UTC date.
23    pub since: Option<Date>,
24}
25
26/// Token + cost rollup over a set of deduplicated requests.
27#[derive(Debug, Default, Clone, Serialize)]
28pub struct Rollup {
29    pub requests: u64,
30    pub input: u64,
31    pub output: u64,
32    pub cache_creation: u64,
33    pub cache_read: u64,
34    /// Requests with at least one unknown core usage field — the sums above are
35    /// best-effort LOWER BOUNDS for those (absence ≠ zero, §8.5).
36    pub incomplete_requests: u64,
37    /// USD cost of the requests whose model has a known price.
38    pub cost_usd: f64,
39    /// Requests that could not be priced (model unknown or not in the snapshot).
40    pub unpriced_requests: u64,
41}
42
43impl Rollup {
44    pub(crate) fn add(&mut self, rec: &UsageRecord, pricing: &PricingTable) {
45        self.requests += 1;
46        let u = &rec.usage;
47        self.input += u.input.unwrap_or(0);
48        self.output += u.output.unwrap_or(0);
49        self.cache_creation += u.cache_creation.unwrap_or(0);
50        self.cache_read += u.cache_read.unwrap_or(0);
51        if u.input.is_none()
52            || u.output.is_none()
53            || u.cache_creation.is_none()
54            || u.cache_read.is_none()
55        {
56            self.incomplete_requests += 1;
57        }
58        match pricing.cost_usd(rec.model.as_deref(), u) {
59            Some(cost) => self.cost_usd += cost,
60            None => self.unpriced_requests += 1,
61        }
62    }
63
64    pub fn total_tokens(&self) -> u64 {
65        self.input + self.output + self.cache_creation + self.cache_read
66    }
67}
68
69/// Per-session line in the summary. Sub-agent transcript spend is folded in.
70#[derive(Debug, Clone, Serialize)]
71pub struct SessionSummary {
72    pub id: String,
73    pub project: Option<String>,
74    pub model: Option<String>,
75    pub started_at: Option<Timestamp>,
76    pub rollup: Rollup,
77    /// Sub-agents attributed to this session (spawn tool calls / folded
78    /// transcripts, whichever evidence is stronger).
79    pub sub_agents: u64,
80    /// Known tokens contributed by folded sub-agent transcripts.
81    pub sub_agent_tokens: u64,
82}
83
84#[derive(Debug, Default, Clone, Serialize)]
85pub struct ToolStat {
86    pub calls: u64,
87    pub input_bytes: u64,
88    /// MCP server, when the tool name carries one (`mcp__<server>__...`).
89    pub server: Option<String>,
90}
91
92/// Everything the renderers (table / JSON / TUI) need.
93#[derive(Debug, Serialize)]
94pub struct Summary {
95    pub agent: String,
96    pub generated_at: Timestamp,
97    pub since: Option<Date>,
98    pub sessions_parsed: u64,
99    pub sessions_failed: u64,
100    /// Unrecognized/corrupt lines skipped leniently across all files.
101    pub skipped_lines: u64,
102    pub dedup: DedupStats,
103    /// Grand totals (deduplicated, includes sub-agent spend).
104    pub totals: Rollup,
105    /// The sub-agent (sidechain) share of `totals`.
106    pub sidechain_totals: Rollup,
107    pub by_model: BTreeMap<String, Rollup>,
108    /// Keyed by UTC date (`YYYY-MM-DD`).
109    pub by_day: BTreeMap<String, Rollup>,
110    /// Sorted by cost, then tokens, descending. Sub-agent transcripts are folded
111    /// into their parent's row.
112    pub by_session: Vec<SessionSummary>,
113    pub by_tool: BTreeMap<String, ToolStat>,
114    /// Models we refused to price (not in the embedded snapshot — §8.7 forbids
115    /// guessing).
116    pub unpriced_models: BTreeSet<String>,
117    pub compactions: u64,
118}
119
120/// Internal accumulator for a session row (parent + folded children).
121#[derive(Default)]
122struct SessionAcc {
123    project: Option<String>,
124    model: Option<String>,
125    started_at: Option<Timestamp>,
126    rollup: Rollup,
127    spawn_calls: u64,
128    children_folded: u64,
129    sub_agent_tokens: u64,
130}
131
132/// Roll deduplicated usage up into a [`Summary`].
133pub fn aggregate(
134    sessions: &[Session],
135    filter: &Filter,
136    sessions_failed: u64,
137    agent: &str,
138    pricing: &PricingTable,
139) -> Summary {
140    let mut summary = Summary {
141        agent: agent.to_string(),
142        generated_at: Timestamp::now(),
143        since: filter.since,
144        sessions_parsed: sessions.len() as u64,
145        sessions_failed,
146        skipped_lines: 0,
147        dedup: DedupStats::default(),
148        totals: Rollup::default(),
149        sidechain_totals: Rollup::default(),
150        by_model: BTreeMap::new(),
151        by_day: BTreeMap::new(),
152        by_session: Vec::new(),
153        by_tool: BTreeMap::new(),
154        unpriced_models: BTreeSet::new(),
155        compactions: 0,
156    };
157    let mut rows: BTreeMap<String, SessionAcc> = BTreeMap::new();
158
159    let in_window = |ts: Option<Timestamp>| match (filter.since, ts) {
160        (None, _) => true,
161        (Some(since), Some(ts)) => utc_date(ts) >= since,
162        (Some(_), None) => false,
163    };
164
165    for session in sessions {
166        summary.skipped_lines += session.skipped_lines;
167
168        let (records, stats) = dedup_session(session, filter.since);
169        summary.dedup.duplicate_lines_collapsed += stats.duplicate_lines_collapsed;
170        summary.dedup.naive_known_tokens += stats.naive_known_tokens;
171        summary.dedup.requests_with_thinking += stats.requests_with_thinking;
172        summary.dedup.requests_with_encrypted_thinking += stats.requests_with_encrypted_thinking;
173        summary.dedup.thinking_chars_total += stats.thinking_chars_total;
174
175        // Tool + compaction stats come straight from events (they carry no usage).
176        for event in &session.events {
177            if !in_window(event.ts) {
178                continue;
179            }
180            if event.kind == EventKind::Compaction {
181                summary.compactions += 1;
182            }
183            for call in &event.tool_calls {
184                let stat = summary.by_tool.entry(call.name.clone()).or_default();
185                stat.calls += 1;
186                stat.input_bytes += call.input_bytes;
187                if stat.server.is_none() {
188                    stat.server.clone_from(&call.server);
189                }
190            }
191        }
192
193        // Attribute the whole transcript to the parent session when this is a
194        // sub-agent file (§8.3); otherwise to itself.
195        let row_id = session
196            .parent_session
197            .clone()
198            .unwrap_or_else(|| session.id.clone());
199        let row = rows.entry(row_id).or_default();
200        if session.parent_session.is_some() {
201            row.children_folded += 1;
202        } else {
203            // Parent metadata wins over anything a child filled in earlier.
204            row.project.clone_from(&session.project);
205            row.model.clone_from(&session.model);
206            row.started_at = session.started_at;
207            row.spawn_calls += session.sub_agents.len() as u64;
208        }
209        if row.project.is_none() {
210            row.project.clone_from(&session.project);
211        }
212
213        for rec in &records {
214            summary.totals.add(rec, pricing);
215            row.rollup.add(rec, pricing);
216            if rec.sidechain {
217                summary.sidechain_totals.add(rec, pricing);
218                row.sub_agent_tokens += rec.usage.known_total();
219            }
220            let model_key = rec.model.clone().unwrap_or_else(|| "(unknown)".into());
221            summary
222                .by_model
223                .entry(model_key)
224                .or_default()
225                .add(rec, pricing);
226            if let Some(ts) = rec.ts {
227                summary
228                    .by_day
229                    .entry(utc_date(ts).to_string())
230                    .or_default()
231                    .add(rec, pricing);
232            }
233            if let Some(model) = &rec.model {
234                if pricing.lookup(model).is_none() {
235                    summary.unpriced_models.insert(model.clone());
236                }
237            }
238        }
239    }
240
241    summary.by_session = rows
242        .into_iter()
243        .filter(|(_, acc)| acc.rollup.requests > 0)
244        .map(|(id, acc)| SessionSummary {
245            id,
246            project: acc.project,
247            model: acc.model,
248            started_at: acc.started_at,
249            // A spawn call and a folded child transcript are usually the same
250            // sub-agent seen from both sides — take the stronger evidence.
251            sub_agents: acc.spawn_calls.max(acc.children_folded),
252            sub_agent_tokens: acc.sub_agent_tokens,
253            rollup: acc.rollup,
254        })
255        .collect();
256    summary.by_session.sort_by(|a, b| {
257        b.rollup
258            .cost_usd
259            .total_cmp(&a.rollup.cost_usd)
260            .then_with(|| b.rollup.total_tokens().cmp(&a.rollup.total_tokens()))
261            .then_with(|| a.id.cmp(&b.id))
262    });
263    summary
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269    use crate::model::{Event, Usage};
270
271    fn assistant_event(request_id: &str, ts: &str, input: u64) -> Event {
272        Event {
273            kind: EventKind::Assistant,
274            ts: Some(ts.parse().unwrap()),
275            request_id: Some(request_id.into()),
276            model: Some("claude-sonnet-4-5".into()),
277            usage: Some(Usage {
278                input: Some(input),
279                output: Some(10),
280                cache_creation: Some(0),
281                cache_read: Some(0),
282                ..Usage::default()
283            }),
284            tool_calls: Vec::new(),
285            sidechain: false,
286            content_summary: None,
287            content_chars: 0,
288            thinking_chars: 0,
289            has_thinking: false,
290            tool_use_id: None,
291            attachment_kind: None,
292            item_count: 0,
293        }
294    }
295
296    fn base_session(id: &str, parent: Option<&str>, events: Vec<Event>) -> Session {
297        Session {
298            id: id.into(),
299            agent: "claude-code".into(),
300            project: Some("proj".into()),
301            model: Some("claude-sonnet-4-5".into()),
302            parent_session: parent.map(str::to_string),
303            started_at: None,
304            ended_at: None,
305            events,
306            sub_agents: Vec::new(),
307            skipped_lines: 0,
308        }
309    }
310
311    /// §8.3: sub-agent transcripts fold into the parent row; nothing is dropped.
312    #[test]
313    fn child_transcripts_fold_into_parent_session() {
314        let parent = base_session(
315            "parent",
316            None,
317            vec![assistant_event("req_p", "2026-06-01T10:00:00Z", 1000)],
318        );
319        let child = base_session(
320            "agent-x",
321            Some("parent"),
322            vec![assistant_event("req_c", "2026-06-01T10:01:00Z", 500)],
323        );
324        let summary = aggregate(
325            &[parent, child],
326            &Filter::default(),
327            0,
328            "claude-code",
329            &PricingTable::embedded(),
330        );
331
332        assert_eq!(summary.by_session.len(), 1, "child folded, not a row");
333        let row = &summary.by_session[0];
334        assert_eq!(row.id, "parent");
335        assert_eq!(row.rollup.input, 1500, "parent + child input");
336        assert_eq!(row.sub_agents, 1);
337        assert_eq!(summary.totals.input, 1500);
338        assert_eq!(summary.sidechain_totals.input, 500);
339    }
340
341    #[test]
342    fn since_is_inclusive_of_the_boundary_date() {
343        let s = base_session(
344            "s",
345            None,
346            vec![
347                assistant_event("req_before", "2026-06-01T23:00:00Z", 111),
348                assistant_event("req_on", "2026-06-02T00:00:00Z", 222),
349            ],
350        );
351        let filter = Filter {
352            since: Some("2026-06-02".parse().unwrap()),
353        };
354        let summary = aggregate(&[s], &filter, 0, "claude-code", &PricingTable::embedded());
355        assert_eq!(summary.totals.requests, 1);
356        assert_eq!(summary.totals.input, 222);
357    }
358}
skiagram_core/analysis/aggregate.rs

skiagram_core/analysis/
aggregate.rs
