From c68cfd7d3e43efcb7dd906de782e485c394a8ae8 Mon Sep 17 00:00:00 2001 From: NessZerra Date: Fri, 17 Jul 2026 12:42:07 +0700 Subject: [PATCH] fix: stale provider publish and cost scan freshness - Add refresh generation so superseded in-flight provider results are ignored - Prune disabled providers from cache when enablement changes - App-driven CostScanOptions bypasses scanner debounce (upstream #2089) - Model-less Codex events stay unpriced/unattributed instead of gpt-5 - Cap JSONL lines at 256 KiB and contain Ultra interleaved totals --- .../src-tauri/src/commands/providers.rs | 118 ++++- .../src-tauri/src/commands/settings.rs | 13 +- .../src-tauri/src/commands/tests.rs | 30 ++ apps/desktop-tauri/src-tauri/src/state.rs | 4 + rust/src/codex_costs.rs | 69 ++- rust/src/core/cost_pricing.rs | 22 + rust/src/core/cost_pricing_tests.rs | 18 + rust/src/core/jsonl_scanner.rs | 466 ++++++++++++++++-- 8 files changed, 668 insertions(+), 72 deletions(-) diff --git a/apps/desktop-tauri/src-tauri/src/commands/providers.rs b/apps/desktop-tauri/src-tauri/src/commands/providers.rs index 2ae9b55846..c6dcb69052 100644 --- a/apps/desktop-tauri/src-tauri/src/commands/providers.rs +++ b/apps/desktop-tauri/src-tauri/src/commands/providers.rs @@ -150,6 +150,37 @@ pub(crate) fn upsert_provider_cache( } } +/// Drop cached snapshots for providers that are no longer enabled. +pub(crate) fn prune_provider_cache_to_enabled( + cache: &mut Vec, + enabled_ids: &[ProviderId], +) { + cache.retain(|snapshot| { + enabled_ids + .iter() + .any(|id| id.cli_name() == snapshot.provider_id) + }); +} + +/// Invalidate in-flight publish work and remove disabled providers from cache. +pub(crate) fn invalidate_provider_refresh_and_prune_disabled( + state: &tauri::State<'_, Mutex>, + enabled_ids: &[ProviderId], +) -> Result<(), String> { + let mut guard = state.lock().map_err(|e| e.to_string())?; + guard.provider_refresh_generation = guard.provider_refresh_generation.wrapping_add(1); + prune_provider_cache_to_enabled(&mut guard.provider_cache, enabled_ids); + // Drop transient-failure counters for providers that left the enabled set. + guard + .transient_provider_failure_counts + .retain(|id, _| enabled_ids.contains(id)); + Ok(()) +} + +pub(crate) fn is_current_provider_refresh_generation(guard: &AppState, generation: u64) -> bool { + guard.provider_refresh_generation == generation +} + /// Core refresh logic, usable from both the Tauri command and tray menu actions. pub(crate) async fn do_refresh_providers(app: &tauri::AppHandle) -> Result<(), String> { do_refresh_providers_with_policy(app, true).await @@ -165,11 +196,18 @@ async fn do_refresh_providers_with_policy( ) -> Result<(), String> { let state = app.state::>(); - if !begin_provider_refresh(&state, force)? { + let Some(generation) = begin_provider_refresh(&state, force)? else { return Ok(()); - } + }; let inputs = ProviderRefreshInputs::load(); + // Ensure cache only contains currently enabled providers for this generation. + if let Ok(mut guard) = state.lock() + && is_current_provider_refresh_generation(&guard, generation) + { + prune_provider_cache_to_enabled(&mut guard.provider_cache, &inputs.enabled_ids); + } + events::emit_refresh_started( app, inputs @@ -180,10 +218,10 @@ async fn do_refresh_providers_with_policy( ); let enabled_count = inputs.enabled_ids.len(); - let handles = spawn_provider_refreshes(app, &inputs); + let handles = spawn_provider_refreshes(app, &inputs, generation); await_provider_refreshes(handles).await; - let error_count = finish_provider_refresh(&state)?; + let error_count = finish_provider_refresh(&state, generation)?; update_tray_and_notifications(app, &state, &inputs.settings, &inputs.token_accounts)?; events::emit_refresh_complete(app, enabled_count, error_count); @@ -195,18 +233,20 @@ async fn do_refresh_providers_with_policy( fn begin_provider_refresh( state: &tauri::State<'_, Mutex>, force: bool, -) -> Result { +) -> Result, String> { let mut guard = state.lock().map_err(|e| e.to_string())?; if guard.is_refreshing { - return Ok(false); + return Ok(None); } if provider_cache_can_skip_refresh(&guard, force) { - return Ok(false); + return Ok(None); } + guard.provider_refresh_generation = guard.provider_refresh_generation.wrapping_add(1); + let generation = guard.provider_refresh_generation; guard.is_refreshing = true; guard.provider_refresh_started_at = Some(std::time::Instant::now()); - Ok(true) + Ok(Some(generation)) } fn provider_cache_can_skip_refresh(guard: &AppState, force: bool) -> bool { @@ -247,6 +287,7 @@ impl ProviderRefreshInputs { fn spawn_provider_refreshes( app: &tauri::AppHandle, inputs: &ProviderRefreshInputs, + generation: u64, ) -> Vec> { let mut handles = Vec::with_capacity(inputs.enabled_ids.len()); let fetch_permits = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_PROVIDER_FETCHES)); @@ -267,25 +308,43 @@ fn spawn_provider_refreshes( let Ok(_permit) = fetch_permits.acquire_owned().await else { return; }; - refresh_provider(app_handle, id, ctx).await; + refresh_provider(app_handle, id, ctx, generation).await; })); } handles } -async fn refresh_provider(app: tauri::AppHandle, id: ProviderId, ctx: FetchContext) { +async fn refresh_provider( + app: tauri::AppHandle, + id: ProviderId, + ctx: FetchContext, + generation: u64, +) { let snapshot = fetch_provider_snapshot(id, ctx).await; let state = app.state::>(); - let snapshot = if let Ok(mut guard) = state.lock() { - let snapshot = preserve_last_good_transient_failure(&mut guard, id, snapshot); - upsert_provider_cache(&mut guard.provider_cache, snapshot.clone()); - snapshot + let published = if let Ok(mut guard) = state.lock() { + if !is_current_provider_refresh_generation(&guard, generation) { + tracing::debug!( + provider = id.cli_name(), + generation, + current = guard.provider_refresh_generation, + "dropping superseded provider refresh result" + ); + None + } else { + let snapshot = preserve_last_good_transient_failure(&mut guard, id, snapshot); + upsert_provider_cache(&mut guard.provider_cache, snapshot.clone()); + Some(snapshot) + } } else { - snapshot + None }; - events::emit_provider_updated(&app, &snapshot); + + if let Some(snapshot) = published { + events::emit_provider_updated(&app, &snapshot); + } } pub(super) fn preserve_last_good_transient_failure( @@ -384,11 +443,30 @@ async fn await_provider_refreshes(handles: Vec>) { } } -fn finish_provider_refresh(state: &tauri::State<'_, Mutex>) -> Result { +fn finish_provider_refresh( + state: &tauri::State<'_, Mutex>, + generation: u64, +) -> Result { let mut guard = state.lock().map_err(|e| e.to_string())?; - guard.is_refreshing = false; - guard.provider_cache_updated_at = Some(std::time::Instant::now()); - guard.provider_refresh_started_at = None; + // Always release the refresh lock when this batch ends so a later + // invalidate (enablement change) cannot leave is_refreshing stuck. + if guard.is_refreshing + && (is_current_provider_refresh_generation(&guard, generation) + || guard.provider_refresh_started_at.is_some()) + { + // Only clear the lock if we still own it or no newer refresh has begun + // (newer begin bumps generation AND sets is_refreshing under the same + // ownership — currently concurrent begin is blocked). + if is_current_provider_refresh_generation(&guard, generation) { + guard.is_refreshing = false; + guard.provider_refresh_started_at = None; + guard.provider_cache_updated_at = Some(std::time::Instant::now()); + } else { + // Superseded: release lock without stamping a "fresh" cache time. + guard.is_refreshing = false; + guard.provider_refresh_started_at = None; + } + } Ok(guard .provider_cache .iter() diff --git a/apps/desktop-tauri/src-tauri/src/commands/settings.rs b/apps/desktop-tauri/src-tauri/src/commands/settings.rs index ed294d74eb..921fb26480 100644 --- a/apps/desktop-tauri/src-tauri/src/commands/settings.rs +++ b/apps/desktop-tauri/src-tauri/src/commands/settings.rs @@ -374,6 +374,15 @@ pub async fn update_settings( crate::commands::clear_provider_local_usage_cache(); } + if refresh_provider_data { + // Invalidate any in-flight publish work and drop disabled providers from + // the live cache before a follow-up refresh starts. + let enabled_ids = settings.get_enabled_provider_ids(); + let state = app.state::>(); + let _ = + crate::commands::invalidate_provider_refresh_and_prune_disabled(&state, &enabled_ids); + } + crate::floatbar::after_settings_saved(&app, &float_bar_patch, &settings, notify_float_bar); if rebuild_tray_menu { crate::tray_bridge::rebuild_tray_menu(&app); @@ -383,7 +392,9 @@ pub async fn update_settings( } if tray_promotion_changed { let new_promoted = settings.promote_tray_icon; - if new_promoted || crate::tray_visibility::should_write_demotion(previous_promoted, new_promoted) { + if new_promoted + || crate::tray_visibility::should_write_demotion(previous_promoted, new_promoted) + { crate::tray_visibility::apply_promotion(new_promoted); } } diff --git a/apps/desktop-tauri/src-tauri/src/commands/tests.rs b/apps/desktop-tauri/src-tauri/src/commands/tests.rs index 32112e7521..2c9f25eb89 100644 --- a/apps/desktop-tauri/src-tauri/src/commands/tests.rs +++ b/apps/desktop-tauri/src-tauri/src/commands/tests.rs @@ -4,6 +4,7 @@ use super::{ NamedRateWindowSnapshot, ProviderSummary, ProviderUsageSnapshot, provider_cookie_source_lookup, provider_region_lookup, validate_external_url, validate_surface_target, }; +use crate::state::AppState; use crate::surface::SurfaceMode; use crate::surface_target::SurfaceTarget; use codexbar::core::{ @@ -696,6 +697,35 @@ fn provider_cache_upsert_replaces_existing_provider() { assert_eq!(cache[0].error.as_deref(), Some("new")); } +#[test] +fn provider_cache_prunes_disabled_providers() { + let metadata = instantiate_provider(ProviderId::Codex).metadata().clone(); + let result = ProviderFetchResult { + usage: codexbar::core::UsageSnapshot::new(codexbar::core::RateWindow::new(10.0)), + cost: None, + wayfinder_usage: None, + source_label: "CLI".to_string(), + }; + let codex = ProviderUsageSnapshot::from_fetch_result(ProviderId::Codex, &metadata, &result); + let claude_meta = instantiate_provider(ProviderId::Claude).metadata().clone(); + let claude = + ProviderUsageSnapshot::from_fetch_result(ProviderId::Claude, &claude_meta, &result); + + let mut cache = vec![codex, claude]; + super::prune_provider_cache_to_enabled(&mut cache, &[ProviderId::Codex]); + + assert_eq!(cache.len(), 1); + assert_eq!(cache[0].provider_id, "codex"); +} + +#[test] +fn superseded_refresh_generation_is_not_current() { + let mut state = AppState::new(); + state.provider_refresh_generation = 3; + assert!(super::is_current_provider_refresh_generation(&state, 3)); + assert!(!super::is_current_provider_refresh_generation(&state, 2)); +} + #[test] fn hiding_codex_spark_rows_preserves_other_extra_usage() { let metadata = instantiate_provider(ProviderId::Codex).metadata().clone(); diff --git a/apps/desktop-tauri/src-tauri/src/state.rs b/apps/desktop-tauri/src-tauri/src/state.rs index e96a5c4adf..d74cc21a71 100644 --- a/apps/desktop-tauri/src-tauri/src/state.rs +++ b/apps/desktop-tauri/src-tauri/src/state.rs @@ -123,6 +123,9 @@ pub struct AppState { pub transient_provider_failure_counts: HashMap, pub provider_cache_updated_at: Option, pub provider_refresh_started_at: Option, + /// Monotonic generation for in-flight provider fetches. Bumped when a new + /// refresh starts or enablement changes so superseded results are ignored. + pub provider_refresh_generation: u64, pub is_refreshing: bool, pub update_state: UpdateState, /// Full update metadata from the last successful check. @@ -185,6 +188,7 @@ impl AppState { transient_provider_failure_counts: HashMap::new(), provider_cache_updated_at: None, provider_refresh_started_at: None, + provider_refresh_generation: 0, is_refreshing: false, update_state: UpdateState::Idle, update_info: None, diff --git a/rust/src/codex_costs.rs b/rust/src/codex_costs.rs index 0888165c91..aa8f697b9c 100644 --- a/rust/src/codex_costs.rs +++ b/rust/src/codex_costs.rs @@ -101,29 +101,46 @@ fn add_codex_tokens_to_summary( return None; } + let model_key = if CostUsagePricing::is_codex_unattributed_model(model) { + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL.to_string() + } else { + model.to_string() + }; + + // Unattributed usage is visible but never priced and must not trigger a + // models.dev catalog refresh (it is deliberately unpriced, not "unknown yet"). + if CostUsagePricing::is_codex_unattributed_model(&model_key) { + summary.input_tokens += tokens.input; + summary.cached_tokens += tokens.cached; + summary.output_tokens += tokens.output; + summary.by_model.entry(model_key.clone()).or_insert(0.0); + add_tokens( + summary.by_model_tokens.entry(model_key).or_default(), + tokens, + ); + return Some(0.0); + } + let uses_fallback_pricing = - CostUsagePricing::codex_cost_usd(model, tokens.input, tokens.cached, tokens.output) + CostUsagePricing::codex_cost_usd(&model_key, tokens.input, tokens.cached, tokens.output) .is_none(); - let cost = codex_cost_usd(model, tokens.input, tokens.cached, tokens.output); + let cost = codex_cost_usd(&model_key, tokens.input, tokens.cached, tokens.output); if uses_fallback_pricing { - summary.unknown_models.insert(model.to_string()); + summary.unknown_models.insert(model_key.clone()); } summary.input_tokens += tokens.input; summary.cached_tokens += tokens.cached; summary.output_tokens += tokens.output; - *summary.by_model.entry(model.to_string()).or_insert(0.0) += cost; + *summary.by_model.entry(model_key.clone()).or_insert(0.0) += cost; - let speed_bucket = codex_speed_bucket(model); + let speed_bucket = codex_speed_bucket(&model_key); *summary .by_speed .entry(speed_bucket.to_string()) .or_insert(0.0) += cost; add_tokens( - summary - .by_model_tokens - .entry(model.to_string()) - .or_default(), + summary.by_model_tokens.entry(model_key).or_default(), tokens, ); add_tokens( @@ -142,6 +159,9 @@ fn codex_records_cost(records: &[CodexUsageRecord], range: &CostUsageDayRange) - for record in records.iter().filter(|record| { CostUsageDayRange::is_in_range(&record.day_key, &range.since_key, &range.until_key) }) { + if CostUsagePricing::is_codex_unattributed_model(&record.model) { + continue; + } let tokens = CodexTokenCounts::from_values(record.input, record.cached, record.output); if !tokens.is_empty() { total_cost += codex_cost_usd(&record.model, tokens.input, tokens.cached, tokens.output); @@ -165,6 +185,9 @@ fn codex_speed_bucket(model: &str) -> &'static str { } fn codex_cost_usd(model: &str, input: u64, cached: u64, output: u64) -> f64 { + if CostUsagePricing::is_codex_unattributed_model(model) { + return 0.0; + } if let Some(cost) = CostUsagePricing::codex_cost_usd(model, input, cached, output) { return cost; } @@ -246,6 +269,34 @@ mod tests { assert!((cost - 2.0).abs() < f64::EPSILON); } + #[test] + fn model_less_codex_usage_is_visible_but_unpriced() { + let target = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let range = CostUsageDayRange::new(target, target); + let records = vec![CodexUsageRecord { + day_key: "2026-05-31".to_string(), + model: CostUsagePricing::CODEX_UNATTRIBUTED_MODEL.to_string(), + input: 55_000_000, + cached: 0, + output: 0, + }]; + let mut summary = CostSummary::default(); + + let (cost, has_tokens) = add_codex_records_to_summary(&mut summary, &records, &range); + + assert!(has_tokens); + assert_eq!(cost, 0.0); + assert_eq!(summary.input_tokens, 55_000_000); + assert_eq!( + summary + .by_model + .get(CostUsagePricing::CODEX_UNATTRIBUTED_MODEL) + .copied(), + Some(0.0) + ); + assert!(summary.unknown_models.is_empty()); + } + #[test] fn records_unknown_codex_model_while_using_fallback_cost() { let target = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); diff --git a/rust/src/core/cost_pricing.rs b/rust/src/core/cost_pricing.rs index 0c70f64a06..01c828e05b 100755 --- a/rust/src/core/cost_pricing.rs +++ b/rust/src/core/cost_pricing.rs @@ -602,9 +602,26 @@ fn codex_cost_from_rates( pub struct CostUsagePricing; impl CostUsagePricing { + /// Sentinel model key for model-less Codex token events. + /// + /// Usage remains visible under this key but is never priced as a real model + /// (including catalog collisions with a generic "unknown" entry). + pub const CODEX_UNATTRIBUTED_MODEL: &'static str = "unknown"; + + /// True when `model` is the unattributed / model-less sentinel. + pub fn is_codex_unattributed_model(model: &str) -> bool { + Self::normalize_codex_model(model) == Self::CODEX_UNATTRIBUTED_MODEL + } + /// Normalize a Codex model name for pricing lookup pub fn normalize_codex_model(raw: &str) -> String { let mut trimmed = raw.trim().to_string(); + if trimmed.is_empty() + || trimmed.eq_ignore_ascii_case("unknown") + || trimmed.eq_ignore_ascii_case("unpriced") + { + return Self::CODEX_UNATTRIBUTED_MODEL.to_string(); + } // Remove "openai/" prefix if let Some(rest) = trimmed.strip_prefix("openai/") { @@ -685,6 +702,11 @@ impl CostUsagePricing { output_tokens: u64, ) -> Option { let key = Self::normalize_codex_model(model); + // Model-less / deliberately unattributed usage stays unpriced even if a + // pricing catalog later contains a colliding generic entry. + if key == Self::CODEX_UNATTRIBUTED_MODEL { + return None; + } if let Some(pricing) = CODEX_PRICING.get(key.as_str()) { let (input_rate, cache_read_rate, output_rate) = if input_tokens > CODEX_LONG_CONTEXT_THRESHOLD { diff --git a/rust/src/core/cost_pricing_tests.rs b/rust/src/core/cost_pricing_tests.rs index f0552a5505..dc1220ba2d 100644 --- a/rust/src/core/cost_pricing_tests.rs +++ b/rust/src/core/cost_pricing_tests.rs @@ -11,6 +11,24 @@ fn test_normalize_codex_model() { CostUsagePricing::normalize_codex_model("gpt-5-codex"), "gpt-5" ); + assert_eq!( + CostUsagePricing::normalize_codex_model(""), + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL + ); + assert_eq!( + CostUsagePricing::normalize_codex_model("unknown"), + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL + ); +} + +#[test] +fn unattributed_codex_usage_stays_unpriced() { + assert!( + CostUsagePricing::codex_cost_usd(CostUsagePricing::CODEX_UNATTRIBUTED_MODEL, 1_000, 0, 500) + .is_none() + ); + assert!(CostUsagePricing::is_codex_unattributed_model("unknown")); + assert!(CostUsagePricing::is_codex_unattributed_model(" ")); } #[test] diff --git a/rust/src/core/jsonl_scanner.rs b/rust/src/core/jsonl_scanner.rs index 31daff28c4..ad560974ea 100755 --- a/rust/src/core/jsonl_scanner.rs +++ b/rust/src/core/jsonl_scanner.rs @@ -14,6 +14,50 @@ use std::fs::{self, File}; use std::io::{BufRead, BufReader, Seek, SeekFrom}; use std::path::{Path, PathBuf}; +/// Maximum retained Codex JSONL line size (upstream session-metadata bound). +const CODEX_JSONL_MAX_LINE_BYTES: usize = 256 * 1024; + +/// Default scanner-side refresh debounce (upstream CostUsageScanner). +pub const DEFAULT_COST_SCAN_REFRESH_MIN_INTERVAL_SECS: u64 = 60; + +/// Options for a cost scan pass. +/// +/// App-driven callers (Tauri refresh / menu open) already own their own TTL, so +/// they use [`CostScanOptions::app_driven`] to bypass the scanner debounce and +/// avoid pinning a stale disk-cache snapshot for the full token-cost TTL +/// (upstream issue #2089). +#[derive(Debug, Clone, Copy)] +pub struct CostScanOptions { + /// Minimum seconds between disk-cache-backed full inspections. + /// Set to 0 to force a fresh scan (app-driven / forceRefresh). + pub refresh_min_interval_secs: u64, +} + +impl Default for CostScanOptions { + fn default() -> Self { + Self { + refresh_min_interval_secs: DEFAULT_COST_SCAN_REFRESH_MIN_INTERVAL_SECS, + } + } +} + +impl CostScanOptions { + /// App-driven or forced refresh: skip the scanner debounce entirely. + pub fn app_driven() -> Self { + Self { + refresh_min_interval_secs: 0, + } + } + + /// Whether a prior scan at `last_scan_unix_ms` is still within the debounce window. + pub fn should_skip_scan(&self, last_scan_unix_ms: i64, now_unix_ms: i64) -> bool { + let refresh_ms = (self.refresh_min_interval_secs as i64).saturating_mul(1000); + refresh_ms > 0 + && last_scan_unix_ms > 0 + && now_unix_ms.saturating_sub(last_scan_unix_ms) <= refresh_ms + } +} + /// Cache for scanned file data #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct CostUsageCache { @@ -113,6 +157,11 @@ pub struct JsonlScanner; struct CodexParserState { current_model: Option, previous_totals: Option, + /// High watermark of observed cumulative totals (never lowered). Used for + /// Ultra interleaved-lineage containment (issue #2037 Phase 1). + totals_watermark: Option, + /// Latched once any cumulative component drops below the watermark. + saw_interleaved_totals: bool, records: Vec, } @@ -188,7 +237,9 @@ impl CodexParserState { fn new(initial_model: Option, initial_totals: Option) -> Self { Self { current_model: initial_model, - previous_totals: initial_totals, + previous_totals: initial_totals.clone(), + totals_watermark: initial_totals, + saw_interleaved_totals: false, records: Vec::new(), } } @@ -222,8 +273,9 @@ impl CodexParserState { fn process_fast_event(&mut self, event: CodexFastEvent<'_>, range: &CostUsageDayRange) { match event { CodexFastEvent::TurnContext { model } => { - if let Some(model) = model.filter(|model| !model.is_empty()) { - self.current_model = Some(model.to_string()); + // Explicit blank model evidence clears stale turn context. + if let Some(raw) = model { + self.current_model = model_evidence(raw).map(str::to_string); } } CodexFastEvent::TokenCount { timestamp, payload } => { @@ -243,18 +295,34 @@ impl CodexParserState { } fn update_current_model(&mut self, obj: &Value) { - if let Some(model) = obj - .get("model") - .or_else(|| obj.get("payload").and_then(|payload| payload.get("model"))) - .or_else(|| { - obj.get("payload") - .and_then(|payload| payload.get("info")) - .and_then(|info| info.get("model")) - }) - .and_then(|v| v.as_str()) - { - self.current_model = Some(model.to_string()); + let candidates = [ + obj.get("model").and_then(|v| v.as_str()), + obj.get("payload") + .and_then(|payload| payload.get("model")) + .and_then(|v| v.as_str()), + obj.get("payload") + .and_then(|payload| payload.get("model_name")) + .and_then(|v| v.as_str()), + obj.get("payload") + .and_then(|payload| payload.get("info")) + .and_then(|info| info.get("model")) + .and_then(|v| v.as_str()), + obj.get("payload") + .and_then(|payload| payload.get("info")) + .and_then(|info| info.get("model_name")) + .and_then(|v| v.as_str()), + ]; + // Only rewrite current_model when the turn_context actually carries a + // model field (including blank, which clears stale attribution). + let has_key = candidates.iter().any(|c| c.is_some()); + if !has_key { + return; } + self.current_model = candidates + .into_iter() + .flatten() + .find_map(model_evidence) + .map(str::to_string); } fn record_token_count(&mut self, obj: &Value, day_key: String) { @@ -269,7 +337,7 @@ impl CodexParserState { } let info = payload.get("info"); - let model = self.token_model(info, payload, obj); + let model = self.resolve_token_model(info, payload, obj); self.record_usage(day_key, &model, delta_input, delta_cached, delta_output); } @@ -282,13 +350,20 @@ impl CodexParserState { return; } - let model = payload + let event_model = payload .info .as_ref() .and_then(|info| info.model.or(info.model_name)) .or(payload.model) - .or(self.current_model.as_deref()) - .unwrap_or("gpt-5") + .and_then(model_evidence); + // Prefer current turn_context model over a conflicting event model, + // matching upstream precedence. Fall back to unattributed (not gpt-5). + let model = self + .current_model + .as_deref() + .and_then(model_evidence) + .or(event_model) + .unwrap_or(CostUsagePricing::CODEX_UNATTRIBUTED_MODEL) .to_string(); self.record_usage(day_key, &model, delta_input, delta_cached, delta_output); } @@ -303,14 +378,19 @@ impl CodexParserState { }); } - fn token_model(&self, info: Option<&Value>, payload: &Value, obj: &Value) -> String { - info.and_then(|i| i.get("model").or(i.get("model_name"))) + fn resolve_token_model(&self, info: Option<&Value>, payload: &Value, obj: &Value) -> String { + let event_model = info + .and_then(|i| i.get("model").or(i.get("model_name"))) .or_else(|| payload.get("model")) .or_else(|| obj.get("model")) .and_then(|v| v.as_str()) - .map(str::to_string) - .or_else(|| self.current_model.clone()) - .unwrap_or_else(|| "gpt-5".to_string()) + .and_then(model_evidence); + self.current_model + .as_deref() + .and_then(model_evidence) + .or(event_model) + .unwrap_or(CostUsagePricing::CODEX_UNATTRIBUTED_MODEL) + .to_string() } fn token_deltas(&mut self, payload: &Value) -> Option<(i32, i32, i32)> { @@ -354,24 +434,141 @@ impl CodexParserState { fn total_usage_delta(&mut self, total: &Value) -> (i32, i32, i32) { let totals = read_token_totals(total); - let previous = self.previous_totals.as_ref(); - let delta_input = (totals.input - previous.map_or(0, |t| t.input)).max(0); - let delta_cached = (totals.cached - previous.map_or(0, |t| t.cached)).max(0); - let delta_output = (totals.output - previous.map_or(0, |t| t.output)).max(0); - - self.previous_totals = Some(totals); - (delta_input, delta_cached, delta_output) + self.apply_totals_delta(totals) } fn fast_total_usage_delta(&mut self, total: CodexFastTotals) -> (i32, i32, i32) { let totals = codex_totals_from_fast(total); - let previous = self.previous_totals.as_ref(); - let delta_input = (totals.input - previous.map_or(0, |t| t.input)).max(0); - let delta_cached = (totals.cached - previous.map_or(0, |t| t.cached)).max(0); - let delta_output = (totals.output - previous.map_or(0, |t| t.output)).max(0); + self.apply_totals_delta(totals) + } + + fn apply_totals_delta(&mut self, totals: CodexTotals) -> (i32, i32, i32) { + self.latch_if_below_watermark(&totals); + + let delta = if self.saw_interleaved_totals { + contained_total_delta( + self.totals_watermark.as_ref(), + self.previous_totals.as_ref(), + &totals, + ) + } else { + let previous = self.previous_totals.as_ref(); + CodexTotals { + input: (totals.input - previous.map_or(0, |t| t.input)).max(0), + cached: (totals.cached - previous.map_or(0, |t| t.cached)).max(0), + output: (totals.output - previous.map_or(0, |t| t.output)).max(0), + } + }; + + self.previous_totals = Some(totals.clone()); + self.raise_watermark(&totals); + (delta.input, delta.cached, delta.output) + } + + fn latch_if_below_watermark(&mut self, totals: &CodexTotals) { + let Some(water) = self.totals_watermark.as_ref() else { + return; + }; + if totals.input < water.input + || totals.cached < water.cached + || totals.output < water.output + { + self.saw_interleaved_totals = true; + } + } + + fn raise_watermark(&mut self, totals: &CodexTotals) { + self.totals_watermark = Some(match self.totals_watermark.as_ref() { + Some(water) => CodexTotals { + input: water.input.max(totals.input), + cached: water.cached.max(totals.cached), + output: water.output.max(totals.output), + }, + None => totals.clone(), + }); + } +} - self.previous_totals = Some(totals); - (delta_input, delta_cached, delta_output) +fn model_evidence(raw: &str) -> Option<&str> { + let trimmed = raw.trim(); + (!trimmed.is_empty()).then_some(trimmed) +} + +/// When interleaved Ultra lineages reset cumulative counters, only count growth +/// above the historical high watermark so rewound branches do not re-add work. +fn contained_total_delta( + watermark: Option<&CodexTotals>, + counted: Option<&CodexTotals>, + current: &CodexTotals, +) -> CodexTotals { + let water = watermark.cloned().unwrap_or(CodexTotals { + input: 0, + cached: 0, + output: 0, + }); + let counted = counted.cloned().unwrap_or(CodexTotals { + input: 0, + cached: 0, + output: 0, + }); + + let component = |water: i32, counted: i32, current: i32| -> i32 { + if current >= water { + (current - water.max(counted)).max(0) + } else { + (current - counted).max(0) + } + }; + + CodexTotals { + input: component(water.input, counted.input, current.input), + cached: component(water.cached, counted.cached, current.cached), + output: component(water.output, counted.output, current.output), + } +} + +/// Read one JSONL line, discarding content when it exceeds `max_bytes`. +/// Returns `(line_without_newline, bytes_consumed_including_newline)`. +fn read_bounded_jsonl_line( + reader: &mut R, + max_bytes: usize, +) -> std::io::Result, usize)>> { + let mut line = Vec::new(); + let mut saw_bytes = false; + let mut discarding = false; + let mut consumed_total = 0; + + loop { + let chunk = reader.fill_buf()?; + if chunk.is_empty() { + return Ok( + saw_bytes.then_some((if discarding { Vec::new() } else { line }, consumed_total)) + ); + } + let newline = chunk.iter().position(|byte| *byte == b'\n'); + let segment_end = newline.unwrap_or(chunk.len()); + let segment = &chunk[..segment_end]; + saw_bytes = true; + + if !discarding { + let remaining = max_bytes.saturating_sub(line.len()); + if segment.len() <= remaining { + line.extend_from_slice(segment); + } else { + line.clear(); + discarding = true; + } + } + + let consumed = segment_end + usize::from(newline.is_some()); + reader.consume(consumed); + consumed_total += consumed; + if newline.is_some() { + return Ok(Some(( + if discarding { Vec::new() } else { line }, + consumed_total, + ))); + } } } @@ -599,12 +796,18 @@ impl JsonlScanner { let mut parser = CodexParserState::new(initial_model, initial_totals); let mut parsed_bytes = start_offset; - let mut line = String::new(); - while reader.read_line(&mut line)? > 0 { - parsed_bytes += line.len() as i64; - parser.process_line(&line, range); - - line.clear(); + while let Some((line_bytes, consumed)) = + read_bounded_jsonl_line(&mut reader, CODEX_JSONL_MAX_LINE_BYTES)? + { + parsed_bytes += consumed as i64; + if line_bytes.is_empty() { + continue; + } + let Ok(line) = std::str::from_utf8(&line_bytes) else { + continue; + }; + let line = line.strip_suffix('\r').unwrap_or(line); + parser.process_line(line, range); } Ok(CodexParseResult { @@ -615,6 +818,15 @@ impl JsonlScanner { }) } + /// Whether a cached scan should be reused under `options` (issue #2089). + pub fn should_skip_cached_scan( + cache: &CostUsageCache, + options: CostScanOptions, + now_unix_ms: i64, + ) -> bool { + options.should_skip_scan(cache.last_scan_unix_ms, now_unix_ms) + } + /// Load cache from disk pub fn load_cache(provider: ProviderId, cache_root: Option<&Path>) -> CostUsageCache { let cache_path = Self::cache_path(provider, cache_root); @@ -820,4 +1032,174 @@ mod tests { assert_eq!(record.model, "gpt-5.5"); assert_eq!((record.input, record.cached, record.output), (45, 12, 8)); } + + #[test] + fn codex_parser_discards_oversized_line_and_recovers_next_record() { + let mut file = tempfile::NamedTempFile::new().expect("temp file"); + let padding = "x".repeat(CODEX_JSONL_MAX_LINE_BYTES); + writeln!( + file, + r#"{{"timestamp":"2026-05-31T10:00:00Z","type":"turn_context","payload":{{"model":"{padding}"}}}}"# + ) + .unwrap(); + writeln!( + file, + r#"{{"timestamp":"2026-05-31T10:00:01Z","type":"event_msg","payload":{{"type":"token_count","info":{{"last_token_usage":{{"input_tokens":9,"cached_input_tokens":2,"output_tokens":1}}}}}}}}"# + ) + .unwrap(); + + let day = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let parsed = JsonlScanner::parse_codex_file( + file.path(), + &CostUsageDayRange::new(day, day), + 0, + None, + None, + ) + .expect("parse"); + + assert_eq!(parsed.records.len(), 1); + assert_eq!( + parsed.records[0].model, + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL + ); + assert_eq!( + ( + parsed.records[0].input, + parsed.records[0].cached, + parsed.records[0].output + ), + (9, 2, 1) + ); + } + + #[test] + fn bounded_jsonl_reader_accepts_exact_limit_without_retaining_larger_input() { + let mut input = vec![b'x'; CODEX_JSONL_MAX_LINE_BYTES]; + input.push(b'\n'); + input.extend_from_slice(b"{\"type\":\"event_msg\"}\n"); + let mut reader = BufReader::with_capacity(64 * 1024, std::io::Cursor::new(input)); + + let (exact, _) = read_bounded_jsonl_line(&mut reader, CODEX_JSONL_MAX_LINE_BYTES) + .expect("read") + .expect("line"); + let (later, _) = read_bounded_jsonl_line(&mut reader, CODEX_JSONL_MAX_LINE_BYTES) + .expect("read") + .expect("line"); + + assert_eq!(exact.len(), CODEX_JSONL_MAX_LINE_BYTES); + assert_eq!(later, br#"{"type":"event_msg"}"#); + } + + #[test] + fn codex_turn_context_wins_over_conflicting_event_model() { + let day = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let range = CostUsageDayRange::new(day, day); + let mut parser = CodexParserState::new(None, None); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:00Z","type":"turn_context","payload":{"model":"gpt-5.5"}}"#, + &range, + ); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:01Z","type":"event_msg","payload":{"type":"token_count","model":"gpt-5.6-sol","info":{"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#, + &range, + ); + + assert_eq!(parser.records[0].model, "gpt-5.5"); + } + + #[test] + fn codex_blank_context_clears_stale_model_and_emits_unattributed_usage() { + let day = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let range = CostUsageDayRange::new(day, day); + let mut parser = CodexParserState::new(Some("gpt-5.5".to_string()), None); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:00Z","type":"turn_context","payload":{"model":" "}}"#, + &range, + ); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:01Z","type":"event_msg","payload":{"type":"token_count","info":{"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#, + &range, + ); + + assert_eq!( + parser.records[0].model, + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL + ); + } + + #[test] + fn codex_model_less_token_event_uses_unpriced_sentinel() { + let day = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let range = CostUsageDayRange::new(day, day); + let mut parser = CodexParserState::new(None, None); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:01Z","type":"event_msg","payload":{"type":"token_count","info":{"last_token_usage":{"input_tokens":10,"cached_input_tokens":0,"output_tokens":2}}}}"#, + &range, + ); + + assert_eq!(parser.records.len(), 1); + assert_eq!( + parser.records[0].model, + CostUsagePricing::CODEX_UNATTRIBUTED_MODEL + ); + } + + #[test] + fn interleaved_lineage_totals_never_exceed_high_watermark_growth() { + let day = NaiveDate::from_ymd_opt(2026, 5, 31).unwrap(); + let range = CostUsageDayRange::new(day, day); + let mut parser = CodexParserState::new(Some("gpt-5.6-sol".to_string()), None); + + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:01Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"cached_input_tokens":0,"output_tokens":20}}}}"#, + &range, + ); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":5,"cached_input_tokens":0,"output_tokens":1}}}}"#, + &range, + ); + parser.process_line( + r#"{"timestamp":"2026-05-31T10:00:03Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":101,"cached_input_tokens":0,"output_tokens":21}}}}"#, + &range, + ); + + let total_input: i32 = parser.records.iter().map(|r| r.input).sum(); + let total_output: i32 = parser.records.iter().map(|r| r.output).sum(); + assert!( + total_input <= 101, + "input inflated to {total_input}, expected <= 101" + ); + assert!( + total_output <= 21, + "output inflated to {total_output}, expected <= 21" + ); + } + + #[test] + fn cost_scan_options_app_driven_bypasses_debounce() { + let debounced = CostScanOptions::default(); + let forced = CostScanOptions::app_driven(); + let last = 1_000_000_i64; + let now = last + 1_000; // 1s later, within 60s window + + assert!(debounced.should_skip_scan(last, now)); + assert!(!forced.should_skip_scan(last, now)); + assert!(!debounced.should_skip_scan(last, last + 61_000)); + + let cache = CostUsageCache { + last_scan_unix_ms: last, + ..Default::default() + }; + assert!(JsonlScanner::should_skip_cached_scan( + &cache, + CostScanOptions::default(), + now + )); + assert!(!JsonlScanner::should_skip_cached_scan( + &cache, + CostScanOptions::app_driven(), + now + )); + } }