diff --git a/rust/src/core/claude_routed_pricing.rs b/rust/src/core/claude_routed_pricing.rs index 1076992c9f..1f6a750b69 100644 --- a/rust/src/core/claude_routed_pricing.rs +++ b/rust/src/core/claude_routed_pricing.rs @@ -147,13 +147,52 @@ pub fn cost_usd_from_pricing_with_threshold( cache_write: i32, output: i32, ) -> f64 { - let input = input.max(0); - let cache_read = cache_read.max(0); - let cache_write = cache_write.max(0); - let output = output.max(0); + cost_usd_from_u64_counts_with_threshold( + pricing, + threshold_tokens, + input.max(0) as u64, + cache_read.max(0) as u64, + cache_write.max(0) as u64, + output.max(0) as u64, + ) +} + +/// Calculate routed cost for local history counters without narrowing them to +/// the signed API token-count type. +pub(crate) fn cost_usd_from_u64_counts_with_threshold( + pricing: models_dev_pricing::DynamicModelPricing, + threshold_tokens: Option, + input: u64, + cache_read: u64, + cache_write: u64, + output: u64, +) -> f64 { let use_tier = threshold_tokens.is_some_and(|threshold| { - (input as u64) + (cache_read as u64) + (cache_write as u64) > threshold + input + .checked_add(cache_read) + .and_then(|total| total.checked_add(cache_write)) + .is_none_or(|total| total > threshold) }); + let rates = selected_cost_rates(pricing, use_tier); + + (input as f64) * rates.input + + (cache_read as f64) * rates.cache_read + + (cache_write as f64) * rates.cache_write + + (output as f64) * rates.output +} + +#[derive(Debug, Clone, Copy)] +struct SelectedCostRates { + input: f64, + cache_read: f64, + cache_write: f64, + output: f64, +} + +fn selected_cost_rates( + pricing: models_dev_pricing::DynamicModelPricing, + use_tier: bool, +) -> SelectedCostRates { let pick = |base: f64, above: Option| { if use_tier { above.unwrap_or(base) @@ -161,39 +200,34 @@ pub fn cost_usd_from_pricing_with_threshold( base } }; - let input_rate = pick( + let input = pick( pricing.input_cost_per_token, pricing.input_cost_per_token_above_threshold, ); - let cache_read_rate = if use_tier { - pricing - .cache_read_input_cost_per_token_above_threshold - .or(pricing.cache_read_input_cost_per_token) - .unwrap_or(input_rate) - } else { - pricing - .cache_read_input_cost_per_token - .unwrap_or(input_rate) - }; - let cache_write_rate = if use_tier { - pricing - .cache_write_input_cost_per_token_above_threshold - .or(pricing.cache_write_input_cost_per_token) - .unwrap_or(input_rate) - } else { - pricing - .cache_write_input_cost_per_token - .unwrap_or(input_rate) - }; - let output_rate = pick( - pricing.output_cost_per_token, - pricing.output_cost_per_token_above_threshold, - ); - (input as f64) * input_rate - + (cache_read as f64) * cache_read_rate - + (cache_write as f64) * cache_write_rate - + (output as f64) * output_rate + SelectedCostRates { + input, + cache_read: if use_tier { + pricing + .cache_read_input_cost_per_token_above_threshold + .or(pricing.cache_read_input_cost_per_token) + .unwrap_or(input) + } else { + pricing.cache_read_input_cost_per_token.unwrap_or(input) + }, + cache_write: if use_tier { + pricing + .cache_write_input_cost_per_token_above_threshold + .or(pricing.cache_write_input_cost_per_token) + .unwrap_or(input) + } else { + pricing.cache_write_input_cost_per_token.unwrap_or(input) + }, + output: pick( + pricing.output_cost_per_token, + pricing.output_cost_per_token_above_threshold, + ), + } } fn effective_threshold(provider: &str, model: &str, catalog_threshold: Option) -> Option { diff --git a/rust/src/core/cost_pricing/claude.rs b/rust/src/core/cost_pricing/claude.rs index 8cdab4f34d..32e4aa4e36 100644 --- a/rust/src/core/cost_pricing/claude.rs +++ b/rust/src/core/cost_pricing/claude.rs @@ -86,20 +86,38 @@ impl CostUsagePricing { cache_read_input_tokens: i32, cache_creation_input_tokens: i32, output_tokens: i32, + ) -> f64 { + Self::claude_cost_usd_u64_from_resolution( + resolution, + u64::try_from(input_tokens).unwrap_or(0), + u64::try_from(cache_read_input_tokens).unwrap_or(0), + u64::try_from(cache_creation_input_tokens).unwrap_or(0), + u64::try_from(output_tokens).unwrap_or(0), + ) + } + + /// Calculate cost from a resolved Claude pricing source without narrowing + /// untrusted local-history counters to the API-oriented signed type. + pub(crate) fn claude_cost_usd_u64_from_resolution( + resolution: ClaudePricingResolution, + input_tokens: u64, + cache_read_input_tokens: u64, + cache_creation_input_tokens: u64, + output_tokens: u64, ) -> f64 { match resolution { ClaudePricingResolution::BuiltIn(pricing) => { fn tiered( - tokens: i32, + tokens: u64, base: f64, above: Option, threshold: Option, ) -> f64 { - let tokens = tokens.max(0); match (threshold, above) { (Some(thresh), Some(above_rate)) => { + let thresh = u64::try_from(thresh).unwrap_or(0); let below = tokens.min(thresh); - let over = (tokens - thresh).max(0); + let over = tokens.saturating_sub(thresh); (below as f64) * base + (over as f64) * above_rate } _ => (tokens as f64) * base, @@ -131,7 +149,7 @@ impl CostUsagePricing { ClaudePricingResolution::ModelsDev { pricing, threshold_tokens, - } => claude_routed_pricing::cost_usd_from_pricing_with_threshold( + } => claude_routed_pricing::cost_usd_from_u64_counts_with_threshold( pricing, threshold_tokens, input_tokens, @@ -196,3 +214,85 @@ impl CostUsagePricing { claude_routed_pricing::input_cost_per_token(model, Self::normalize_claude_model(model)) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn routed_pricing() -> (models_dev_pricing::DynamicModelPricing, Option) { + let snapshot = models_dev_pricing::ModelsDevPricingSnapshot::from_catalog_json_for_tests( + r#"{ + "anthropic": {"models": {"threshold-fixture": {"id": "threshold-fixture", "cost": { + "input": 2, "output": 4, "cache_read": 0.25, "cache_write": 3, + "context_over_200k": {"input": 7, "output": 11, "cache_read": 0.5, "cache_write": 9} + }}}} + }"#, + ) + .expect("pricing fixture"); + let resolution = CostUsagePricing::resolve_claude_pricing( + "anthropic/threshold-fixture", + "anthropic/threshold-fixture", + Some(&snapshot), + ) + .expect("Models.dev pricing"); + let ClaudePricingResolution::ModelsDev { + pricing, + threshold_tokens, + } = resolution + else { + panic!("expected Models.dev pricing"); + }; + (pricing, threshold_tokens) + } + + #[test] + fn models_dev_u64_cost_uses_routed_rates_for_every_token_field() { + let (pricing, threshold) = routed_pricing(); + assert_eq!(threshold, Some(200_000)); + + for (input, cache_read, cache_write, output, expected_usd) in [ + (10_000, 2_000, 1_000, 500, 0.0255), + (199_999, 1, 0, 25, 0.400_098_25), + (200_000, 1, 0, 25, 1.400_275_5), + (220_000, 10_000, 2_000, 50, 1.563_55), + ] { + let actual = CostUsagePricing::claude_cost_usd_u64_from_resolution( + ClaudePricingResolution::ModelsDev { + pricing, + threshold_tokens: threshold, + }, + input, + cache_read, + cache_write, + output, + ); + let routed = claude_routed_pricing::cost_usd_from_u64_counts_with_threshold( + pricing, + threshold, + input, + cache_read, + cache_write, + output, + ); + assert!((actual - expected_usd).abs() < 1e-12); + assert!((actual - routed).abs() < 1e-12); + } + } + + #[test] + fn models_dev_u64_counter_overflow_selects_above_threshold_rates() { + let (pricing, threshold) = routed_pricing(); + let actual = CostUsagePricing::claude_cost_usd_u64_from_resolution( + ClaudePricingResolution::ModelsDev { + pricing, + threshold_tokens: threshold, + }, + u64::MAX, + 1, + 0, + 0, + ); + let expected = (u64::MAX as f64) * 7e-6 + 0.5e-6; + assert!((actual - expected).abs() < 1e-6); + } +} diff --git a/rust/src/cost_scanner.rs b/rust/src/cost_scanner.rs index 619f9ba7b0..3c7fe4a06d 100755 --- a/rust/src/cost_scanner.rs +++ b/rust/src/cost_scanner.rs @@ -435,7 +435,9 @@ struct ClaudeUsageRecord { output: u64, cache_create: u64, cache_read: u64, - cost: f64, + /// `None` means pricing was unavailable or produced a non-finite value. + /// Keep that distinct from a real zero-dollar row. + cost: Option, } /// Exact Claude request rows retained for quota-window projection. @@ -468,6 +470,7 @@ struct ClaudeFileScanResult { malformed_lines: u32, incomplete_requests: u32, read_failures: u32, + aggregation_failures: u32, } impl ClaudeFileScanResult { @@ -478,10 +481,16 @@ impl ClaudeFileScanResult { .incomplete_requests .saturating_add(other.incomplete_requests); self.read_failures = self.read_failures.saturating_add(other.read_failures); + self.aggregation_failures = self + .aggregation_failures + .saturating_add(other.aggregation_failures); } fn is_complete(self) -> bool { - self.malformed_lines == 0 && self.incomplete_requests == 0 && self.read_failures == 0 + self.malformed_lines == 0 + && self.incomplete_requests == 0 + && self.read_failures == 0 + && self.aggregation_failures == 0 } } @@ -590,23 +599,35 @@ impl CostScanner { if projects_dir.exists() { let mut seen = HashSet::new(); let mut pricing = ClaudeScanPricingResolver::default(); - let mut handle_file = |path: &Path| { - let file_result = scan_claude_file_with_pricing( - path, - &cutoff, - &mut seen, - cancel, - &mut pricing, - |record| { - add_claude_record_to_summary(&mut summary, record); - }, - ); - if file_result.counted > 0 { - summary.sessions_count += 1; - } - claude_scan.absorb(file_result); + let traversal_read_failures = { + let mut handle_file = |path: &Path| { + let mut aggregation_complete = true; + let mut file_result = scan_claude_file_with_pricing( + path, + &cutoff, + &mut seen, + cancel, + &mut pricing, + |record| { + aggregation_complete &= record.timestamp.is_some(); + aggregation_complete &= + add_claude_record_to_summary(&mut summary, record); + }, + ); + if !aggregation_complete { + file_result.aggregation_failures = + file_result.aggregation_failures.saturating_add(1); + } + if file_result.counted > 0 { + summary.sessions_count += 1; + } + claude_scan.absorb(file_result); + }; + self.walk_claude_files(&projects_dir, &cutoff, cancel, &mut handle_file) }; - self.walk_claude_files(&projects_dir, &cutoff, cancel, &mut handle_file); + claude_scan.read_failures = claude_scan + .read_failures + .saturating_add(traversal_read_failures); } // OMP / pi-compatible anthropic rows, deduped across shared files. @@ -661,35 +682,45 @@ impl CostScanner { let mut quota_records = Vec::new(); let mut scan_result = ClaudeFileScanResult::default(); - let mut missing_timestamp = false; if projects_dir.exists() { let mut seen = HashSet::new(); let mut pricing = ClaudeScanPricingResolver::default(); - self.walk_claude_files(&projects_dir, &cutoff, cancel, &mut |path| { - let mut file_has_usage = false; - let file_result = scan_claude_file_with_pricing( - path, - &cutoff, - &mut seen, - cancel, - &mut pricing, - |record| { - file_has_usage = true; - add_claude_record_to_summary(&mut summary, record); - add_claude_record_to_daily_costs(&mut daily_cost, record); - add_claude_record_to_daily_tokens(&mut daily_tokens, record); - if let Some(quota_record) = quota_history_record_from_usage(record) { - quota_records.push(quota_record); - } else { - missing_timestamp = true; - } - }, - ); - if file_has_usage { - summary.sessions_count += 1; - } - scan_result.absorb(file_result); - }); + let traversal_read_failures = + self.walk_claude_files(&projects_dir, &cutoff, cancel, &mut |path| { + let mut file_has_usage = false; + let mut aggregation_complete = true; + let mut file_result = scan_claude_file_with_pricing( + path, + &cutoff, + &mut seen, + cancel, + &mut pricing, + |record| { + file_has_usage = true; + aggregation_complete &= record.timestamp.is_some(); + aggregation_complete &= + add_claude_record_to_summary(&mut summary, record); + aggregation_complete &= + add_claude_record_to_daily_costs(&mut daily_cost, record); + aggregation_complete &= + add_claude_record_to_daily_tokens(&mut daily_tokens, record); + if let Some(quota_record) = quota_history_record_from_usage(record) { + quota_records.push(quota_record); + } + }, + ); + if !aggregation_complete { + file_result.aggregation_failures = + file_result.aggregation_failures.saturating_add(1); + } + if file_has_usage { + summary.sessions_count += 1; + } + scan_result.absorb(file_result); + }); + scan_result.read_failures = scan_result + .read_failures + .saturating_add(traversal_read_failures); } crate::pi_session_cost::scan_pi_compatible_into( @@ -700,10 +731,7 @@ impl CostScanner { &mut HashSet::new(), ); - let complete = projects_dir.exists() - && !is_cancelled(cancel) - && scan_result.is_complete() - && !missing_timestamp; + let complete = projects_dir.exists() && !is_cancelled(cancel) && scan_result.is_complete(); finalize_claude_summary( &mut summary, projects_dir.exists(), @@ -801,36 +829,52 @@ impl CostScanner { cutoff: &DateTime, cancel: Option<&AtomicBool>, on_file: &mut F, - ) where + ) -> u32 + where F: FnMut(&Path), { if is_cancelled(cancel) { - return; + return 0; } let entries = match fs::read_dir(dir) { Ok(e) => e, - Err(_) => return, + Err(_) => return 1, }; - for entry in entries.flatten() { + let mut read_failures = 0u32; + for entry in entries { if is_cancelled(cancel) { break; } + let entry = match entry { + Ok(entry) => entry, + Err(_) => { + read_failures = read_failures.saturating_add(1); + continue; + } + }; let path = entry.path(); - if path.is_dir() { - self.walk_claude_files(&path, cutoff, cancel, on_file); - } else if path.extension().is_some_and(|e| e == "jsonl") { - // Check file modification time - if let Ok(metadata) = fs::metadata(&path) - && let Ok(modified) = metadata.modified() - { - let modified_dt: DateTime = modified.into(); - if modified_dt >= *cutoff { - on_file(&path); + match fs::metadata(&path) { + Ok(metadata) if metadata.is_dir() => { + read_failures = read_failures + .saturating_add(self.walk_claude_files(&path, cutoff, cancel, on_file)); + } + Ok(metadata) if path.extension().is_some_and(|e| e == "jsonl") => { + match metadata.modified() { + Ok(modified) => { + let modified_dt: DateTime = modified.into(); + if modified_dt >= *cutoff { + on_file(&path); + } + } + Err(_) => read_failures = read_failures.saturating_add(1), } } + Ok(_) => {} + Err(_) => read_failures = read_failures.saturating_add(1), } } + read_failures } } @@ -854,20 +898,6 @@ where scan_claude_file_with_pricing(path, cutoff, seen, cancel, &mut pricing, on_record).counted } -fn for_each_claude_usage_record_with_pricing( - path: &Path, - cutoff: &DateTime, - seen: &mut HashSet, - cancel: Option<&AtomicBool>, - pricing: &mut ClaudeScanPricingResolver, - on_record: F, -) -> usize -where - F: FnMut(&ClaudeUsageRecord), -{ - scan_claude_file_with_pricing(path, cutoff, seen, cancel, pricing, on_record).counted -} - fn scan_claude_file_with_pricing( path: &Path, cutoff: &DateTime, @@ -1007,7 +1037,7 @@ fn claude_usage_record_from_event_with_pricing( let cache_create_1h = usage.one_hour_cache_creation_tokens(cache_create); let pricing_known = pricing.is_known(model); - let cost = pricing.cost_usd_with_cache_ttl( + let computed_cost = pricing.cost_usd_with_cache_ttl( model, input, cache_create, @@ -1015,6 +1045,7 @@ fn claude_usage_record_from_event_with_pricing( cache_read, output, ); + let cost = computed_cost.is_finite().then_some(computed_cost); Some(ClaudeUsageRecord { model: model.to_string(), @@ -1033,25 +1064,52 @@ fn claude_usage_record_from_event_with_pricing( }) } -fn add_claude_record_to_summary(summary: &mut CostSummary, record: &ClaudeUsageRecord) { +fn add_claude_record_to_summary(summary: &mut CostSummary, record: &ClaudeUsageRecord) -> bool { if !record.pricing_known { summary.unknown_models.insert(record.model.clone()); } - summary.input_tokens += record.input; - summary.output_tokens += record.output; - summary.cached_tokens += record.cache_create + record.cache_read; - summary.total_cost_usd += record.cost; + let mut complete = checked_add_assign(&mut summary.input_tokens, record.input); + complete &= checked_add_assign(&mut summary.output_tokens, record.output); + let cached = record.cache_create.checked_add(record.cache_read); + complete &= cached.is_some_and(|value| checked_add_assign(&mut summary.cached_tokens, value)); - *summary.by_model.entry(record.model.clone()).or_insert(0.0) += record.cost; + if let Some(cost) = record.cost { + complete &= checked_add_finite(&mut summary.total_cost_usd, cost); + complete &= checked_add_finite( + summary.by_model.entry(record.model.clone()).or_insert(0.0), + cost, + ); + } else { + complete = false; + } let model_tokens = summary .by_model_tokens .entry(record.model.clone()) .or_default(); - model_tokens.input_tokens += record.input; - model_tokens.output_tokens += record.output; - model_tokens.cached_tokens += record.cache_create + record.cache_read; + complete &= checked_add_assign(&mut model_tokens.input_tokens, record.input); + complete &= checked_add_assign(&mut model_tokens.output_tokens, record.output); + complete &= + cached.is_some_and(|value| checked_add_assign(&mut model_tokens.cached_tokens, value)); + complete +} + +fn checked_add_assign(total: &mut u64, value: u64) -> bool { + let Some(sum) = total.checked_add(value) else { + return false; + }; + *total = sum; + true +} + +fn checked_add_finite(total: &mut f64, value: f64) -> bool { + let sum = *total + value; + if !value.is_finite() || !sum.is_finite() { + return false; + } + *total = sum; + true } fn quota_history_record_from_usage(record: &ClaudeUsageRecord) -> Option { @@ -1080,9 +1138,9 @@ fn quota_history_record_from_usage(record: &ClaudeUsageRecord) -> Option= 0.0, + cost_is_complete: record.pricing_known && record.cost.is_some_and(|cost| cost >= 0.0), dedup_key, attribution: ClaudeHistoryAttribution::Unavailable, }) @@ -1094,9 +1152,9 @@ fn quota_history_record_from_usage(record: &ClaudeUsageRecord) -> Option>, record: &ClaudeUsageRecord, -) { +) -> bool { let Some(timestamp) = record.timestamp else { - return; + return true; }; let date_str = timestamp .with_timezone(&Local) @@ -1104,8 +1162,18 @@ fn add_claude_record_to_daily_costs( .format("%Y-%m-%d") .to_string(); if let Some(cost) = daily_costs.get_mut(&date_str) { - *cost = Some(cost.unwrap_or(0.0) + record.cost); + let Some(record_cost) = record.cost else { + *cost = None; + return false; + }; + let sum = cost.unwrap_or(0.0) + record_cost; + if !sum.is_finite() { + *cost = None; + return false; + } + *cost = Some(sum); } + true } /// Check if any cost usage sources are available @@ -1188,20 +1256,32 @@ pub fn get_daily_cost_history(provider: &str, days: u32) -> Vec<(String, Option< let mut seen = HashSet::new(); let mut pricing = ClaudeScanPricingResolver::default(); let mut claude_scan = ClaudeFileScanResult::default(); - let mut handle_file = |path: &Path| { - let file_result = scan_claude_file_with_pricing( - path, - &cutoff, - &mut seen, - None, - &mut pricing, - |record| { - add_claude_record_to_daily_costs(&mut daily_costs, record); - }, - ); - claude_scan.absorb(file_result); + let traversal_read_failures = { + let mut handle_file = |path: &Path| { + let mut aggregation_complete = true; + let mut file_result = scan_claude_file_with_pricing( + path, + &cutoff, + &mut seen, + None, + &mut pricing, + |record| { + aggregation_complete &= record.timestamp.is_some(); + aggregation_complete &= + add_claude_record_to_daily_costs(&mut daily_costs, record); + }, + ); + if !aggregation_complete { + file_result.aggregation_failures = + file_result.aggregation_failures.saturating_add(1); + } + claude_scan.absorb(file_result); + }; + scanner.walk_claude_files(&projects_dir, &cutoff, None, &mut handle_file) }; - scanner.walk_claude_files(&projects_dir, &cutoff, None, &mut handle_file); + claude_scan.read_failures = claude_scan + .read_failures + .saturating_add(traversal_read_failures); if claude_scan.is_complete() { for slot in daily_costs.values_mut() { if slot.is_none() { @@ -1287,26 +1367,31 @@ pub fn get_daily_token_history(provider: &str, days: u32) -> (Vec<(String, u64)> } "claude" => { // Per-day token breakdown from the same de-duplicated record walk - // as the cost chart. The full walk is authoritative, so the - // Refreshing marker never applies here. + // as the cost chart. Only a complete valid walk establishes + // authoritative coverage of the requested history window. let projects_dir = scanner.get_claude_projects_dir(); if projects_dir.exists() { let cutoff = Utc::now() - Duration::days(days as i64); let mut seen = HashSet::new(); let mut pricing = ClaudeScanPricingResolver::default(); - let mut handle_file = |path: &Path| { - for_each_claude_usage_record_with_pricing( - path, - &cutoff, - &mut seen, - None, - &mut pricing, - |record| { - add_claude_record_to_daily_tokens(&mut daily_tokens, record); - }, - ); + let mut claude_scan = ClaudeFileScanResult::default(); + let traversal_read_failures = { + let mut handle_file = |path: &Path| { + let file_result = scan_claude_file_for_daily_tokens( + path, + &cutoff, + &mut seen, + &mut pricing, + &mut daily_tokens, + ); + claude_scan.absorb(file_result); + }; + scanner.walk_claude_files(&projects_dir, &cutoff, None, &mut handle_file) }; - scanner.walk_claude_files(&projects_dir, &cutoff, None, &mut handle_file); + claude_scan.read_failures = claude_scan + .read_failures + .saturating_add(traversal_read_failures); + mark_claude_daily_token_coverage(&mut covered_days, &daily_tokens, claude_scan); } } "pi" => { @@ -1327,14 +1412,12 @@ pub fn get_daily_token_history(provider: &str, days: u32) -> (Vec<(String, u64)> let mut result: Vec<(String, u64)> = daily_tokens.into_iter().collect(); result.sort_by(|a, b| a.0.cmp(&b.0)); - // Codex only: the bounded catch-up may not have reached the requested - // depth yet. Incomplete = history exists but the oldest quarter of the - // window has no scanned day. - let incomplete = if provider == "pi" { - // Pi scans are bounded filesystem walks, so a complete parse covers - // the requested window even when the roots contain no sessions. + let incomplete = if matches!(provider, "claude" | "pi") { + // A complete filesystem scan covers the requested window even when + // the roots contain no sessions; failed scans leave coverage empty. covered_days.is_empty() } else { + // Codex catch-up may not have reached the oldest quarter of the window. provider == "codex" && !covered_days.is_empty() && covered_days.len() < days as usize @@ -1349,9 +1432,9 @@ pub fn get_daily_token_history(provider: &str, days: u32) -> (Vec<(String, u64)> fn add_claude_record_to_daily_tokens( daily_tokens: &mut HashMap, record: &ClaudeUsageRecord, -) { +) -> bool { let Some(timestamp) = record.timestamp else { - return; + return true; }; let date_str = timestamp .with_timezone(&Local) @@ -1359,6 +1442,41 @@ fn add_claude_record_to_daily_tokens( .format("%Y-%m-%d") .to_string(); if let Some(slot) = daily_tokens.get_mut(&date_str) { - *slot += record.input + record.output; + let Some(tokens) = record.input.checked_add(record.output) else { + return false; + }; + return checked_add_assign(slot, tokens); + } + true +} + +fn scan_claude_file_for_daily_tokens( + path: &Path, + cutoff: &DateTime, + seen: &mut HashSet, + pricing: &mut ClaudeScanPricingResolver, + daily_tokens: &mut HashMap, +) -> ClaudeFileScanResult { + let mut aggregation_failures = 0u32; + let mut result = scan_claude_file_with_pricing(path, cutoff, seen, None, pricing, |record| { + if record.timestamp.is_none() || !add_claude_record_to_daily_tokens(daily_tokens, record) { + aggregation_failures = aggregation_failures.saturating_add(1); + } + }); + result.aggregation_failures = result + .aggregation_failures + .saturating_add(aggregation_failures); + result +} + +fn mark_claude_daily_token_coverage( + covered_days: &mut HashSet, + daily_tokens: &HashMap, + scan_result: ClaudeFileScanResult, +) { + if scan_result.is_complete() { + covered_days.extend(daily_tokens.keys().cloned()); + } else { + covered_days.clear(); } } diff --git a/rust/src/cost_scanner/claude_pricing.rs b/rust/src/cost_scanner/claude_pricing.rs index 89bcbe138a..5d3d251036 100644 --- a/rust/src/cost_scanner/claude_pricing.rs +++ b/rust/src/cost_scanner/claude_pricing.rs @@ -139,22 +139,16 @@ impl ClaudeScanPricingResolver { let cache_create_1h = cache_create_1h.min(cache_create); let cache_create_5m = cache_create.saturating_sub(cache_create_1h); - #[allow( - clippy::cast_possible_truncation, - reason = "clamped to i32::MAX before casting" - )] - let clamp = |value: u64| value.min(i32::MAX as u64) as i32; - let resolved = self.resolve(model); let billable = resolved.or_else(|| self.resolve(FALLBACK_CLAUDE_MODEL)); let base = billable .map(|pricing| { - CostUsagePricing::claude_cost_usd_from_resolution( + CostUsagePricing::claude_cost_usd_u64_from_resolution( pricing, - clamp(input), - clamp(cache_read), - clamp(cache_create_5m), - clamp(output), + input, + cache_read, + cache_create_5m, + output, ) }) .unwrap_or(0.0); diff --git a/rust/src/cost_scanner/tests.rs b/rust/src/cost_scanner/tests.rs index 3b16c92a1c..219aed2e48 100644 --- a/rust/src/cost_scanner/tests.rs +++ b/rust/src/cost_scanner/tests.rs @@ -395,7 +395,7 @@ fn counts_claude_usage_once_across_duplicate_records() { assert_eq!(record.output, 50); assert_eq!(record.cache_create, 10); assert_eq!(record.cache_read, 20); - assert!(record.cost > 0.0); + assert!(record.cost.is_some_and(|cost| cost > 0.0)); let cutoff = DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z") .unwrap() @@ -495,6 +495,157 @@ fn malformed_claude_history_stays_unknown_while_valid_empty_history_is_known_zer assert!(!malformed_summary.known_zero); } +#[test] +fn claude_daily_token_coverage_requires_a_complete_valid_scan() { + let root = tempfile::tempdir().unwrap(); + let cutoff = Utc::now() - Duration::days(1); + let valid_path = root.path().join("valid.jsonl"); + let timestamp = Utc::now() - Duration::hours(1); + let today = timestamp + .with_timezone(&Local) + .date_naive() + .format("%Y-%m-%d") + .to_string(); + std::fs::write( + &valid_path, + format!( + "{}\n", + claude_transcript_line( + ×tamp.to_rfc3339(), + "requestId", + "req_valid", + "msg_valid" + ) + ), + ) + .unwrap(); + + let mut valid_tokens = HashMap::from([(today.clone(), 0)]); + let valid_result = scan_claude_file_for_daily_tokens( + &valid_path, + &cutoff, + &mut HashSet::new(), + &mut ClaudeScanPricingResolver::default(), + &mut valid_tokens, + ); + assert!(valid_result.is_complete()); + let mut covered_days = HashSet::new(); + mark_claude_daily_token_coverage(&mut covered_days, &valid_tokens, valid_result); + assert!(covered_days.contains(&today)); + + let assert_uncovered = |path: &Path| { + let mut daily_tokens = HashMap::from([(today.clone(), 0)]); + let result = scan_claude_file_for_daily_tokens( + path, + &cutoff, + &mut HashSet::new(), + &mut ClaudeScanPricingResolver::default(), + &mut daily_tokens, + ); + assert!(!result.is_complete()); + let mut covered_days = HashSet::from(["stale-coverage".to_string()]); + mark_claude_daily_token_coverage(&mut covered_days, &daily_tokens, result); + assert!(covered_days.is_empty()); + result + }; + + let malformed_path = root.path().join("malformed.jsonl"); + std::fs::write(&malformed_path, b"{malformed\n").unwrap(); + assert_eq!(assert_uncovered(&malformed_path).malformed_lines, 1); + + let incomplete_path = root.path().join("incomplete.jsonl"); + std::fs::write( + &incomplete_path, + r#"{"type":"assistant","message":{"id":"msg_preliminary","model":"gpt-5.6-sol","stop_reason":null,"usage":{"input_tokens":1000}}}"#, + ) + .unwrap(); + assert_eq!(assert_uncovered(&incomplete_path).incomplete_requests, 1); + + let missing_timestamp_path = root.path().join("missing-timestamp.jsonl"); + std::fs::write( + &missing_timestamp_path, + r#"{"type":"assistant","requestId":"req_no_timestamp","message":{"id":"msg_no_timestamp","model":"claude-sonnet-4-6","usage":{"input_tokens":1000,"output_tokens":500}}}"#, + ) + .unwrap(); + assert_eq!( + assert_uncovered(&missing_timestamp_path).aggregation_failures, + 1 + ); + + let unreadable_path = root.path().join("missing.jsonl"); + assert_eq!(assert_uncovered(&unreadable_path).read_failures, 1); + + let scanner = CostScanner::new(1); + let missing_directory = root.path().join("missing-directory"); + let traversal_read_failures = + scanner.walk_claude_files(&missing_directory, &cutoff, None, &mut |_| {}); + assert_eq!(traversal_read_failures, 1); + let mut covered_days = HashSet::from([today]); + mark_claude_daily_token_coverage( + &mut covered_days, + &valid_tokens, + ClaudeFileScanResult { + read_failures: traversal_read_failures, + ..ClaudeFileScanResult::default() + }, + ); + assert!(covered_days.is_empty()); +} + +#[test] +fn public_claude_daily_token_dispatch_reports_incomplete_fixture_scans() { + const CHILD_MARKER: &str = "CODEXBAR_CLAUDE_DAILY_TOKEN_TEST_CHILD"; + const CHILD_DONE: &str = "isolated Claude daily-history fixture verified"; + if std::env::var_os(CHILD_MARKER).is_some() { + let config_dir = std::env::var_os("CLAUDE_CONFIG_DIR") + .map(PathBuf::from) + .expect("child receives isolated Claude config directory"); + let projects_dir = config_dir.join("projects"); + let project_dir = projects_dir.join("fixture-project"); + let (complete_history, incomplete) = get_daily_token_history("claude", 1); + assert!(!incomplete, "valid fixture scan should establish coverage"); + assert!(complete_history.iter().any(|(_, tokens)| *tokens > 0)); + + std::fs::write(project_dir.join("malformed.jsonl"), b"{malformed\n").unwrap(); + let (partial_history, incomplete) = get_daily_token_history("claude", 1); + assert!( + incomplete, + "malformed fixture should leave coverage incomplete" + ); + assert_eq!(partial_history, complete_history); + println!("{CHILD_DONE}"); + return; + } + + let config_dir = tempfile::tempdir().unwrap(); + let project_dir = config_dir.path().join("projects").join("fixture-project"); + std::fs::create_dir_all(&project_dir).unwrap(); + let timestamp = Utc::now().to_rfc3339(); + std::fs::write( + project_dir.join("valid.jsonl"), + format!( + "{}\n", + claude_transcript_line(×tamp, "requestId", "req_public", "msg_public") + ), + ) + .unwrap(); + + let test_thread = std::thread::current(); + let test_name = test_thread.name().expect("test harness names this thread"); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", test_name, "--nocapture", "--test-threads=1"]) + .env(CHILD_MARKER, "1") + .env("CLAUDE_CONFIG_DIR", config_dir.path()) + .output() + .expect("spawn isolated exact-test child"); + assert!( + output.status.success() && String::from_utf8_lossy(&output.stdout).contains(CHILD_DONE), + "fixture child failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + #[test] fn classifies_vertex_ai_claude_metadata_without_changing_anthropic_rows() { let cases = [ @@ -552,9 +703,17 @@ fn shared_claude_reader_excludes_vertex_rows_but_keeps_anthropic_usage() { let anthropic = format!( r#"{{"type":"assistant","timestamp":"{timestamp}","requestId":"req_anthropic","message":{{"id":"msg_anthropic","model":"claude-sonnet-4-6","usage":{{"input_tokens":10,"output_tokens":5}}}}}}"# ); - let vertex = format!( - r#"{{"type":"assistant","timestamp":"{timestamp}","requestId":"req_vrtx_123","message":{{"id":"msg_vrtx_123","model":"claude-sonnet-4-6","usage":{{"input_tokens":1000,"output_tokens":500}}}}}}"# - ); + let vertex = serde_json::json!({ + "type": "assistant", + "timestamp": timestamp, + "requestId": "req_vrtx_123", + "message": { + "id": "msg_vrtx_123", + "model": "claude-sonnet-4-6", + "usage": {"input_tokens": u64::MAX, "output_tokens": u64::MAX} + } + }) + .to_string(); std::fs::write(&path, format!("{anthropic}\n{vertex}\n")).unwrap(); let cutoff = Utc::now() - Duration::days(30); @@ -569,6 +728,86 @@ fn shared_claude_reader_excludes_vertex_rows_but_keeps_anthropic_usage() { let _removed = std::fs::remove_file(&path); } +#[test] +fn oversized_claude_history_preserves_independent_components_and_fails_closed() { + let first: ClaudeEvent = serde_json::from_str(&format!( + r#"{{"type":"assistant","timestamp":"2026-09-20T12:00:00Z","requestId":"req_overflow_1","message":{{"id":"msg_overflow_1","model":"claude-sonnet-4-6","usage":{{"input_tokens":{},"output_tokens":2}}}}}}"#, + u64::MAX + )) + .unwrap(); + let second: ClaudeEvent = serde_json::from_str( + r#"{"type":"assistant","timestamp":"2026-09-20T12:01:00Z","requestId":"req_overflow_2","message":{"id":"msg_overflow_2","model":"claude-sonnet-4-6","usage":{"input_tokens":1,"output_tokens":3}}}"#, + ) + .unwrap(); + let first = claude_usage_record_from_event(&first).expect("first usage row"); + let second = claude_usage_record_from_event(&second).expect("second usage row"); + let mut summary = CostSummary::default(); + + assert!(add_claude_record_to_summary(&mut summary, &first)); + assert!(!add_claude_record_to_summary(&mut summary, &second)); + assert_eq!(summary.input_tokens, u64::MAX); + assert_eq!(summary.output_tokens, 5); + assert!(summary.total_cost_usd.is_finite()); + + finalize_claude_summary( + &mut summary, + true, + ClaudeFileScanResult { + counted: 2, + aggregation_failures: 1, + ..ClaudeFileScanResult::default() + }, + false, + ); + assert!(!summary.history_coverage_established); + assert!(!summary.known_zero); +} + +#[test] +fn oversized_single_claude_row_keeps_cost_but_marks_combined_quota_tokens_unknown() { + let event: ClaudeEvent = serde_json::from_str(&format!( + r#"{{"type":"assistant","timestamp":"2026-09-20T12:00:00Z","requestId":"req_combined_overflow","message":{{"id":"msg_combined_overflow","model":"claude-sonnet-4-6","usage":{{"input_tokens":{},"output_tokens":1}}}}}}"#, + u64::MAX + )) + .unwrap(); + let record = claude_usage_record_from_event(&event).expect("usage row"); + let quota = quota_history_record_from_usage(&record).expect("timestamped quota row"); + + assert!(record.cost.is_some_and(f64::is_finite)); + assert_eq!(quota.tokens, None); + assert!(!quota.tokens_are_complete); + assert!(quota.cost_usd.is_some_and(f64::is_finite)); + assert!(quota.cost_is_complete); +} + +#[test] +fn nonfinite_claude_price_is_unknown_instead_of_zero() { + let snapshot = crate::core::ModelsDevPricingSnapshot::from_catalog_json_for_tests( + r#"{ + "anthropic": {"models": {"claude-test-extreme-price": { + "id": "claude-test-extreme-price", "cost": {"input": 1e308, "output": 1} + }}} + }"#, + ) + .expect("pricing fixture"); + let mut pricing = ClaudeScanPricingResolver::with_snapshot(snapshot); + let event: ClaudeEvent = serde_json::from_str(&format!( + r#"{{"type":"assistant","timestamp":"2026-09-20T12:00:00Z","requestId":"req_nonfinite","message":{{"id":"msg_nonfinite","model":"claude-test-extreme-price","usage":{{"input_tokens":{},"output_tokens":1}}}}}}"#, + u64::MAX + )) + .unwrap(); + let record = + claude_usage_record_from_event_with_pricing(&event, &mut pricing).expect("usage row"); + let mut summary = CostSummary::default(); + + assert_eq!(record.cost, None); + assert!(!add_claude_record_to_summary(&mut summary, &record)); + assert_eq!(summary.input_tokens, u64::MAX); + assert_eq!(summary.output_tokens, 1); + assert_eq!(summary.total_cost_usd, 0.0); + assert!(!summary.known_zero); +} + fn claude_transcript_line( timestamp: &str, request_key: &str,