skiagram_core/analysis/
dedup.rs

1//! Request-level deduplication — THE critical accounting step (CLAUDE.md §8.1).
2//!
3//! WHY: Claude Code (and agents like it) write one JSONL line per content block,
4//! and every line repeats the request's `message.usage`. A single API request
5//! therefore appears as 2–10 lines sharing one `requestId` (observed on real
6//! data: 83 lines -> 26 requests; 642 -> 262). Naively summing per-line usage
7//! multiplies token counts by that factor.
8//!
9//! RULE: group assistant events by `requestId` and take the field-wise MAX of
10//! each usage counter (see [`crate::model::Usage::merge_max`]). Lines of one
11//! request either repeat identical numbers or grow monotonically while
12//! streaming, so MAX recovers the final per-request value in both cases without
13//! double counting. Events with no `requestId` are never merged with each other.
14//!
15//! This pass also performs thinking ATTRIBUTION (§8.2). Claude Code's
16//! `output_tokens` ALREADY includes extended-thinking tokens (verified on real
17//! data — thinking-only requests report output far larger than their visible
18//! text), so there is no undercount to "fix". Instead we measure the visible
19//! thinking chars per request (the basis for an estimated thinking-token SHARE of
20//! output) and flag requests whose thinking was encrypted/redacted, where that
21//! share is unmeasurable (absence ≠ zero, §8.5).
22
23use std::collections::HashMap;
24
25use jiff::civil::Date;
26use jiff::Timestamp;
27use serde::Serialize;
28
29use crate::analysis::{utc_date, EST_CHARS_PER_TOKEN};
30use crate::model::{EventKind, Session, Usage};
31
32/// One deduplicated API request with its finalized usage.
33#[derive(Debug, Clone, Serialize)]
34pub struct UsageRecord {
35    pub session_id: String,
36    /// Parent session id when this spend came from a sub-agent transcript.
37    pub parent_session: Option<String>,
38    pub project: Option<String>,
39    pub request_id: Option<String>,
40    pub model: Option<String>,
41    pub ts: Option<Timestamp>,
42    pub usage: Usage,
43    /// Request belongs to a sub-agent (sidechain) transcript.
44    pub sidechain: bool,
45    /// Visible extended-thinking chars measured across this request's lines — the
46    /// basis for thinking ATTRIBUTION. These tokens are ALREADY counted inside
47    /// `usage.output` (Claude Code, verified), so this answers "how much of output
48    /// was thinking", never an addition to the billable total. 0 when there was no
49    /// thinking, or when every thinking block was encrypted (`thinking_encrypted`).
50    pub thinking_chars: u64,
51    /// Thinking blocks were present but none were measurable (encrypted/redacted:
52    /// `"thinking":""` + signature). The thinking share of output is unknown here —
53    /// surfaced as such, never guessed (CLAUDE.md §8.5). On the sampled machine
54    /// ~85% of thinking requests were encrypted, so this is the common case.
55    pub thinking_encrypted: bool,
56    /// Thinking blocks were present at all (measurable or encrypted).
57    pub has_thinking: bool,
58}
59
60/// Proof-of-work counters from the dedup pass.
61#[derive(Debug, Default, Clone, Serialize)]
62pub struct DedupStats {
63    /// Extra assistant lines merged away (lines − requests). If this is N > 0,
64    /// a naive parser would have counted N lines' usage twice.
65    pub duplicate_lines_collapsed: u64,
66    /// What naive per-line summation would have reported (for the overcount
67    /// stat shown to the user).
68    pub naive_known_tokens: u64,
69    /// Requests that carried thinking blocks (measurable or encrypted).
70    pub requests_with_thinking: u64,
71    /// Requests whose thinking was present but entirely encrypted/unmeasurable, so
72    /// the thinking share of their output can't be measured (absence ≠ zero, §8.5).
73    pub requests_with_encrypted_thinking: u64,
74    /// Total measured visible thinking chars — the basis for an estimated
75    /// thinking-token attribution. Already included in `output`; never added to
76    /// billable totals.
77    pub thinking_chars_total: u64,
78}
79
80impl DedupStats {
81    /// Estimated thinking tokens from measured visible thinking chars — a LOWER
82    /// BOUND: extended-thinking text tokenizes denser than the `chars/4` rule of
83    /// thumb (~1–3 chars/token observed), and encrypted thinking contributes 0
84    /// measurable chars though it still costs output tokens. These tokens are
85    /// already inside `output` and are never billed again.
86    pub fn thinking_tokens_estimate(&self) -> u64 {
87        self.thinking_chars_total / EST_CHARS_PER_TOKEN
88    }
89}
90
91#[derive(Default)]
92struct Acc {
93    usage: Usage,
94    model: Option<String>,
95    request_id: Option<String>,
96    ts: Option<Timestamp>,
97    lines: u64,
98    thinking_chars: u64,
99    has_thinking: bool,
100    sidechain: bool,
101}
102
103/// Collapse a session's assistant events into per-request usage records.
104///
105/// `since` filters at the event level (UTC date, inclusive) so the returned
106/// stats describe the same window as the records. Events without a timestamp
107/// are excluded when a filter is set (they cannot be proven in-range).
108pub fn dedup_session(session: &Session, since: Option<Date>) -> (Vec<UsageRecord>, DedupStats) {
109    let mut stats = DedupStats::default();
110    let mut keys: Vec<String> = Vec::new();
111    let mut groups: HashMap<String, Acc> = HashMap::new();
112
113    for (idx, event) in session.events.iter().enumerate() {
114        if event.kind != EventKind::Assistant {
115            continue;
116        }
117        if let Some(since) = since {
118            match event.ts {
119                Some(ts) if utc_date(ts) >= since => {}
120                _ => continue,
121            }
122        }
123        // Assistant lines without usage (API errors, synthetic lines) carry no
124        // accountable spend. Their absence is "unknown", not zero (§8.5).
125        let Some(usage) = event.usage else { continue };
126        stats.naive_known_tokens += usage.known_total();
127
128        // No requestId -> a key unique to this line, so it is never merged.
129        let key = event
130            .request_id
131            .clone()
132            .unwrap_or_else(|| format!("\u{0}line:{idx}"));
133        let acc = groups.entry(key.clone()).or_insert_with(|| {
134            keys.push(key);
135            Acc::default()
136        });
137        if acc.lines > 0 {
138            stats.duplicate_lines_collapsed += 1;
139        }
140        acc.lines += 1;
141        acc.usage = acc.usage.merge_max(usage);
142        if acc.model.is_none() {
143            acc.model.clone_from(&event.model);
144        }
145        if acc.request_id.is_none() {
146            acc.request_id.clone_from(&event.request_id);
147        }
148        acc.ts = match (acc.ts, event.ts) {
149            (Some(a), Some(b)) => Some(a.min(b)),
150            (a, b) => a.or(b),
151        };
152        // Thinking is spread across the request's lines — sum what's measurable.
153        acc.thinking_chars += event.thinking_chars;
154        acc.has_thinking |= event.has_thinking;
155        acc.sidechain |= event.sidechain;
156    }
157
158    let records = keys
159        .into_iter()
160        .filter_map(|key| {
161            let acc = groups.remove(&key)?;
162            let thinking_chars = acc.thinking_chars;
163            // Thinking present but nothing measurable -> encrypted/redacted: the
164            // thinking share of output is unknown, surfaced not guessed (§8.5).
165            let thinking_encrypted = acc.has_thinking && thinking_chars == 0;
166            if acc.has_thinking {
167                stats.requests_with_thinking += 1;
168            }
169            if thinking_encrypted {
170                stats.requests_with_encrypted_thinking += 1;
171            }
172            stats.thinking_chars_total += thinking_chars;
173            Some(UsageRecord {
174                session_id: session.id.clone(),
175                parent_session: session.parent_session.clone(),
176                project: session.project.clone(),
177                request_id: acc.request_id,
178                model: acc.model,
179                ts: acc.ts,
180                usage: acc.usage,
181                sidechain: acc.sidechain || session.parent_session.is_some(),
182                thinking_chars,
183                thinking_encrypted,
184                has_thinking: acc.has_thinking,
185            })
186        })
187        .collect();
188
189    (records, stats)
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195    use crate::model::Event;
196
197    fn usage(input: u64, output: u64, cc: u64, cr: u64) -> Usage {
198        Usage {
199            input: Some(input),
200            output: Some(output),
201            cache_creation: Some(cc),
202            cache_read: Some(cr),
203            ..Usage::default()
204        }
205    }
206
207    fn assistant(request_id: Option<&str>, u: Option<Usage>) -> Event {
208        Event {
209            kind: EventKind::Assistant,
210            ts: Some("2026-06-01T10:00:00Z".parse().unwrap()),
211            request_id: request_id.map(str::to_string),
212            model: Some("claude-sonnet-4-5".into()),
213            usage: u,
214            tool_calls: Vec::new(),
215            sidechain: false,
216            content_summary: None,
217            content_chars: 0,
218            thinking_chars: 0,
219            has_thinking: false,
220            tool_use_id: None,
221            attachment_kind: None,
222            item_count: 0,
223        }
224    }
225
226    fn session(events: Vec<Event>) -> Session {
227        Session {
228            id: "s1".into(),
229            agent: "claude-code".into(),
230            project: Some("proj".into()),
231            model: None,
232            parent_session: None,
233            started_at: None,
234            ended_at: None,
235            events,
236            sub_agents: Vec::new(),
237            skipped_lines: 0,
238        }
239    }
240
241    /// The headline case: 3 lines, one request, identical repeated usage.
242    /// A naive sum would report 3x the real spend.
243    #[test]
244    fn identical_duplicate_lines_collapse_to_one_request() {
245        let u = usage(1000, 200, 300, 5000);
246        let s = session(vec![
247            assistant(Some("req_1"), Some(u)),
248            assistant(Some("req_1"), Some(u)),
249            assistant(Some("req_1"), Some(u)),
250        ]);
251        let (records, stats) = dedup_session(&s, None);
252        assert_eq!(records.len(), 1);
253        assert_eq!(records[0].usage, u, "deduped == single request's usage");
254        assert_eq!(stats.duplicate_lines_collapsed, 2);
255        assert_eq!(stats.naive_known_tokens, 3 * u.known_total());
256    }
257
258    /// Streaming partials grow monotonically; MAX recovers the final value.
259    #[test]
260    fn streaming_partials_take_field_wise_max() {
261        let s = session(vec![
262            assistant(Some("req_1"), Some(usage(1000, 50, 300, 5000))),
263            assistant(Some("req_1"), Some(usage(1000, 120, 300, 5000))),
264            assistant(Some("req_1"), Some(usage(1000, 200, 300, 5000))),
265        ]);
266        let (records, _) = dedup_session(&s, None);
267        assert_eq!(records.len(), 1);
268        assert_eq!(records[0].usage.output, Some(200));
269        assert_eq!(records[0].usage.input, Some(1000), "input not multiplied");
270    }
271
272    #[test]
273    fn lines_without_request_id_are_never_merged() {
274        let u = usage(10, 5, 0, 0);
275        let s = session(vec![assistant(None, Some(u)), assistant(None, Some(u))]);
276        let (records, stats) = dedup_session(&s, None);
277        assert_eq!(records.len(), 2);
278        assert_eq!(stats.duplicate_lines_collapsed, 0);
279    }
280
281    #[test]
282    fn missing_usage_is_unknown_not_zero() {
283        let s = session(vec![assistant(Some("req_err"), None)]);
284        let (records, stats) = dedup_session(&s, None);
285        assert!(
286            records.is_empty(),
287            "no usage -> no record, not a zero record"
288        );
289        assert_eq!(stats.naive_known_tokens, 0);
290    }
291
292    /// §8.2: thinking is ATTRIBUTED, not "reconciled". `output_tokens` already
293    /// includes thinking, so we measure visible thinking chars and flag the
294    /// encrypted case (share unmeasurable) — we never invent an undercount.
295    #[test]
296    fn thinking_is_attributed_and_encrypted_is_flagged() {
297        // Visible thinking: measurable chars, already inside the 800 output tokens.
298        let mut visible = assistant(Some("req_vis"), Some(usage(1500, 800, 0, 0)));
299        visible.has_thinking = true;
300        visible.thinking_chars = 1600; // ~400 est. thinking tokens (a lower bound)
301
302        // Encrypted thinking: present but nothing measurable.
303        let mut encrypted = assistant(Some("req_enc"), Some(usage(1500, 500, 0, 0)));
304        encrypted.has_thinking = true;
305        encrypted.thinking_chars = 0;
306
307        let (records, stats) = dedup_session(&session(vec![visible, encrypted]), None);
308        assert_eq!(stats.requests_with_thinking, 2);
309        assert_eq!(stats.requests_with_encrypted_thinking, 1);
310        assert_eq!(stats.thinking_chars_total, 1600);
311        assert_eq!(
312            stats.thinking_tokens_estimate(),
313            400,
314            "1600 / 4, a lower bound"
315        );
316
317        let vis = records
318            .iter()
319            .find(|r| r.request_id.as_deref() == Some("req_vis"))
320            .unwrap();
321        let enc = records
322            .iter()
323            .find(|r| r.request_id.as_deref() == Some("req_enc"))
324            .unwrap();
325        assert_eq!(vis.thinking_chars, 1600);
326        assert!(!vis.thinking_encrypted, "visible thinking is measurable");
327        assert!(enc.has_thinking);
328        assert!(enc.thinking_encrypted, "0 measurable chars -> encrypted");
329        assert_eq!(enc.thinking_chars, 0);
330    }
331
332    #[test]
333    fn since_filters_events_by_utc_date() {
334        let mut old = assistant(Some("req_old"), Some(usage(100, 10, 0, 0)));
335        old.ts = Some("2026-06-01T10:00:00Z".parse().unwrap());
336        let mut new = assistant(Some("req_new"), Some(usage(200, 20, 0, 0)));
337        new.ts = Some("2026-06-02T09:00:00Z".parse().unwrap());
338
339        let since: Date = "2026-06-02".parse().unwrap();
340        let (records, stats) = dedup_session(&session(vec![old, new]), Some(since));
341        assert_eq!(records.len(), 1);
342        assert_eq!(records[0].request_id.as_deref(), Some("req_new"));
343        assert_eq!(stats.naive_known_tokens, 220);
344    }
345}
skiagram_core/analysis/dedup.rs

skiagram_core/analysis/
dedup.rs
