diff --git a/CHANGELOG.md b/CHANGELOG.md index 78a489e6b5c..ae6a8eda9cb 100755 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ ## [Windows] Unreleased +- Codex CLI: add opt-in `cost --identified-only` request subtotals across local and SSH (`--remote`) histories, deduplicating strong request identities and keeping partial results without estimating a combined price, plus `--group-by project` reports that show the partial-history notice even when no projects are listed (upstream #4447, #4472). - OpenCode Go: reuse unchanged local history scans (keyed by `opencode.db` and its `-wal` sidecar size and modification time, at most one 4 MiB payload kept), and in Auto overlay Console quota from a stored or automatically discovered browser session before falling back to the local estimate; cookie source Off never reads a browser (upstream #4410, #4445). - Add **Parallel** prepaid balance: reads the account balance and any pending debit from the platform.parallel.ai console using your browser session or a pasted Cookie header. A signed-out console moves on to the next browser session and reports an expired session when none remain (upstream #4479, #4508). - Add **fal** account credit balance: reads the credit balance and account name from fal's billing API with your API key (`FAL_KEY` or Settings) and explains when the key lacks the billing read permission (upstream #4438). diff --git a/docs/CLI.md b/docs/CLI.md index 5a5903cd4db..18c0e481041 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -92,6 +92,8 @@ codexbar cost codexbar cost -p codex -f json --pretty codexbar cost -p codex --remote user@mac-host codexbar cost -p codex --format json --summary-only --provider-native-only --days 30 +codexbar cost -p codex --group-by project +codexbar cost -p codex --identified-only --format json --days 30 --remote user@build-host codexbar cost --period month-to-date --json codexbar cost --period all --json codexbar cost --days 7 --json @@ -108,7 +110,16 @@ Antigravity history that stopped short is never shown as exact. A scan whose dec Claude/Codex costs come from local session logs. Antigravity exposes local **token history only** through `cost`; dollar cost remains unknown rather than becoming a false `$0`. Other providers may differ; do not assume upstream Cursor dashboard cost behavior unless implemented in this tree. -`--remote` adds one separate native Codex report fetched through non-interactive SSH; overlapping local and remote histories are never combined. `--summary-only` emits the versioned, path-free JSON contract used by the remote comparison and accepts only `--provider codex --format json`. Both modes reject session grouping and other provider selections. +`--remote` adds one separate native Codex report fetched through non-interactive SSH; overlapping local and remote histories are never combined. `--summary-only` emits the versioned, path-free JSON contract used by the remote comparison and accepts only `--provider codex --format json`. Both modes reject project/session grouping and other provider selections. + +`--group-by project` lists Codex local projects from the workspace index (text only; other providers print a not-available line and JSON output is unchanged). When the local cost scan has incomplete coverage, the report shows the same `Coverage: partial (history catch-up in progress)` notice as the default Codex text report, including when no project rows are available. Hide personal information replaces project names and drops paths. + +**Identified-request subtotal (experimental).** `codexbar cost --provider codex --identified-only --format json --days 30` exports a versioned native request ledger (`schemaVersion` 1) over one inclusive UTC window. `--identified-end ` pins the end (default: now); the start is UTC midnight `days - 1` days earlier, and saved periods and time zones do not change it. Add `--remote ` to return per-source `reports` plus an **identified-request subtotal** for this machine and that host; the remote host needs a CLI that supports `--identified-only`. + +- Only owned native `token_usage_record` events with thread and response IDs contribute. Legacy `token_count` usage without a matching identified request, and unidentified usage, is excluded and counted in `excludedCount`; the subtotal is not complete account usage. Foreign-provider routes (for example `deepseek/...`) are not Codex usage. +- Identical copies count once. Contradictory copies of one identity are excluded for the whole merge, even after a later matching copy. Identities are length-framed SHA-256 hashes, so rows carry timestamps and token categories without conversation text, model names, paths, or raw session IDs. +- Combined prices are unavailable (`costUSD: null`); no dollar estimate is made. Exports are limited to 50,000 records and conflict IDs and 32 MiB of remote output; exceeding a limit rejects that source instead of truncating it. An unavailable source keeps the successful source data, marks the result `partial`, and exits nonzero. Readable empty history is available; malformed lines, dropped usage lines, or files that change during the scan mark the data partial. +- Requires explicit `--provider codex` and JSON, and `--days` 1..=365 (30 by default). It cannot be combined with `--summary-only`, `--period`, or `--group-by`. The separate-host `--remote` summary without `--identified-only` is unchanged. Codex local-history scans use a 60-second scanner-side debounce for ordinary disk-cache reads. This is separate from the desktop provider refresh setting. With Adaptive refresh off, **Manual** (`refresh_interval_secs = 0`) disables the recurring desktop refresh timer, but it does not forbid startup/stale-aware reads, explicit refreshes, or pending Codex catch-up scans. Low Power Mode floors recurring automatic refreshes to 30 minutes; explicit/manual work remains immediate. diff --git a/rust/src/agent_sessions/remote.rs b/rust/src/agent_sessions/remote.rs index e6ff1aca54d..d4e1be343c0 100644 --- a/rust/src/agent_sessions/remote.rs +++ b/rust/src/agent_sessions/remote.rs @@ -171,14 +171,62 @@ impl RemoteSessionFetcher { history_days: u32, force_refresh: bool, ) -> Result { - let host = Self::validate_codex_cost_host(host)?; if !(1..=365).contains(&history_days) { return Err("history days must be between 1 and 365".to_string()); } - let refresh = if force_refresh { " --refresh" } else { "" }; + Self::codex_cli_options( + host, + &format!( + "cost --provider codex --format json --summary-only --provider-native-only --days {history_days}{refresh}" + ), + ) + } + + /// Fetch the identified-request ledger (upstream #4447) over the same + /// one-shot SSH boundary, with an explicit UTC millisecond window. + pub(crate) async fn fetch_codex_identified_usage( + &self, + host: &str, + history_days: u32, + end_unix_ms: i64, + ) -> Result { + let options = Self::codex_identified_options(host, history_days, end_unix_ms)?; + let result = Self::codex_cost_runner() + .run_async("ssh", None, &options) + .await + .map_err(|error| error.to_string())?; + if result.timed_out || result.exit_code != Some(0) { + return Err("remote identified usage command failed".to_string()); + } + if result.text.len() > crate::codex_costs::MAX_IDENTIFIED_OUTPUT_BYTES { + return Err("remote identified usage output is too large".to_string()); + } + Ok(result.text) + } + + pub(crate) fn codex_identified_options( + host: &str, + history_days: u32, + end_unix_ms: i64, + ) -> Result { + if !(1..=365).contains(&history_days) || end_unix_ms < 0 { + return Err("invalid identified usage window".to_string()); + } + Self::codex_cli_options( + host, + &format!( + "cost --provider codex --format json --identified-only --days {history_days} --identified-end {end_unix_ms}" + ), + ) + } + + /// Select the executable before scanning: a failed scan must never invoke + /// a fallback scan. + fn codex_cli_options(host: &str, cli_arguments: &str) -> Result { + let host = Self::validate_codex_cost_host(host)?; let command = format!( - "if command -v codexbar >/dev/null 2>&1; then exec codexbar cost --provider codex --format json --summary-only --provider-native-only --days {history_days}{refresh}; else exec /Applications/CodexBar.app/Contents/Helpers/CodexBarCLI cost --provider codex --format json --summary-only --provider-native-only --days {history_days}{refresh}; fi" + "if command -v codexbar >/dev/null 2>&1; then exec codexbar {cli_arguments}; else exec /Applications/CodexBar.app/Contents/Helpers/CodexBarCLI {cli_arguments}; fi" ); let remote_command = format!("'{command}'"); @@ -420,4 +468,25 @@ mod tests { assert!(RemoteSessionFetcher::codex_cost_options("build-host", 0, false).is_err()); assert!(RemoteSessionFetcher::codex_cost_options("build-host", 366, false).is_err()); } + + #[test] + fn codex_identified_options_request_the_exact_window_noninteractively() { + let options = + RemoteSessionFetcher::codex_identified_options("build-host", 7, 1_700_000_000_000) + .unwrap(); + assert!(options.extra_args.contains(&"BatchMode=yes".to_string())); + assert!( + options + .extra_args + .contains(&"StrictHostKeyChecking=yes".to_string()) + ); + let command = options.extra_args.last().unwrap(); + assert!(command.contains( + "cost --provider codex --format json --identified-only --days 7 --identified-end 1700000000000" + )); + assert!(!command.contains("--summary-only")); + assert!(RemoteSessionFetcher::codex_identified_options("-oProxyCommand=x", 7, 0).is_err()); + assert!(RemoteSessionFetcher::codex_identified_options("build-host", 0, 0).is_err()); + assert!(RemoteSessionFetcher::codex_identified_options("build-host", 7, -1).is_err()); + } } diff --git a/rust/src/cli/cost.rs b/rust/src/cli/cost.rs index efaef2dbf36..47cd64a7e6e 100755 --- a/rust/src/cli/cost.rs +++ b/rust/src/cli/cost.rs @@ -9,7 +9,8 @@ use super::cost_period::{ }; use super::usage::{OutputFormat, ProviderSelection}; use crate::codex_costs::{ - CodexHostCostReport, CodexHostCostsArgs, CodexHostOutcome, run_codex_host_costs, + CodexHostCostReport, CodexHostCostsArgs, CodexHostOutcome, IdentifiedUsageArgs, + run_codex_host_costs, run_codex_identified_usage, }; use crate::codex_workspaces::short_session_id; use crate::core::{CostScanOptions, ProviderId}; @@ -61,11 +62,11 @@ pub struct CostArgs { #[arg(long = "provider-native-only")] pub provider_native_only: bool, - /// Group text output by Codex local conversation/session. - #[arg(long = "group-by", value_parser = ["session"])] + /// Group Codex text output by local project or conversation/session. + #[arg(long = "group-by", value_parser = ["project", "session"])] pub group_by: Option, - /// Also report native Codex costs from one SSH host as a separate report. + /// Read one SSH host; separate summaries unless --identified-only is selected. #[arg(long)] pub remote: Option, @@ -73,6 +74,14 @@ pub struct CostArgs { #[arg(long = "summary-only")] pub summary_only: bool, + /// Native Codex identified-request subtotal JSON; prices unavailable. + #[arg(long = "identified-only")] + pub identified_only: bool, + + /// Identified subtotal window end, inclusive Unix milliseconds (UTC). + #[arg(long = "identified-end", value_name = "UNIX_MS")] + pub identified_end: Option, + /// Antigravity only: when a recorded model has no known public price, wait /// for one bounded models.dev pricing refresh and rescan. Without it the /// CLI starts no pricing download. @@ -83,18 +92,24 @@ pub struct CostArgs { #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum CostGroupBy { None, + Project, Session, } impl CostGroupBy { fn from_arg(raw: Option<&str>) -> Self { match raw { + Some("project") => Self::Project, Some("session") => Self::Session, _ => Self::None, } } } +/// Shown by the default and project Codex reports when the local cost scan +/// has not fully covered the window. +const PARTIAL_HISTORY_NOTICE: &str = " Coverage: partial (history catch-up in progress)"; + /// Run the cost command pub async fn run(args: CostArgs) -> anyhow::Result<()> { let format = if args.json { @@ -107,6 +122,27 @@ pub async fn run(args: CostArgs) -> anyhow::Result<()> { let group_by = CostGroupBy::from_arg(args.group_by.as_deref()); let use_color = !args.no_color && is_terminal(); + // The identified window is explicit UTC milliseconds; saved periods and + // time zones never apply, so dispatch before resolving the period. + if args.identified_only || args.identified_end.is_some() { + return run_codex_identified_usage(&IdentifiedUsageArgs { + identified_only: args.identified_only, + provider_is_codex_only: args + .provider + .as_deref() + .is_some_and(|provider| provider.trim().eq_ignore_ascii_case("codex")), + json: format == OutputFormat::Json, + days: args.days, + end_unix_ms: args.identified_end, + remote: args.remote.clone(), + period_set: args.period.is_some(), + group_by_set: args.group_by.is_some(), + summary_only: args.summary_only, + pretty: args.pretty, + }) + .await; + } + let settings = Settings::load(); let host_summary = args.remote.is_some() || args.summary_only; let period = resolve_period( @@ -230,7 +266,13 @@ pub async fn run(args: CostArgs) -> anyhow::Result<()> { match format { OutputFormat::Text => { - print_text_output(&results, use_color, period, group_by); + print_text_output( + &results, + use_color, + period, + group_by, + settings.hide_personal_info, + ); } OutputFormat::Json => { // Upstream 0.60.4 refreshes OpenCodex prices before the JSON @@ -316,6 +358,7 @@ fn print_text_output( use_color: bool, period: CostReportingPeriod, group_by: CostGroupBy, + hide_personal_info: bool, ) { let label = period.label(); for (i, result) in results.iter().enumerate() { @@ -330,7 +373,11 @@ fn print_text_output( println!("{title}"); } - if let Some(history) = result.token_history.as_ref() { + if group_by == CostGroupBy::Project && result.provider == "codex" { + print_codex_project_output(result, period, hide_personal_info); + } else if group_by == CostGroupBy::Project { + println!(" Project grouping is only available for Codex local projects"); + } else if let Some(history) = result.token_history.as_ref() { print_local_token_history(history); } else if group_by == CostGroupBy::Session && result.provider == "codex" { print_codex_session_output(result, period); @@ -415,7 +462,7 @@ fn print_text_output( // A16 (upstream 0.48.0): coverage status for Codex. if result.provider == "codex" && !result.summary.history_coverage_established { - println!(" Coverage: partial (history catch-up in progress)"); + println!("{PARTIAL_HISTORY_NOTICE}"); } } @@ -525,6 +572,68 @@ fn print_codex_session_output(result: &CostResult, period: CostReportingPeriod) println!(" Not a subscription bill or plan value · local usage × public API prices"); } +fn print_codex_project_output( + result: &CostResult, + period: CostReportingPeriod, + hide_personal_info: bool, +) { + let index = crate::codex_workspaces::CodexWorkspacesIndex::new(rolling_window_days(period)); + let snapshot = index.load_snapshot(false, |_| {}).map(|mut snapshot| { + if hide_personal_info { + snapshot.redact_for_privacy(); + } + snapshot + }); + for line in render_codex_project_lines(result.summary.history_coverage_established, &snapshot) { + println!("{line}"); + } +} + +/// Project report lines (upstream #4472): incomplete history carries the same +/// partial-history notice as the default report, even with no project rows. +fn render_codex_project_lines( + cost_history_established: bool, + snapshot: &Result, +) -> Vec { + let mut lines = Vec::new(); + let catalog_partial = snapshot + .as_ref() + .is_ok_and(|snapshot| snapshot.source_status.is_partial()); + if !cost_history_established || catalog_partial { + lines.push(PARTIAL_HISTORY_NOTICE.to_string()); + } + let snapshot = match snapshot { + Ok(snapshot) => snapshot, + Err(err) => { + lines.push(format!(" Project history unavailable: {err}")); + return lines; + } + }; + lines.push(format!(" Projects (last {} days):", snapshot.history_days)); + if snapshot.projects.is_empty() { + lines.push(" —".to_string()); + } + for project in &snapshot.projects { + let cost = if project.cost_estimate.unknown_tokens > 0 { + format!("~${:.2} partial", project.cost_estimate.known_usd) + } else { + format!("${:.2}", project.cost_estimate.known_usd) + }; + lines.push(format!( + " {}: {cost} · {} tokens", + project.display_name, + format_number(project.totals.total_tokens) + )); + if let Some(path) = project.path.as_deref() { + lines.push(format!(" {path}")); + } + } + lines.push( + " Not a subscription bill or plan value · local usage × public API prices".to_string(), + ); + lines +} + /// Print JSON output fn build_json_payloads( results: &[CostResult], @@ -986,6 +1095,107 @@ mod tests { fn group_by_defaults_none_and_accepts_session() { assert_eq!(CostGroupBy::from_arg(None), CostGroupBy::None); assert_eq!(CostGroupBy::from_arg(Some("session")), CostGroupBy::Session); + assert_eq!(CostGroupBy::from_arg(Some("project")), CostGroupBy::Project); + assert_eq!( + parse_args(&["--group-by", "project"]).group_by.as_deref(), + Some("project") + ); + } + + #[test] + fn identified_flags_parse_with_remote() { + let args = parse_args(&[ + "--provider", + "codex", + "--format", + "json", + "--identified-only", + "--identified-end", + "1700000000000", + "--remote", + "build-host", + ]); + assert!(args.identified_only); + assert_eq!(args.identified_end, Some(1_700_000_000_000)); + assert_eq!(args.remote.as_deref(), Some("build-host")); + assert!(!parse_args(&[]).identified_only); + } + + fn project_snapshot( + projects: Vec, + ) -> crate::codex_workspaces::CodexLocalProjectUsageSnapshot { + crate::codex_workspaces::CodexLocalProjectUsageSnapshot { + updated_at: chrono::DateTime::from_timestamp(1_700_000_000, 0).unwrap(), + history_days: 30, + scope_signature: String::new(), + indexed_file_count: 0, + skipped_file_count: 0, + total: Default::default(), + sessions: Vec::new(), + projects, + daily: Vec::new(), + source_status: crate::codex_workspaces::SourceStatus::Complete, + } + } + + #[test] + fn project_report_marks_partial_history_even_when_empty() { + let empty: Result<_, String> = Ok(project_snapshot(Vec::new())); + let partial = render_codex_project_lines(false, &empty); + assert_eq!(partial[0], PARTIAL_HISTORY_NOTICE); + assert!(partial.contains(&" Projects (last 30 days):".to_string())); + assert!(partial.contains(&" —".to_string())); + + let complete = render_codex_project_lines(true, &empty); + assert!(!complete.iter().any(|line| line == PARTIAL_HISTORY_NOTICE)); + assert!(complete.contains(&" —".to_string())); + + let unavailable: Result = + Err("locked".to_string()); + let lines = render_codex_project_lines(false, &unavailable); + assert_eq!(lines[0], PARTIAL_HISTORY_NOTICE); + assert_eq!(lines[1], " Project history unavailable: locked"); + + let mut locked_catalog = project_snapshot(Vec::new()); + locked_catalog.source_status = crate::codex_workspaces::SourceStatus::CatalogLocked; + let lines = render_codex_project_lines(true, &Ok::<_, String>(locked_catalog)); + assert_eq!(lines[0], PARTIAL_HISTORY_NOTICE); + assert_eq!( + lines + .iter() + .filter(|line| *line == PARTIAL_HISTORY_NOTICE) + .count(), + 1 + ); + } + + #[test] + fn project_report_lists_projects_with_partial_costs() { + use crate::codex_workspaces::{CostEstimate, ProjectUsage, UsageTotals}; + let project = |name: &str, path: Option<&str>, unknown_tokens: u64| ProjectUsage { + id: name.to_string(), + display_name: name.to_string(), + path: path.map(str::to_string), + totals: UsageTotals::from_parts(1_000, 0, 234), + cost_estimate: CostEstimate { + known_usd: 1.5, + unknown_tokens, + }, + session_count: 1, + latest_activity: None, + top_model: None, + top_sessions: Vec::new(), + is_projectless: false, + }; + let snapshot: Result<_, String> = Ok(project_snapshot(vec![ + project("alpha", Some("C:\\src\\alpha"), 0), + project("beta", None, 10), + ])); + let lines = render_codex_project_lines(true, &snapshot); + assert!(lines.contains(&" alpha: $1.50 · 1,234 tokens".to_string())); + assert!(lines.contains(&" C:\\src\\alpha".to_string())); + assert!(lines.contains(&" beta: ~$1.50 partial · 1,234 tokens".to_string())); + assert!(!lines.iter().any(|line| line == " —")); } #[test] diff --git a/rust/src/codex_costs.rs b/rust/src/codex_costs.rs index 2b631ea0e38..65d227f1155 100644 --- a/rust/src/codex_costs.rs +++ b/rust/src/codex_costs.rs @@ -4,10 +4,14 @@ //! local-scan aggregation and pricing logic. mod host_costs; +mod identified_requests; mod quota_windows; mod summary_contract; pub(crate) use host_costs::{CodexHostCostsArgs, HostOutputFormat, run_codex_host_costs}; +pub(crate) use identified_requests::{ + IdentifiedUsageArgs, MAX_IDENTIFIED_OUTPUT_BYTES, run_codex_identified_usage, +}; pub use quota_windows::{CodexQuotaWindow, codex_quota_windows_from_cache}; pub(crate) use summary_contract::{ CodexCostSummary, CodexHostCostReport, CodexHostCostWindow, CodexHostOutcome, diff --git a/rust/src/codex_costs/identified_requests.rs b/rust/src/codex_costs/identified_requests.rs new file mode 100644 index 00000000000..e9d7b09b944 --- /dev/null +++ b/rust/src/codex_costs/identified_requests.rs @@ -0,0 +1,972 @@ +//! Opt-in Codex identified-request subtotals (upstream #4447). +//! +//! Only owned native `token_usage_record` events that carry a thread and a +//! response identity contribute. Identities are length-framed SHA-256 hashes, +//! so exports carry timestamps and token categories without conversation +//! text, model names, paths, or raw session ids. Combined prices are +//! deliberately unavailable (`costUSD: null`). + +use std::collections::{HashMap, HashSet}; +use std::fs::{self, File}; +use std::future::Future; +use std::io::{BufRead, BufReader}; +use std::path::{Path, PathBuf}; +use std::time::SystemTime; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; + +use crate::agent_sessions::RemoteSessionFetcher; +use crate::core::{CostUsagePricing, JsonlScanner}; + +pub(crate) const MAX_IDENTIFIED_RECORDS: usize = 50_000; +pub(crate) const MAX_IDENTIFIED_OUTPUT_BYTES: usize = 32 * 1024 * 1024; +pub(crate) const MAX_IDENTIFIED_END_UNIX_MS: i64 = 253_402_300_799_999; +const DAY_MS: i64 = 86_400_000; +const MAX_LINE_BYTES: usize = 256 * 1024; + +pub(crate) const LOCAL_IDENTIFIED_UNAVAILABLE: &str = + "Local native Codex identified usage is unavailable."; +pub(crate) const REMOTE_IDENTIFIED_UNAVAILABLE: &str = + "Remote identified usage is unavailable. Check SSH and update its CLI for --identified-only."; +pub(crate) const IDENTIFIED_USAGE_ARGS_ERROR: &str = "Use --identified-only --provider codex with JSON, --days 1...365 and one optional --remote host. --identified-end must be Unix milliseconds; period, grouping, breakdown and separate host summaries are unsupported."; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum IdentifiedUsageError { + Unavailable, + InvalidReport, +} + +/// A provider-owned request; input includes cached input, while reasoning is +/// part of output. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CodexRequestRecord { + pub id: String, + pub timestamp_unix_ms: i64, + pub input_tokens: i64, + pub cache_read_tokens: i64, + pub output_tokens: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reasoning_tokens: Option, +} + +impl CodexRequestRecord { + fn usage(&self) -> (i64, i64, i64, Option) { + ( + self.input_tokens, + self.cache_read_tokens, + self.output_tokens, + self.reasoning_tokens, + ) + } +} + +/// One exact inclusive UTC window. This identifies requests without claiming +/// account-wide usage. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CodexRequestLedger { + pub schema_version: u32, + pub provider: String, + pub start_unix_ms: i64, + pub end_unix_ms: i64, + pub records: Vec, + #[serde(rename = "conflictingIDs")] + pub conflicting_ids: Vec, + pub excluded_count: i64, + pub partial: bool, +} + +/// Numeric identified-request subtotal; pricing is deliberately unavailable. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CodexRequestSubtotal { + pub kind: &'static str, + pub total_tokens: i64, + pub request_count: usize, + pub conflict_count: usize, + pub excluded_count: i64, + pub partial: bool, + #[serde(rename = "costUSD")] + pub cost_usd: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize)] +pub(crate) struct CodexIdentifiedSource { + pub host: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub ledger: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CodexIdentifiedResult { + pub reports: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub subtotal: Option, + pub all_sources_available: bool, + pub partial: bool, +} + +fn is_valid_window(start_unix_ms: i64, end_unix_ms: i64) -> bool { + start_unix_ms >= 0 + && end_unix_ms <= MAX_IDENTIFIED_END_UNIX_MS + && end_unix_ms >= start_unix_ms + && start_unix_ms % DAY_MS == 0 + && end_unix_ms - start_unix_ms < 365 * DAY_MS +} + +/// The inclusive start of a `days`-long UTC window ending at `end_unix_ms`: +/// UTC midnight `days - 1` days before the end's day. +pub(crate) fn identified_window_start(days: u32, end_unix_ms: i64) -> Option { + if !(1..=365).contains(&days) || !(0..=MAX_IDENTIFIED_END_UNIX_MS).contains(&end_unix_ms) { + return None; + } + let start = end_unix_ms.div_euclid(DAY_MS) * DAY_MS - i64::from(days - 1) * DAY_MS; + (start >= 0).then_some(start) +} + +fn is_digest(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +fn digest(components: &[&str]) -> String { + let framed: String = components + .iter() + .map(|component| format!("{}:{component}", component.len())) + .collect(); + Sha256::digest(framed.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +impl CodexRequestLedger { + pub(crate) fn empty(start_unix_ms: i64, end_unix_ms: i64) -> Self { + Self { + schema_version: 1, + provider: "codex".to_string(), + start_unix_ms, + end_unix_ms, + records: Vec::new(), + conflicting_ids: Vec::new(), + excluded_count: 0, + partial: false, + } + } + + pub(crate) fn validate( + &self, + start_unix_ms: i64, + end_unix_ms: i64, + ) -> Result<(), IdentifiedUsageError> { + let unique_conflicts: HashSet<&str> = + self.conflicting_ids.iter().map(String::as_str).collect(); + let header_ok = self.schema_version == 1 + && self.provider == "codex" + && self.start_unix_ms == start_unix_ms + && self.end_unix_ms == end_unix_ms + && is_valid_window(start_unix_ms, end_unix_ms) + && self.records.len() <= MAX_IDENTIFIED_RECORDS + && self.conflicting_ids.len() <= MAX_IDENTIFIED_RECORDS + && self.excluded_count >= 0 + && unique_conflicts.len() == self.conflicting_ids.len() + && self.conflicting_ids.iter().all(|id| is_digest(id)); + if !header_ok { + return Err(IdentifiedUsageError::InvalidReport); + } + let mut total: i64 = 0; + for record in &self.records { + let record_ok = is_digest(&record.id) + && (start_unix_ms..=end_unix_ms).contains(&record.timestamp_unix_ms) + && record.input_tokens >= 0 + && (0..=record.input_tokens).contains(&record.cache_read_tokens) + && record.output_tokens >= 0 + && record + .reasoning_tokens + .is_none_or(|reasoning| (0..=record.output_tokens).contains(&reasoning)); + let sum = record.input_tokens.checked_add(record.output_tokens); + total = match (record_ok, sum.and_then(|sum| total.checked_add(sum))) { + (true, Some(total)) => total, + _ => return Err(IdentifiedUsageError::InvalidReport), + }; + } + Ok(()) + } + + /// Merge source ledgers into one subtotal. Conflicts remain excluded even + /// when a later source supplies a clean observation of the identity. + pub(crate) fn merge(ledgers: &[Self]) -> Result { + let first = ledgers.first().ok_or(IdentifiedUsageError::Unavailable)?; + let mut records: HashMap<&str, &CodexRequestRecord> = HashMap::new(); + let mut conflicts: HashSet<&str> = HashSet::new(); + let mut excluded: i64 = 0; + for ledger in ledgers { + ledger.validate(first.start_unix_ms, first.end_unix_ms)?; + excluded = excluded + .checked_add(ledger.excluded_count) + .ok_or(IdentifiedUsageError::InvalidReport)?; + conflicts.extend(ledger.conflicting_ids.iter().map(String::as_str)); + for row in &ledger.records { + if let Some(previous) = records.get(row.id.as_str()) + && previous.usage() != row.usage() + { + conflicts.insert(row.id.as_str()); + } + if records + .get(row.id.as_str()) + .is_none_or(|previous| previous.timestamp_unix_ms > row.timestamp_unix_ms) + { + records.insert(row.id.as_str(), row); + } + } + } + let mut total: i64 = 0; + let mut request_count = 0; + for row in records + .values() + .filter(|row| !conflicts.contains(row.id.as_str())) + { + total = row + .input_tokens + .checked_add(row.output_tokens) + .and_then(|sum| total.checked_add(sum)) + .ok_or(IdentifiedUsageError::InvalidReport)?; + request_count += 1; + } + Ok(CodexRequestSubtotal { + kind: "identified-request-subtotal", + total_tokens: total, + request_count, + conflict_count: conflicts.len(), + excluded_count: excluded, + partial: excluded > 0 || !conflicts.is_empty() || ledgers.iter().any(|l| l.partial), + cost_usd: None, + }) + } +} + +/// Collect the local ledger and, optionally, one remote ledger. A failed +/// source keeps the available ones and marks the result partial. +pub(crate) async fn collect_identified_usage( + remote: Option<&str>, + start_unix_ms: i64, + end_unix_ms: i64, + local: L, + fetch_remote: R, +) -> CodexIdentifiedResult +where + L: FnOnce() -> Result, + R: FnOnce(String) -> Fut, + Fut: Future>, +{ + let mut reports: Vec = Vec::new(); + let mut subtotal = None; + + let mut accept = |host: &str, + fetched: Result, + unavailable: &str, + reports: &mut Vec| { + let accepted = fetched.and_then(|ledger| { + ledger.validate(start_unix_ms, end_unix_ms)?; + // Reject an overflowing source or union while the earlier sources + // remain available. + let mut ledgers: Vec = reports + .iter() + .filter_map(|report| report.ledger.clone()) + .collect(); + ledgers.push(ledger.clone()); + let merged = CodexRequestLedger::merge(&ledgers)?; + Ok((ledger, merged)) + }); + match accepted { + Ok((ledger, merged)) => { + subtotal = Some(merged); + reports.push(CodexIdentifiedSource { + host: host.to_string(), + ledger: Some(ledger), + error: None, + }); + } + Err(_) => reports.push(CodexIdentifiedSource { + host: host.to_string(), + ledger: None, + error: Some(unavailable.to_string()), + }), + } + }; + + accept("local", local(), LOCAL_IDENTIFIED_UNAVAILABLE, &mut reports); + if let Some(host) = remote { + let fetched = fetch_remote(host.to_string()).await; + accept(host, fetched, REMOTE_IDENTIFIED_UNAVAILABLE, &mut reports); + } + + let all_sources_available = reports.iter().all(|report| report.ledger.is_some()); + let partial = !all_sources_available || subtotal.as_ref().is_none_or(|s| s.partial); + CodexIdentifiedResult { + reports, + subtotal, + all_sources_available, + partial, + } +} + +/// Native Codex session roots: `sessions` plus its sibling `archived_sessions`. +fn default_codex_roots() -> Vec { + let Some(root) = JsonlScanner::default_codex_sessions_root() else { + return Vec::new(); + }; + let archived = (root.file_name().is_some_and(|name| name == "sessions")) + .then(|| root.parent().map(|parent| parent.join("archived_sessions"))) + .flatten(); + std::iter::once(root).chain(archived).collect() +} + +pub(crate) fn load_local_ledger( + start_unix_ms: i64, + end_unix_ms: i64, +) -> Result { + load_ledger_from_roots(&default_codex_roots(), start_unix_ms, end_unix_ms) +} + +#[derive(Default)] +struct LedgerBuilder { + records: HashMap, + conflicts: HashSet, + exceeded_limit: bool, + excluded: i64, + partial: bool, +} + +impl LedgerBuilder { + fn mark_conflict(&mut self, id: String) { + if self.conflicts.len() < MAX_IDENTIFIED_RECORDS || self.conflicts.contains(&id) { + self.conflicts.insert(id); + } else { + self.exceeded_limit = true; + } + } + + fn exclude(&mut self) { + self.partial = true; + self.excluded = self.excluded.saturating_add(1); + } + + fn observe(&mut self, row: CodexRequestRecord) { + if let Some(previous) = self.records.get(&row.id) + && previous.usage() != row.usage() + { + self.mark_conflict(row.id.clone()); + } + if self.records.len() < MAX_IDENTIFIED_RECORDS || self.records.contains_key(&row.id) { + if self + .records + .get(&row.id) + .is_none_or(|previous| previous.timestamp_unix_ms > row.timestamp_unix_ms) + { + self.records.insert(row.id.clone(), row); + } + } else { + self.exceeded_limit = true; + } + } +} + +pub(crate) fn load_ledger_from_roots( + roots: &[PathBuf], + start_unix_ms: i64, + end_unix_ms: i64, +) -> Result { + let mut ledger = CodexRequestLedger::empty(start_unix_ms, end_unix_ms); + ledger.validate(start_unix_ms, end_unix_ms)?; + let mut builder = LedgerBuilder::default(); + let mut files = Vec::new(); + let mut found_root = false; + for root in roots { + match fs::metadata(root) { + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Ok(metadata) if metadata.is_dir() => {} + _ => { + builder.partial = true; + continue; + } + } + match collect_jsonl_files(root, &mut files) { + Ok(complete) => { + found_root = true; + builder.partial |= !complete; + } + Err(_) => builder.partial = true, + } + } + if !found_root { + return Err(IdentifiedUsageError::Unavailable); + } + files.sort(); + for file in &files { + if last_written_before(file, start_unix_ms) { + continue; + } + scan_file(file, start_unix_ms, end_unix_ms, &mut builder); + if builder.exceeded_limit { + return Err(IdentifiedUsageError::InvalidReport); + } + } + + let mut records: Vec<_> = builder.records.into_values().collect(); + records.sort_by(|lhs, rhs| lhs.id.cmp(&rhs.id)); + let mut conflicts: Vec<_> = builder.conflicts.into_iter().collect(); + conflicts.sort(); + ledger.records = records; + ledger.excluded_count = builder.excluded; + ledger.partial = builder.partial || builder.excluded > 0 || !conflicts.is_empty(); + ledger.conflicting_ids = conflicts; + ledger.validate(start_unix_ms, end_unix_ms)?; + Ok(ledger) +} + +/// Clock-skew allowance between a record's timestamp and the file's mtime. +const MTIME_SKEW_MARGIN_MS: i64 = 24 * 60 * 60 * 1000; + +/// A file last modified before the window cannot hold an in-window record, +/// so the scan (and the remote SSH budget) can skip it. Unreadable metadata +/// keeps the file so the scan still decides. +fn last_written_before(file: &Path, start_unix_ms: i64) -> bool { + let Some(modified_ms) = fs::metadata(file) + .and_then(|metadata| metadata.modified()) + .ok() + .and_then(|modified| modified.duration_since(std::time::UNIX_EPOCH).ok()) + .and_then(|elapsed| i64::try_from(elapsed.as_millis()).ok()) + else { + return false; + }; + modified_ms.saturating_add(MTIME_SKEW_MARGIN_MS) < start_unix_ms +} + +/// Recursively list `*.jsonl` files. Returns `Ok(false)` when a nested +/// directory could not be read, so the caller can mark the scan partial. +fn collect_jsonl_files(root: &Path, files: &mut Vec) -> std::io::Result { + let mut complete = true; + let mut pending = vec![root.to_path_buf()]; + let mut is_root = true; + while let Some(dir) = pending.pop() { + let entries = match fs::read_dir(&dir) { + Ok(entries) => entries, + Err(error) if is_root => return Err(error), + Err(_) => { + complete = false; + continue; + } + }; + is_root = false; + for entry in entries { + let Ok(entry) = entry else { + complete = false; + continue; + }; + let path = entry.path(); + match entry.file_type() { + Ok(kind) if kind.is_dir() => pending.push(path), + Ok(_) => { + if path + .extension() + .is_some_and(|ext| ext.eq_ignore_ascii_case("jsonl")) + { + files.push(path); + } + } + Err(_) => complete = false, + } + } + } + Ok(complete) +} + +type Usage = (i64, i64, i64, i64); + +fn non_empty_str(value: Option<&Value>) -> Option<&str> { + value + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) +} + +fn token_int( + usage: &serde_json::Map, + key: &str, + default: Option, +) -> Option { + let Some(value) = usage.get(key) else { + return default; + }; + if let Some(number) = value.as_i64() { + return (number >= 0).then_some(number); + } + value + .as_f64() + .filter(|number| number.is_finite() && *number >= 0.0 && number.fract() == 0.0) + .filter(|number| *number < i64::MAX as f64) + .map(|number| { + #[allow( + clippy::cast_possible_truncation, + reason = "the value is a finite, non-negative whole number below i64::MAX" + )] + let whole = number as i64; + whole + }) +} + +/// `(input, cached, output, reasoning)` from one request-usage object. +fn request_usage(value: Option<&Value>) -> Option { + let usage = value?.as_object()?; + let input = token_int(usage, "input_tokens", None)?; + let output = token_int(usage, "output_tokens", None)?; + let cached = token_int(usage, "cached_input_tokens", Some(0))?; + let reasoning = token_int(usage, "reasoning_output_tokens", Some(0))?; + (cached <= input && reasoning <= output && input.checked_add(output).is_some()) + .then_some((input, cached, output, reasoning)) +} + +fn totals_at_least(total: Usage, usage: Usage) -> bool { + total.0 >= usage.0 && total.1 >= usage.1 && total.2 >= usage.2 && total.3 >= usage.3 +} + +fn turn_id(payload: &Value) -> Option { + [Some(payload), payload.get("info")] + .into_iter() + .flatten() + .find_map(|container| { + ["turn_id", "turnId", "id"].iter().find_map(|key| { + container + .get(*key) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string) + }) + }) +} + +struct RequestUsageRecord { + timestamp: String, + thread_id: String, + session_id: Option, + response_id: String, + turn_id: Option, + model: Option, + usage: Usage, + thread_total: Usage, +} + +fn request_usage_record(object: &Value) -> Option { + let payload = object.get("payload")?; + let thread_id = non_empty_str(payload.get("thread_id"))?; + let response_id = non_empty_str(payload.get("response_id"))?; + let timestamp = object.get("timestamp")?.as_str()?; + let usage = request_usage(payload.get("usage"))?; + let thread_total = request_usage(payload.get("thread_token_usage"))?; + if !totals_at_least(thread_total, usage) { + return None; + } + let session_id = non_empty_str(payload.get("session_id")); + if payload.get("session_id").is_some() && session_id.is_none() { + return None; + } + Some(RequestUsageRecord { + timestamp: timestamp.to_string(), + thread_id: thread_id.to_string(), + session_id: session_id.map(str::to_string), + response_id: response_id.to_string(), + turn_id: turn_id(payload), + model: non_empty_str(payload.get("model")).map(str::to_string), + usage, + thread_total, + }) +} + +fn timestamp_unix_ms(timestamp: &str) -> Option { + let parsed = chrono::DateTime::parse_from_rfc3339(timestamp.trim()).ok()?; + let round_up = parsed.timestamp_subsec_nanos() % 1_000_000 >= 500_000; + Some(parsed.timestamp_millis() + i64::from(round_up)) +} + +fn file_fingerprint(path: &Path) -> Option<(u64, Option)> { + fs::metadata(path) + .ok() + .map(|metadata| (metadata.len(), metadata.modified().ok())) +} + +enum JsonlLine { + Complete(Vec), + /// Over [`MAX_LINE_BYTES`]; only the retained prefix is kept. + Oversized(Vec), +} + +fn read_line(reader: &mut R, consumed: &mut u64) -> std::io::Result> { + let mut bytes = Vec::new(); + let mut length = 0_usize; + let mut saw_any = false; + loop { + let buffer = reader.fill_buf()?; + if buffer.is_empty() { + break; + } + saw_any = true; + let (chunk, done) = match buffer.iter().position(|byte| *byte == b'\n') { + Some(index) => (&buffer[..index], Some(index + 1)), + None => (buffer, None), + }; + let room = MAX_LINE_BYTES.saturating_sub(bytes.len()); + bytes.extend_from_slice(&chunk[..chunk.len().min(room)]); + length = length.saturating_add(chunk.len()); + let used = done.unwrap_or(buffer.len()); + reader.consume(used); + *consumed = consumed.saturating_add(used as u64); + if done.is_some() { + break; + } + } + if !saw_any { + return Ok(None); + } + if bytes.last() == Some(&b'\r') { + bytes.pop(); + } + Ok(Some(if length > MAX_LINE_BYTES { + JsonlLine::Oversized(bytes) + } else { + JsonlLine::Complete(bytes) + })) +} + +fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool { + haystack + .windows(needle.len()) + .any(|window| window == needle) +} + +struct LegacyCandidate { + usage: Usage, + total: Option, +} + +fn scan_file(path: &Path, start_unix_ms: i64, end_unix_ms: i64, builder: &mut LedgerBuilder) { + let before = file_fingerprint(path); + let session_id = JsonlScanner::read_codex_session_metadata(path) + .ok() + .and_then(|metadata| metadata.session_id); + let Ok(file) = File::open(path) else { + builder.partial = true; + return; + }; + let mut reader = BufReader::new(file); + let mut parsed_bytes = 0_u64; + let mut current_model: Option = None; + let mut current_turn: Option = None; + let mut turn_models: HashMap = HashMap::new(); + let mut record_totals: HashSet = HashSet::new(); + let mut record_usages: HashSet = HashSet::new(); + let mut legacy: Vec = Vec::new(); + let mut previous_legacy_total: Option = None; + let in_window = |ms: i64| (start_unix_ms..=end_unix_ms).contains(&ms); + + loop { + let line = match read_line(&mut reader, &mut parsed_bytes) { + Ok(Some(line)) => line, + Ok(None) => break, + Err(_) => { + builder.partial = true; + break; + } + }; + let bytes = match line { + JsonlLine::Complete(bytes) => bytes, + JsonlLine::Oversized(prefix) => { + // Oversized prompt-bearing lines are common; only a dropped + // usage line makes the identified history incomplete. + if contains_bytes(&prefix, b"token_usage_record") + || contains_bytes(&prefix, b"\"token_count\"") + { + builder.partial = true; + } + continue; + } + }; + if bytes.iter().all(u8::is_ascii_whitespace) { + continue; + } + let object = match serde_json::from_slice::(&bytes) { + Ok(object) if object.is_object() => object, + _ => { + builder.partial = true; + continue; + } + }; + let payload = object.get("payload").unwrap_or(&Value::Null); + match object.get("type").and_then(Value::as_str) { + Some("turn_context") => { + let model = non_empty_str(payload.get("model")).map(str::to_string); + let turn = turn_id(payload); + if let (Some(turn), Some(model)) = (&turn, &model) { + turn_models.insert(turn.clone(), model.clone()); + } + if model.is_some() { + current_model = model; + } + if turn.is_some() { + current_turn = turn; + } + } + Some("event_msg") => match payload.get("type").and_then(Value::as_str) { + Some("task_started") => { + if let Some(turn) = turn_id(payload) { + current_turn = Some(turn); + } + } + Some("token_count") => { + let info = payload.get("info"); + let Some(usage) = request_usage(info.and_then(|v| v.get("last_token_usage"))) + else { + continue; + }; + let total = request_usage(info.and_then(|v| v.get("total_token_usage"))); + // Codex repeats an unchanged cumulative counter; only a + // changed total describes new legacy usage. + if total.is_some() && total == previous_legacy_total { + continue; + } + previous_legacy_total = total; + let model = current_model.as_deref().unwrap_or_default(); + let timestamp = object + .get("timestamp") + .and_then(Value::as_str) + .and_then(timestamp_unix_ms); + if usage.0.saturating_add(usage.2) > 0 + && timestamp.is_some_and(in_window) + && CostUsagePricing::counts_toward_codex_subscription(model) + { + legacy.push(LegacyCandidate { usage, total }); + } + } + _ => {} + }, + Some("token_usage_record") => { + let Some(session_id) = session_id.as_deref() else { + // Without a session identity no record can be proven owned. + builder.partial = true; + continue; + }; + match request_usage_record(&object) { + Some(record) => { + // Copied parent records still mirror their copied + // legacy counters, so both identify those counters. + record_totals.insert(record.thread_total); + record_usages.insert(record.usage); + let owned = record.thread_id == session_id + && record + .session_id + .as_deref() + .is_none_or(|id| id == session_id); + if !owned { + continue; + } + let turn = record.turn_id.clone().or_else(|| current_turn.clone()); + let model = record + .model + .clone() + .or_else(|| turn.as_ref().and_then(|t| turn_models.get(t).cloned())) + .or_else(|| current_model.clone()); + if model.as_deref().is_some_and(|model| { + !CostUsagePricing::counts_toward_codex_subscription(model) + }) { + continue; + } + let Some(milliseconds) = timestamp_unix_ms(&record.timestamp) else { + builder.exclude(); + continue; + }; + if !in_window(milliseconds) { + continue; + } + let (input, cached, output, reasoning) = record.usage; + builder.observe(CodexRequestRecord { + id: digest(&["codex", &record.thread_id, &record.response_id]), + timestamp_unix_ms: milliseconds, + input_tokens: input, + cache_read_tokens: cached, + output_tokens: output, + reasoning_tokens: Some(reasoning.clamp(0, output.max(0))), + }); + } + None => { + let Some(thread_id) = non_empty_str(payload.get("thread_id")) else { + builder.partial = true; + continue; + }; + let record_session = non_empty_str(payload.get("session_id")); + if payload.get("session_id").is_some() && record_session.is_none() { + builder.partial = true; + continue; + } + let owned = thread_id == session_id + && record_session.is_none_or(|id| id == session_id); + if !owned { + continue; + } + let timestamp = object + .get("timestamp") + .and_then(Value::as_str) + .and_then(timestamp_unix_ms); + if let Some(milliseconds) = timestamp { + if !in_window(milliseconds) { + continue; + } + if let Some(response_id) = non_empty_str(payload.get("response_id")) { + builder.mark_conflict(digest(&["codex", thread_id, response_id])); + } + } + builder.exclude(); + } + } + } + _ => {} + } + } + + // A legacy counter that mirrors any identified request in this file + // (including a copied parent request) is the same usage; only unmatched + // legacy usage is reported as excluded. + let unmatched = legacy + .iter() + .filter(|candidate| { + !candidate + .total + .is_some_and(|total| record_totals.contains(&total)) + && !record_usages.contains(&candidate.usage) + }) + .count(); + builder.excluded = builder + .excluded + .saturating_add(i64::try_from(unmatched).unwrap_or(i64::MAX)); + + let after = file_fingerprint(path); + if before.is_none() || before.map(|(size, _)| size) != Some(parsed_bytes) || before != after { + builder.partial = true; + } +} + +/// Decode a remote CLI export; anything unexpected fails closed. +pub(crate) fn decode_remote_ledger( + output: &str, + start_unix_ms: i64, + end_unix_ms: i64, +) -> Result { + if output.len() > MAX_IDENTIFIED_OUTPUT_BYTES { + return Err(IdentifiedUsageError::InvalidReport); + } + let ledger: CodexRequestLedger = + serde_json::from_str(output.trim()).map_err(|_| IdentifiedUsageError::InvalidReport)?; + ledger.validate(start_unix_ms, end_unix_ms)?; + Ok(ledger) +} + +/// Validated inputs for `cost --identified-only`. +pub(crate) struct IdentifiedUsageArgs { + pub(crate) identified_only: bool, + pub(crate) provider_is_codex_only: bool, + pub(crate) json: bool, + pub(crate) days: Option, + pub(crate) end_unix_ms: Option, + pub(crate) remote: Option, + pub(crate) period_set: bool, + pub(crate) group_by_set: bool, + pub(crate) summary_only: bool, + pub(crate) pretty: bool, +} + +/// The resolved `(days, start, end)` window, or the shared usage error. +pub(crate) fn identified_usage_window( + args: &IdentifiedUsageArgs, + now_unix_ms: i64, +) -> Result<(u32, i64, i64), &'static str> { + let days = args.days.unwrap_or(30); + let end = args.end_unix_ms.unwrap_or(now_unix_ms); + let valid = args.identified_only + && args.provider_is_codex_only + && args.json + && !args.period_set + && !args.group_by_set + && !args.summary_only; + if !valid { + return Err(IDENTIFIED_USAGE_ARGS_ERROR); + } + let start = identified_window_start(days, end).ok_or(IDENTIFIED_USAGE_ARGS_ERROR)?; + Ok((days, start, end)) +} + +pub(crate) async fn run_codex_identified_usage(args: &IdentifiedUsageArgs) -> anyhow::Result<()> { + let (days, start, end) = identified_usage_window(args, chrono::Utc::now().timestamp_millis()) + .map_err(anyhow::Error::msg)?; + let remote = args + .remote + .as_deref() + .map(RemoteSessionFetcher::validate_codex_cost_host) + .transpose() + .map_err(anyhow::Error::msg)?; + + let result = collect_identified_usage( + remote.as_deref(), + start, + end, + || load_local_ledger(start, end), + |host| async move { + let output = RemoteSessionFetcher::default() + .fetch_codex_identified_usage(&host, days, end) + .await + .map_err(|_| IdentifiedUsageError::Unavailable)?; + decode_remote_ledger(&output, start, end) + }, + ) + .await; + + println!( + "{}", + identified_usage_json(remote.is_some(), &result, args.pretty)? + ); + if !result.all_sources_available { + anyhow::bail!( + "Identified Codex usage is incomplete: a source was unavailable; available results were retained." + ); + } + Ok(()) +} + +/// A local-only success prints the bare ledger (the document a remote host +/// returns); any other outcome prints the source reports and subtotal. +pub(crate) fn identified_usage_json( + has_remote: bool, + result: &CodexIdentifiedResult, + pretty: bool, +) -> serde_json::Result { + let local_ledger = result.reports.first().and_then(|r| r.ledger.as_ref()); + let value = match (has_remote, local_ledger) { + (false, Some(ledger)) => serde_json::to_value(ledger)?, + _ => serde_json::to_value(result)?, + }; + if pretty { + serde_json::to_string_pretty(&value) + } else { + serde_json::to_string(&value) + } +} + +#[cfg(test)] +#[path = "identified_requests_tests.rs"] +mod tests; diff --git a/rust/src/codex_costs/identified_requests_tests.rs b/rust/src/codex_costs/identified_requests_tests.rs new file mode 100644 index 00000000000..f8b04764e26 --- /dev/null +++ b/rust/src/codex_costs/identified_requests_tests.rs @@ -0,0 +1,520 @@ +use super::*; +use serde_json::json; + +// 2023-11-14T22:13:20Z; its UTC day starts at 1_699_920_000_000. +const END: i64 = 1_700_000_000_000; +const DAY_START: i64 = 1_699_920_000_000; + +fn record(id_seed: &str, timestamp_unix_ms: i64, input: i64, output: i64) -> CodexRequestRecord { + CodexRequestRecord { + id: digest(&["codex", "thread", id_seed]), + timestamp_unix_ms, + input_tokens: input, + cache_read_tokens: 0, + output_tokens: output, + reasoning_tokens: Some(0), + } +} + +fn ledger(records: Vec) -> CodexRequestLedger { + let mut ledger = CodexRequestLedger::empty(DAY_START, END); + ledger.records = records; + ledger +} + +#[test] +fn digest_is_length_framed_sha256_hex() { + let id = digest(&["codex", "thread", "response"]); + assert!(is_digest(&id)); + // Framing keeps component boundaries distinct. + assert_ne!(id, digest(&["codex", "threadr", "esponse"])); + assert_eq!(id, digest(&["codex", "thread", "response"])); +} + +#[test] +fn window_starts_at_utc_midnight_days_before_end() { + assert_eq!(identified_window_start(1, END), Some(DAY_START)); + assert_eq!( + identified_window_start(30, END), + Some(DAY_START - 29 * DAY_MS) + ); + assert_eq!(identified_window_start(0, END), None); + assert_eq!(identified_window_start(366, END), None); + assert_eq!(identified_window_start(2, 0), None); + assert_eq!( + identified_window_start(1, MAX_IDENTIFIED_END_UNIX_MS + 1), + None + ); +} + +#[test] +fn merge_counts_identical_copies_once_and_excludes_conflicts() { + let local = ledger(vec![ + record("a", END - 10, 100, 10), + record("b", END - 10, 200, 20), + ]); + let remote = ledger(vec![ + record("a", END - 5, 100, 10), + // Contradictory copy of `b`: excluded throughout the merge. + record("b", END - 5, 999, 20), + record("c", END - 5, 50, 5), + ]); + let subtotal = CodexRequestLedger::merge(&[local, remote]).unwrap(); + assert_eq!(subtotal.request_count, 2); + assert_eq!(subtotal.total_tokens, 110 + 55); + assert_eq!(subtotal.conflict_count, 1); + assert!(subtotal.partial); + + let value = serde_json::to_value(&subtotal).unwrap(); + assert_eq!(value["kind"], "identified-request-subtotal"); + assert!(value["costUSD"].is_null()); + assert!(value.as_object().unwrap().contains_key("costUSD")); +} + +#[test] +fn merge_keeps_source_conflicts_after_a_clean_later_copy() { + let mut local = ledger(vec![]); + let conflicted = record("x", END - 1, 10, 1); + local.conflicting_ids = vec![conflicted.id.clone()]; + let remote = ledger(vec![conflicted]); + let subtotal = CodexRequestLedger::merge(&[local, remote]).unwrap(); + assert_eq!(subtotal.request_count, 0); + assert_eq!(subtotal.conflict_count, 1); + assert!(subtotal.partial); +} + +#[test] +fn clean_merge_is_complete() { + let subtotal = CodexRequestLedger::merge(&[ledger(vec![record("a", END, 1, 2)])]).unwrap(); + assert!(!subtotal.partial); + assert_eq!(subtotal.total_tokens, 3); +} + +#[test] +fn validate_rejects_invalid_ledgers() { + assert!( + ledger(vec![record("a", END, 1, 1)]) + .validate(DAY_START, END) + .is_ok() + ); + // Wrong window. + assert!(ledger(vec![]).validate(DAY_START + 1, END).is_err()); + // Out-of-window record. + assert!( + ledger(vec![record("a", END + 1, 1, 1)]) + .validate(DAY_START, END) + .is_err() + ); + // Cached above input, reasoning above output, negative tokens. + let mut bad = record("a", END, 1, 1); + bad.cache_read_tokens = 2; + assert!(ledger(vec![bad]).validate(DAY_START, END).is_err()); + let mut bad = record("a", END, 1, 1); + bad.reasoning_tokens = Some(2); + assert!(ledger(vec![bad]).validate(DAY_START, END).is_err()); + let mut bad = record("a", END, -1, 1); + bad.cache_read_tokens = -1; + assert!(ledger(vec![bad]).validate(DAY_START, END).is_err()); + // Non-digest identities. + let mut bad = record("a", END, 1, 1); + bad.id = "raw-session-id".to_string(); + assert!(ledger(vec![bad]).validate(DAY_START, END).is_err()); + // Aggregate overflow. + let huge = vec![ + record("a", END, i64::MAX / 2, 0), + record("b", END, i64::MAX / 2, 0), + record("c", END, i64::MAX / 2, 0), + ]; + assert!(ledger(huge).validate(DAY_START, END).is_err()); +} + +#[test] +fn ledger_json_uses_the_versioned_wire_names() { + let value = serde_json::to_value(ledger(vec![record("a", END, 3, 2)])).unwrap(); + for key in [ + "schemaVersion", + "provider", + "startUnixMs", + "endUnixMs", + "records", + "conflictingIDs", + "excludedCount", + "partial", + ] { + assert!(value.get(key).is_some(), "missing {key}"); + } + let row = &value["records"][0]; + for key in [ + "id", + "timestampUnixMs", + "inputTokens", + "cacheReadTokens", + "outputTokens", + "reasoningTokens", + ] { + assert!(row.get(key).is_some(), "missing {key}"); + } + let round_trip = decode_remote_ledger(&value.to_string(), DAY_START, END).unwrap(); + assert_eq!(round_trip.records.len(), 1); +} + +#[test] +fn remote_decoder_fails_closed() { + assert_eq!( + decode_remote_ledger("not json", DAY_START, END), + Err(IdentifiedUsageError::InvalidReport) + ); + let other_window = serde_json::to_string(&ledger(vec![])).unwrap(); + assert!(decode_remote_ledger(&other_window, DAY_START - DAY_MS, END).is_err()); + let oversized = " ".repeat(MAX_IDENTIFIED_OUTPUT_BYTES + 1); + assert!(decode_remote_ledger(&oversized, DAY_START, END).is_err()); +} + +fn line(value: serde_json::Value) -> String { + format!("{value}\n") +} + +fn usage(input: i64, cached: i64, output: i64) -> serde_json::Value { + json!({"input_tokens": input, "cached_input_tokens": cached, "output_tokens": output, "reasoning_output_tokens": 0}) +} + +fn usage_record( + thread: &str, + response: &str, + timestamp: &str, + request: serde_json::Value, + total: serde_json::Value, +) -> String { + line(json!({ + "type": "token_usage_record", + "timestamp": timestamp, + "payload": {"thread_id": thread, "response_id": response, "usage": request, "thread_token_usage": total} + })) +} + +fn session_file(root: &Path, name: &str, session_id: &str, body: &str) { + let dir = root.join("2023").join("11").join("14"); + fs::create_dir_all(&dir).unwrap(); + let meta = line(json!({"type": "session_meta", "payload": {"id": session_id}})); + fs::write(dir.join(name), format!("{meta}{body}")).unwrap(); +} + +#[test] +fn local_scan_identifies_owned_requests_and_reports_exclusions() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + let ts = "2023-11-14T12:00:00.000Z"; + let mut body = String::new(); + body += &line(json!({"type": "turn_context", "payload": {"model": "gpt-5", "turn_id": "t1"}})); + body += &usage_record("s1", "r1", ts, usage(100, 40, 10), usage(100, 40, 10)); + // Mirrored legacy counter for r1: not excluded. + body += &line(json!({ + "type": "event_msg", "timestamp": ts, + "payload": {"type": "token_count", "info": {"last_token_usage": usage(100, 40, 10), "total_token_usage": usage(100, 40, 10)}} + })); + // Verbatim replay of r1 counts once. + body += &usage_record("s1", "r1", ts, usage(100, 40, 10), usage(100, 40, 10)); + body += &usage_record("s1", "r2", ts, usage(50, 0, 5), usage(150, 40, 15)); + // Copied parent record: not owned, ignored. + body += &usage_record("parent", "r9", ts, usage(7, 0, 7), usage(7, 0, 7)); + // Outside the window: ignored. + body += &usage_record( + "s1", + "r3", + "2023-11-13T12:00:00Z", + usage(9, 0, 9), + usage(159, 40, 24), + ); + // Unmatched legacy usage: excluded. + body += &line(json!({ + "type": "event_msg", "timestamp": ts, + "payload": {"type": "token_count", "info": {"last_token_usage": usage(30, 0, 3), "total_token_usage": usage(500, 40, 50)}} + })); + session_file(&root, "rollout-2023-11-14T12-00-00-s1.jsonl", "s1", &body); + + let ledger = load_ledger_from_roots(&[root], DAY_START, END).unwrap(); + assert_eq!(ledger.records.len(), 2); + assert_eq!(ledger.excluded_count, 1); + assert!(ledger.conflicting_ids.is_empty()); + assert!(ledger.partial); + let first = ledger + .records + .iter() + .find(|row| row.id == digest(&["codex", "s1", "r1"])) + .unwrap(); + assert_eq!(first.input_tokens, 100); + assert_eq!(first.cache_read_tokens, 40); + assert_eq!(first.output_tokens, 10); + assert_eq!(first.timestamp_unix_ms, 1_699_963_200_000); + + let serialized = serde_json::to_string(&ledger).unwrap(); + assert!(!serialized.contains("s1")); + assert!(!serialized.contains("gpt-5")); +} + +#[test] +fn files_last_written_before_the_window_are_skipped() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + let ts = "2023-11-14T12:00:00Z"; + let body = usage_record("s1", "r1", ts, usage(10, 0, 1), usage(10, 0, 1)); + let name = "rollout-2023-11-14T12-00-00-s1.jsonl"; + session_file(&root, name, "s1", &body); + let path = root.join("2023").join("11").join("14").join(name); + let set_mtime = |unix_ms: i64| { + let time = std::time::UNIX_EPOCH + std::time::Duration::from_millis(unix_ms as u64); + fs::File::options() + .write(true) + .open(&path) + .unwrap() + .set_modified(time) + .unwrap(); + }; + + set_mtime(DAY_START - 2 * 24 * 60 * 60 * 1000); + let ledger = load_ledger_from_roots(std::slice::from_ref(&root), DAY_START, END).unwrap(); + assert!(ledger.records.is_empty()); + + // Inside the clock-skew margin the file is still scanned. + set_mtime(DAY_START - 60 * 60 * 1000); + let ledger = load_ledger_from_roots(&[root], DAY_START, END).unwrap(); + assert_eq!(ledger.records.len(), 1); +} + +#[test] +fn copied_parent_legacy_counters_are_not_excluded_from_a_fork() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + let ts = "2023-11-14T12:00:00Z"; + let mut body = String::new(); + body += &line(json!({"type": "turn_context", "payload": {"model": "gpt-5"}})); + body += &usage_record("parent", "r1", ts, usage(70, 0, 7), usage(70, 0, 7)); + body += &line(json!({ + "type": "event_msg", "timestamp": ts, + "payload": {"type": "token_count", "info": {"last_token_usage": usage(70, 0, 7), "total_token_usage": usage(70, 0, 7)}} + })); + body += &usage_record("child", "r2", ts, usage(5, 0, 1), usage(75, 0, 8)); + session_file( + &root, + "rollout-2023-11-14T12-00-00-child.jsonl", + "child", + &body, + ); + + let ledger = load_ledger_from_roots(&[root], DAY_START, END).unwrap(); + assert_eq!(ledger.records.len(), 1); + assert_eq!(ledger.excluded_count, 0); + assert!(!ledger.partial); +} + +#[test] +fn local_scan_marks_contradictions_rejections_and_foreign_routes() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + let ts = "2023-11-14T12:00:00Z"; + let mut body = String::new(); + body += &usage_record("s2", "r1", ts, usage(10, 0, 1), usage(10, 0, 1)); + body += &usage_record("s2", "r1", ts, usage(20, 0, 1), usage(20, 0, 1)); + // Rejected owned record with an identity: conflict + excluded. + body += &line(json!({ + "type": "token_usage_record", "timestamp": ts, + "payload": {"thread_id": "s2", "response_id": "r2", "usage": {"input_tokens": -1}} + })); + // Foreign-provider route is not Codex subscription usage. + body += &line(json!({ + "type": "token_usage_record", "timestamp": ts, + "payload": {"thread_id": "s2", "response_id": "r3", "model": "deepseek/deepseek-chat", + "usage": usage(5, 0, 5), "thread_token_usage": usage(25, 0, 7)} + })); + session_file(&root, "rollout-s2.jsonl", "s2", &body); + + let ledger = load_ledger_from_roots(&[root], DAY_START, END).unwrap(); + assert!( + ledger + .records + .iter() + .all(|row| row.id != digest(&["codex", "s2", "r3"])) + ); + assert_eq!(ledger.conflicting_ids.len(), 2); + assert_eq!(ledger.excluded_count, 1); + assert!(ledger.partial); + let subtotal = CodexRequestLedger::merge(&[ledger]).unwrap(); + assert_eq!(subtotal.request_count, 0); +} + +#[test] +fn malformed_lines_mark_the_ledger_partial() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + let body = format!( + "{{not json\n{}", + usage_record( + "s3", + "r1", + "2023-11-14T12:00:00Z", + usage(1, 0, 1), + usage(1, 0, 1) + ) + ); + session_file(&root, "rollout-s3.jsonl", "s3", &body); + let ledger = load_ledger_from_roots(&[root], DAY_START, END).unwrap(); + assert_eq!(ledger.records.len(), 1); + assert_eq!(ledger.excluded_count, 0); + assert!(ledger.partial); +} + +#[test] +fn readable_empty_history_is_available_and_missing_history_is_not() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("sessions"); + fs::create_dir_all(&root).unwrap(); + let ledger = load_ledger_from_roots( + &[root, temp.path().join("archived_sessions")], + DAY_START, + END, + ) + .unwrap(); + assert!(ledger.records.is_empty()); + assert!(!ledger.partial); + + assert_eq!( + load_ledger_from_roots(&[temp.path().join("missing")], DAY_START, END), + Err(IdentifiedUsageError::Unavailable) + ); +} + +#[tokio::test] +async fn remote_failure_retains_the_local_subtotal_and_marks_partial() { + let result = collect_identified_usage( + Some("build-host"), + DAY_START, + END, + || Ok(ledger(vec![record("a", END, 4, 1)])), + |_| async { Err(IdentifiedUsageError::Unavailable) }, + ) + .await; + assert!(!result.all_sources_available); + assert!(result.partial); + assert_eq!(result.reports.len(), 2); + assert_eq!(result.reports[1].host, "build-host"); + assert_eq!( + result.reports[1].error.as_deref(), + Some(REMOTE_IDENTIFIED_UNAVAILABLE) + ); + let subtotal = result.subtotal.as_ref().unwrap(); + assert_eq!(subtotal.total_tokens, 5); + assert!(!subtotal.partial); + + let json: serde_json::Value = + serde_json::from_str(&identified_usage_json(true, &result, false).unwrap()).unwrap(); + assert_eq!(json["allSourcesAvailable"], false); + assert!(json["reports"][1].get("ledger").is_none()); + assert!(json["subtotal"]["costUSD"].is_null()); +} + +#[tokio::test] +async fn remote_ledger_deduplicates_with_local() { + let result = collect_identified_usage( + Some("build-host"), + DAY_START, + END, + || Ok(ledger(vec![record("a", END, 4, 1)])), + |_| async { Ok(ledger(vec![record("a", END, 4, 1), record("b", END, 2, 2)])) }, + ) + .await; + assert!(result.all_sources_available); + assert!(!result.partial); + let subtotal = result.subtotal.unwrap(); + assert_eq!(subtotal.request_count, 2); + assert_eq!(subtotal.total_tokens, 9); +} + +#[tokio::test] +async fn overflowing_remote_union_is_rejected_while_local_remains() { + let result = collect_identified_usage( + Some("build-host"), + DAY_START, + END, + || Ok(ledger(vec![record("a", END, i64::MAX - 10, 0)])), + |_| async { Ok(ledger(vec![record("b", END, 100, 0)])) }, + ) + .await; + assert!(!result.all_sources_available); + assert_eq!(result.subtotal.unwrap().total_tokens, i64::MAX - 10); +} + +#[tokio::test] +async fn local_only_success_prints_the_bare_ledger() { + let result = collect_identified_usage( + None, + DAY_START, + END, + || Ok(ledger(vec![])), + |_| async { Err(IdentifiedUsageError::Unavailable) }, + ) + .await; + let json: serde_json::Value = + serde_json::from_str(&identified_usage_json(false, &result, false).unwrap()).unwrap(); + assert_eq!(json["schemaVersion"], 1); + assert!(json.get("reports").is_none()); + + let failed = collect_identified_usage( + None, + DAY_START, + END, + || Err(IdentifiedUsageError::Unavailable), + |_| async { Err(IdentifiedUsageError::Unavailable) }, + ) + .await; + assert!(failed.partial); + assert!(failed.subtotal.is_none()); + let json: serde_json::Value = + serde_json::from_str(&identified_usage_json(false, &failed, false).unwrap()).unwrap(); + assert_eq!(json["reports"][0]["error"], LOCAL_IDENTIFIED_UNAVAILABLE); + assert!(json.get("subtotal").is_none()); +} + +fn valid_args() -> IdentifiedUsageArgs { + IdentifiedUsageArgs { + identified_only: true, + provider_is_codex_only: true, + json: true, + days: None, + end_unix_ms: Some(END), + remote: None, + period_set: false, + group_by_set: false, + summary_only: false, + pretty: false, + } +} + +#[test] +fn identified_args_require_explicit_codex_json_without_other_modes() { + assert_eq!( + identified_usage_window(&valid_args(), 0), + Ok((30, DAY_START - 29 * DAY_MS, END)) + ); + let mut now = valid_args(); + now.end_unix_ms = None; + assert_eq!(identified_usage_window(&now, END).unwrap().2, END); + + let rejected: [fn(&mut IdentifiedUsageArgs); 8] = [ + |a| a.identified_only = false, + |a| a.provider_is_codex_only = false, + |a| a.json = false, + |a| a.period_set = true, + |a| a.group_by_set = true, + |a| a.summary_only = true, + |a| a.days = Some(366), + |a| a.end_unix_ms = Some(MAX_IDENTIFIED_END_UNIX_MS + 1), + ]; + for mutate in rejected { + let mut args = valid_args(); + mutate(&mut args); + assert_eq!( + identified_usage_window(&args, END), + Err(IDENTIFIED_USAGE_ARGS_ERROR) + ); + } +}