diff --git a/rust/src/codex_costs/quota_windows/tests.rs b/rust/src/codex_costs/quota_windows/tests.rs index 004368ac008..fa257ce686b 100644 --- a/rust/src/codex_costs/quota_windows/tests.rs +++ b/rust/src/codex_costs/quota_windows/tests.rs @@ -56,6 +56,8 @@ fn cache_with_source_rows(rows: Vec) -> CostUsageCache { mtime_unix_ms: 1, prefix_hash: 1, rows, + scan_since_key: None, + scan_until_key: None, }, ); cache @@ -101,6 +103,8 @@ fn deduplicates_source_rows_by_file_identity() { mtime_unix_ms: 1, prefix_hash: 1, rows: vec![row(timestamp, 7, 3, Some("gpt-5.5"))], + scan_since_key: None, + scan_until_key: None, }; let mut cache = cache_with_source_rows(Vec::new()); cache.codex_source_rows = HashMap::from([ diff --git a/rust/src/core/jsonl_scanner.rs b/rust/src/core/jsonl_scanner.rs index 3e0a9813bb7..f46a1b69400 100755 --- a/rust/src/core/jsonl_scanner.rs +++ b/rust/src/core/jsonl_scanner.rs @@ -303,6 +303,14 @@ pub struct CodexSourceRowCache { pub mtime_unix_ms: i64, pub prefix_hash: u64, pub rows: Vec, + /// Reporting partition the rows were read for. Rows are filtered to that + /// partition, so they are recovery evidence only for the same range. Kept + /// per entry because a partial pass can replace some entries while the + /// cache-wide range keys still describe an earlier completed scan. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scan_since_key: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scan_until_key: Option, } /// Per-file usage tracking diff --git a/rust/src/core/jsonl_scanner/codex.rs b/rust/src/core/jsonl_scanner/codex.rs index 9a78f8b04da..a839d789730 100644 --- a/rust/src/core/jsonl_scanner/codex.rs +++ b/rust/src/core/jsonl_scanner/codex.rs @@ -42,8 +42,9 @@ use crate::cost_reporting_period::cost_bucket_zone; /// Codex turn id on source rows so Priority trace evidence can be matched. /// Version 7 keys ChatGPT-plan Auto-review requests apart (upstream #4452) so /// cached day aggregates from older parsers, which priced them as unpriced, -/// are rebuilt. Rebuild older artifacts. -pub(crate) const CODEX_CACHE_SCHEMA_VERSION: u32 = 7; +/// are rebuilt. Version 8 rebuilds model pricing corrupted by source-row recovery across +/// different reporting partitions. Rebuild older artifacts. +pub(crate) const CODEX_CACHE_SCHEMA_VERSION: u32 = 8; /// Whether a persisted Codex cache artifact matches the current schema. /// A mismatched artifact (e.g. a pre-64-bit cache from an older release) is diff --git a/rust/src/core/jsonl_scanner/codex/source_rows.rs b/rust/src/core/jsonl_scanner/codex/source_rows.rs index 3d0a8e3a5f5..778cf248561 100644 --- a/rust/src/core/jsonl_scanner/codex/source_rows.rs +++ b/rust/src/core/jsonl_scanner/codex/source_rows.rs @@ -183,9 +183,25 @@ pub(crate) fn row_cache( mtime_unix_ms: source_mtime_unix_ms(metadata.modified().ok()), prefix_hash: source_prefix_hash(file_path, metadata.len())?, rows, + scan_since_key: None, + scan_until_key: None, }) } +impl CodexSourceRowCache { + pub(crate) fn record_scan_range(&mut self, range: &CostUsageDayRange) { + self.scan_since_key = Some(range.scan_since_key.clone()); + self.scan_until_key = Some(range.scan_until_key.clone()); + } + + /// Whether these rows were read for exactly `range`. Entries without a + /// recorded range predate per-entry ranges and never match. + pub(crate) fn was_read_for(&self, range: &CostUsageDayRange) -> bool { + self.scan_since_key.as_deref() == Some(range.scan_since_key.as_str()) + && self.scan_until_key.as_deref() == Some(range.scan_until_key.as_str()) + } +} + /// Verify the retained source prefix before using cached request pricing. /// A same-size rewrite is rejected when its modification time changes; /// append-only growth is accepted only when the indexed prefix is intact. diff --git a/rust/src/core/jsonl_scanner/codex/source_rows_tests.rs b/rust/src/core/jsonl_scanner/codex/source_rows_tests.rs index 1fba1b18f6d..7122e18f860 100644 --- a/rust/src/core/jsonl_scanner/codex/source_rows_tests.rs +++ b/rust/src/core/jsonl_scanner/codex/source_rows_tests.rs @@ -147,6 +147,34 @@ fn source_row_cache_rejects_changed_prefix_and_accepts_append() { assert!(!row_cache_matches(&path, &changed, &cached)); } +#[test] +fn source_row_cache_is_evidence_only_for_the_range_it_was_read_for() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("session.jsonl"); + std::fs::write(&path, b"row\n").unwrap(); + let metadata = std::fs::metadata(&path).unwrap(); + let range = |since: &str, until: &str| CostUsageDayRange { + since_key: since.to_string(), + until_key: until.to_string(), + scan_since_key: since.to_string(), + scan_until_key: until.to_string(), + }; + let thirty_days = range("2026-09-24", "2026-10-23"); + let one_day = range("2026-10-23", "2026-10-23"); + + let mut cached = row_cache(&path, &metadata, Vec::new()).expect("source cache"); + assert!( + !cached.was_read_for(&thirty_days), + "legacy entry has no range" + ); + + cached.record_scan_range(&one_day); + assert!(cached.was_read_for(&one_day)); + // A shorter pass that replaced this entry must not be read back as + // evidence for the longer window, whatever the cache-wide keys say. + assert!(!cached.was_read_for(&thirty_days)); +} + #[test] fn source_row_cache_hashes_the_entire_cached_prefix() { let root = tempfile::tempdir().unwrap(); diff --git a/rust/src/cost_scanner/codex/flat_sources.rs b/rust/src/cost_scanner/codex/flat_sources.rs index e646e75f64d..eec14a3fd80 100644 --- a/rust/src/cost_scanner/codex/flat_sources.rs +++ b/rust/src/cost_scanner/codex/flat_sources.rs @@ -446,6 +446,8 @@ mod tests { mtime_unix_ms: 0, prefix_hash: 0, rows: Vec::new(), + scan_since_key: None, + scan_until_key: None, }, ); cache.codex_fork_rows.insert("from".to_string(), Vec::new()); diff --git a/rust/src/cost_scanner/codex/priority_trace/tests/scanner.rs b/rust/src/cost_scanner/codex/priority_trace/tests/scanner.rs index 5b1fcfb7e49..84dfff7ec64 100644 --- a/rust/src/cost_scanner/codex/priority_trace/tests/scanner.rs +++ b/rust/src/cost_scanner/codex/priority_trace/tests/scanner.rs @@ -756,6 +756,8 @@ fn budget_pruning_rebuilds_priority_days_from_retained_files() { mtime_unix_ms: 0, prefix_hash: 0, rows: vec![standard_row(day, input, turn)], + scan_since_key: None, + scan_until_key: None, }, ); cursor.request_sources.insert( diff --git a/rust/src/cost_scanner/codex/scan.rs b/rust/src/cost_scanner/codex/scan.rs index e6ca018a196..b095aa6377f 100644 --- a/rust/src/cost_scanner/codex/scan.rs +++ b/rust/src/cost_scanner/codex/scan.rs @@ -41,6 +41,14 @@ fn codex_source_row_plan( let recoverable_cache = cache .codex_source_rows .get(&path.to_string_lossy().to_string()) + // Source rows are filtered to the reporting partition. Comparing a + // different window mistakes newly included history for lost pricing + // evidence and turns otherwise known models into unattributed usage. + // A changed partition must establish its evidence from the source; + // durable Priority trace evidence is applied separately below. The + // range is checked per entry: the cache-wide range keys describe the + // last completed scan, and a partial pass can replace entries first. + .filter(|cached| cached.was_read_for(scan_range)) .filter(|cached| !cached.rows.is_empty() && row_cache_matches(path, &metadata, cached)) .cloned(); Some(CodexSourceRowPlan { @@ -53,7 +61,12 @@ fn codex_source_row_plan( /// Apply a source-row plan: recover cached pricing when required, retain the /// resulting rows as evidence, and unprice the file's day map when recovery /// left appended rows unvalidated. -fn apply_codex_source_row_plan(cache: &mut CostUsageCache, key: &str, plan: CodexSourceRowPlan) { +fn apply_codex_source_row_plan( + cache: &mut CostUsageCache, + key: &str, + plan: CodexSourceRowPlan, + scan_range: &CostUsageDayRange, +) { let CodexSourceRowPlan { metadata, source_rows, @@ -68,9 +81,10 @@ fn apply_codex_source_row_plan(cache: &mut CostUsageCache, key: &str, plan: Code } else { source_rows }; - let Some(source_cache) = row_cache(&key_path(key), &metadata, rows) else { + let Some(mut source_cache) = row_cache(&key_path(key), &metadata, rows) else { return; }; + source_cache.record_scan_range(scan_range); if requires_recovery && let Some(usage) = cache.files.get_mut(key) { // Rebuild from recovered rows even when a new row is intentionally // unresolved. Leaving the normal parser's model-priced days in place @@ -413,7 +427,7 @@ pub(super) fn scan_codex_detailed_with_cache( incomplete_processed.push(key); stats.files_deferred = stats.files_deferred.saturating_add(1); } else if let Some(plan) = codex_source_row_plan(&cache, &candidate.path, scan_range) { - apply_codex_source_row_plan(&mut cache, &key, plan); + apply_codex_source_row_plan(&mut cache, &key, plan, scan_range); } } unprocessed.extend(deferred_paths); diff --git a/rust/src/cost_scanner/tests/period.rs b/rust/src/cost_scanner/tests/period.rs index 1daf48ff68b..de4f32b9f8a 100644 --- a/rust/src/cost_scanner/tests/period.rs +++ b/rust/src/cost_scanner/tests/period.rs @@ -153,3 +153,59 @@ fn codex_month_to_date_excludes_the_previous_month() { ); assert_eq!(mtd, 100); } + +#[test] +fn changing_history_windows_keeps_source_model_pricing_complete() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache = root.path().join("cache"); + let today = Local::now().date_naive(); + let folder = sessions.join(today.format("%Y/%m/%d").to_string()); + std::fs::create_dir_all(&folder).unwrap(); + let mut lines = Vec::new(); + for age in [8, 1, 0] { + let timestamp = Local + .from_local_datetime(&(today - Duration::days(age)).and_hms_opt(12, 0, 0).unwrap()) + .earliest() + .unwrap() + .with_timezone(&Utc) + .to_rfc3339(); + lines.push( + serde_json::json!({ + "type": "turn_context", "timestamp": timestamp, + "payload": {"model": "gpt-5"} + }) + .to_string(), + ); + lines.push(serde_json::json!({ + "type": "event_msg", "timestamp": timestamp, + "payload": {"type": "token_count", "info": { + "last_token_usage": {"input_tokens": 100, "cached_input_tokens": 0, "output_tokens": 0} + }} + }).to_string()); + } + std::fs::write(folder.join("history.jsonl"), lines.join("\n") + "\n").unwrap(); + for days in [30, 1, 7, 30, 1, 30] { + let summary = CostScanner::new(days) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache) + .with_sessions_dirs(vec![sessions.clone()]) + .scan_codex(); + assert!(summary.history_coverage_established); + assert!( + !summary.model_pricing_completeness.is_partial(), + "known source models lost pricing after switching to {days} days: {:?}", + summary.model_pricing_completeness + ); + assert_eq!( + summary.input_tokens, + if days >= 9 { + 300 + } else if days > 1 { + 200 + } else { + 100 + } + ); + } +}