fix(codex): dedupe goal rollout token events (#1237)

* fix(codex): dedupe copied rollout token events

Align the streaming Codex group aggregation dedupe key with the event loader so copied token_count rows from goal or rollout session files are counted once instead of once per session file.

This prevents daily and monthly table reports from double-counting copied Codex history while preserving the same token tuple based duplicate detection already used by JSON/all-agent loading paths.

* test(codex): cover rollout dedupe threading modes

Extend the copied rollout token regression test across both local and parallel Codex aggregation paths.

This addresses CodeRabbit feedback and keeps the dedupe contract covered for each SharedArgs single_thread mode.

* fix(codex): preserve session report dedupe scope

Keep Codex rollout token dedupe across files for daily, weekly, and monthly reports while restoring session-scoped dedupe keys for session reports.

The previous dedupe key intentionally removed session_id to collapse copied root/goal rollout token_count rows, but that same global key ran before session grouping and could drop a distinct session bucket when two session files had identical token usage rows. Session reports now include the session hash in the key, while aggregate period reports use a zero session component to retain cross-session dedupe.

Adds a regression test that identical token usage rows in two session files remain visible as two session groups across both single-threaded and parallel aggregation.

Author: ryoppippi

rust/crates/ccusage/src/adapter/codex/aggregate.rs

@@ -237,7 +237,7 @@ fn add_event_to_groups(
     };
     let timestamp = parse_ts_timestamp(&event.timestamp)
         .ok_or_else(|| crate::cli_error(format!("Invalid Codex timestamp: {}", event.timestamp)))?;
-    if !insert_event_key(event, timestamp, model, seen) {
+    if !insert_event_key(event, timestamp, model, kind, seen) {
         return Ok(());
     }
     add_deduped_event_to_groups(event, model, timestamp, kind, timezone, shared, groups)
@@ -257,7 +257,7 @@ fn add_event_to_groups_local(
         .ok_or_else(|| crate::cli_error(format!("Invalid Codex timestamp: {}", event.timestamp)))?;
     if !aggregation
         .seen
-        .insert(codex_event_key(event, timestamp, model))
+        .insert(codex_event_key(event, timestamp, model, kind))
     {
         return Ok(());
     }
@@ -341,9 +341,10 @@ fn insert_event_key(
     event: &CodexTokenUsageEvent,
     timestamp: crate::TimestampMs,
     model: &str,
+    kind: AgentReportKind,
     seen: &CodexDedupeShards,
 ) -> bool {
-    let key = codex_event_key(event, timestamp, model);
+    let key = codex_event_key(event, timestamp, model, kind);
     let mut hasher = FxHasher::default();
     key.hash(&mut hasher);
     let shard_index = hasher.finish() as usize % seen.len();
@@ -354,10 +355,16 @@ fn codex_event_key(
     event: &CodexTokenUsageEvent,
     timestamp: crate::TimestampMs,
     model: &str,
+    kind: AgentReportKind,
 ) -> CodexEventKey {
+    let (session_hash, session_len) = if kind == AgentReportKind::Session {
+        (hash_text(&event.session_id), event.session_id.len())
+    } else {
+        (0, 0)
+    };
     (
-        hash_text(&event.session_id),
-        event.session_id.len(),
+        session_hash,
+        session_len,
         timestamp,
         hash_text(model),
         model.len(),
@@ -463,6 +470,98 @@ mod tests {
 
     use crate::adapter::codex::paths::CodexUsageSource;
 
+    #[test]
+    fn dedupes_copied_token_usage_across_session_files() {
+        let usage_line = json!({
+            "timestamp": "2026-05-29T08:01:00.000Z",
+            "type": "event_msg",
+            "payload": {
+                "type": "token_count",
+                "info": {
+                    "model": "gpt-5.2",
+                    "last_token_usage": {
+                        "input_tokens": 1_000,
+                        "cached_input_tokens": 100,
+                        "output_tokens": 200,
+                        "reasoning_output_tokens": 20,
+                        "total_tokens": 1_200,
+                    },
+                },
+            },
+        })
+        .to_string();
+        let fixture = fs_fixture!({
+            "sessions/root.jsonl": &usage_line,
+            "sessions/goal.jsonl": &usage_line,
+        });
+        for single_thread in [true, false] {
+            let shared = SharedArgs {
+                single_thread,
+                timezone: Some("UTC".to_string()),
+                ..SharedArgs::default()
+            };
+
+            let groups = load_groups_from_directory(
+                &fixture.path("sessions"),
+                &shared,
+                AgentReportKind::Daily,
+            )
+            .unwrap();
+
+            assert_eq!(groups.len(), 1);
+            let group = groups.get("2026-05-29").unwrap();
+            assert_eq!(group.input_tokens, 1_000);
+            assert_eq!(group.cached_input_tokens, 100);
+            assert_eq!(group.output_tokens, 200);
+            assert_eq!(group.reasoning_output_tokens, 20);
+            assert_eq!(group.total_tokens, 1_200);
+        }
+    }
+
+    #[test]
+    fn keeps_matching_token_usage_in_distinct_session_groups() {
+        let usage_line = json!({
+            "timestamp": "2026-05-29T08:01:00.000Z",
+            "type": "event_msg",
+            "payload": {
+                "type": "token_count",
+                "info": {
+                    "model": "gpt-5.2",
+                    "last_token_usage": {
+                        "input_tokens": 1_000,
+                        "cached_input_tokens": 100,
+                        "output_tokens": 200,
+                        "reasoning_output_tokens": 20,
+                        "total_tokens": 1_200,
+                    },
+                },
+            },
+        })
+        .to_string();
+        let fixture = fs_fixture!({
+            "sessions/root.jsonl": &usage_line,
+            "sessions/goal.jsonl": &usage_line,
+        });
+        for single_thread in [true, false] {
+            let shared = SharedArgs {
+                single_thread,
+                timezone: Some("UTC".to_string()),
+                ..SharedArgs::default()
+            };
+
+            let groups = load_groups_from_directory(
+                &fixture.path("sessions"),
+                &shared,
+                AgentReportKind::Session,
+            )
+            .unwrap();
+
+            assert_eq!(groups.len(), 2);
+            assert_eq!(groups["root"].input_tokens, 1_000);
+            assert_eq!(groups["goal"].input_tokens, 1_000);
+        }
+    }
+
     #[test]
     fn aggregates_active_copy_when_archived_file_has_same_relative_path() {
         let active_usage = [

rust/crates/ccusage/src/adapter/codex/mod.rs

@@ -96,7 +96,7 @@ mod tests {
     }
 
     #[test]
-    fn keeps_matching_grouped_codex_usage_events_from_distinct_sessions() {
+    fn dedupes_matching_grouped_codex_usage_events_from_distinct_sessions() {
         let usage_line = r#"{"timestamp":"2026-01-02T00:00:00.000Z","type":"event_msg","payload":{"type":"token_count","info":{"model":"gpt-5","last_token_usage":{"input_tokens":100,"cached_input_tokens":10,"output_tokens":50,"reasoning_output_tokens":0,"total_tokens":150}}}}"#;
         let fixture = fs_fixture!({
             "sessions/session-a.jsonl": usage_line,
@@ -113,10 +113,10 @@ mod tests {
 
         assert_eq!(groups.len(), 1);
         let group = groups.get("2026-01-02").unwrap();
-        assert_eq!(group.input_tokens, 200);
-        assert_eq!(group.cached_input_tokens, 20);
-        assert_eq!(group.output_tokens, 100);
-        assert_eq!(group.total_tokens, 300);
+        assert_eq!(group.input_tokens, 100);
+        assert_eq!(group.cached_input_tokens, 10);
+        assert_eq!(group.output_tokens, 50);
+        assert_eq!(group.total_tokens, 150);
     }
 
     #[test]