#![deny(clippy::all)]
mod aggregator;
pub mod bucket_tz;
mod cc_mirror;
pub mod clients;
pub mod content_extractor;
pub mod fs_atomic;
pub mod mcp;
mod message_cache;
pub mod model_alias;
pub mod model_identity;
pub mod opencode_model_name;
mod parser;
pub mod paths;
pub mod pricing;
mod provider_identity;
pub mod scanner;
pub mod sessionize;
pub mod sessions;
mod workspace;
pub use workspace::{
workspace_bucket, workspace_label_overrides, WorkspaceLabeler, WorktreeRollup,
};
pub mod tui_signal;
#[cfg(test)]
mod upstream_sync_tests;
pub mod wiki;
pub use aggregator::*;
pub use bucket_tz::{bucket_timezone, parse_bucket_timezone, set_bucket_timezone, BucketTimezone};
pub use clients::{ClientCounts, ClientDef, ClientId, PathRoot};
pub use model_alias::ModelAliasMap;
pub use model_identity::resolve_model_id;
pub use parser::*;
pub use scanner::*;
pub use sessionize::{
compute_daily_active_time, compute_time_metrics, sessionize, SessionInterval, TimeMetrics,
DEFAULT_IDLE_GAP_MS,
};
pub use sessions::{CostSource, UnifiedMessage};
use rayon::prelude::*;
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Instant;
pub(crate) fn strip_parenthesized_reasoning_tier(model_id: &str) -> Option<&str> {
let without_closing_paren = model_id.strip_suffix(')')?;
let (base_model, tier) = without_closing_paren.rsplit_once('(')?;
if base_model.is_empty() || base_model.trim() != base_model {
return None;
}
if !matches!(
tier,
"minimal" | "low" | "medium" | "high" | "xhigh" | "auto" | "none"
) {
return None;
}
Some(base_model)
}
pub fn canonical_model_id(model_id: &str) -> String {
normalize_syntactic(model_id)
}
pub fn canonical_model_id_for_client(client: &str, model_id: &str) -> String {
canonical_model_id(&model_identity::resolve_model_id(client, model_id))
}
pub fn normalize_model_for_grouping(model_id: &str) -> String {
model_alias::global().apply(normalize_syntactic(model_id))
}
pub fn model_name_for_grouping(client: &str, provider_id: &str, model_id: &str) -> String {
let resolved_model_id = model_identity::resolve_model_id(client, model_id);
let fallback = normalize_model_for_grouping(&resolved_model_id);
if client == "opencode" {
opencode_model_name::global()
.display_name(provider_id, &resolved_model_id)
.map(str::to_string)
.unwrap_or(fallback)
} else {
fallback
}
}
pub(crate) fn normalize_syntactic(model_id: &str) -> String {
let mut name = model_id.to_lowercase();
if let Some(base_model) = strip_parenthesized_reasoning_tier(&name) {
name = base_model.to_string();
}
if name.len() > 9 {
let potential_date = &name[name.len() - 8..];
if potential_date.chars().all(|c| c.is_ascii_digit())
&& name.as_bytes()[name.len() - 9] == b'-'
{
name = name[..name.len() - 9].to_string();
}
}
if name.contains("claude") {
let chars: Vec<char> = name.chars().collect();
let mut result = String::with_capacity(name.len());
for i in 0..chars.len() {
if chars[i] == '.'
&& i > 0
&& i < chars.len() - 1
&& chars[i - 1].is_ascii_digit()
&& chars[i + 1].is_ascii_digit()
{
result.push('-');
} else {
result.push(chars[i]);
}
}
name = result;
}
if let Some(canonical) = normalize_anthropic_prefixed_claude_model(&name) {
name = canonical;
}
name
}
fn normalize_anthropic_prefixed_claude_model(model_id: &str) -> Option<String> {
let rest = model_id.strip_prefix("anthropic/claude-")?;
let mut parts = rest.split('-');
let major = parts.next()?;
let minor = parts.next()?;
let family = parts.next()?;
if parts.next().is_some() {
return None;
}
if !matches!(family, "opus" | "sonnet" | "haiku") {
return None;
}
Some(format!("claude-{family}-{major}-{minor}"))
}
fn retain_for_requested_clients(
client: &str,
model_id: &str,
provider_id: &str,
requested: &HashSet<&str>,
) -> bool {
requested.contains(client)
|| (requested.contains("claude") && client.starts_with("cc-mirror/"))
|| (requested.contains("gjc") && client.eq_ignore_ascii_case("9router"))
|| (requested.contains("synthetic")
&& sessions::synthetic::matches_synthetic_filter(client, model_id, provider_id))
}
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize)]
pub enum GroupBy {
Model,
#[default]
ClientModel,
ClientProviderModel,
WorkspaceModel,
Session,
ClientSession,
}
impl std::fmt::Display for GroupBy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
GroupBy::Model => write!(f, "model"),
GroupBy::ClientModel => write!(f, "client,model"),
GroupBy::ClientProviderModel => write!(f, "client,provider,model"),
GroupBy::WorkspaceModel => write!(f, "workspace,model"),
GroupBy::Session => write!(f, "session,model"),
GroupBy::ClientSession => write!(f, "client,session,model"),
}
}
}
impl std::str::FromStr for GroupBy {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let normalized: String = s.split(',').map(|p| p.trim()).collect::<Vec<_>>().join(",");
match normalized.to_lowercase().as_str() {
"model" => Ok(GroupBy::Model),
"client,model" | "client-model" => Ok(GroupBy::ClientModel),
"client,provider,model" | "client-provider-model" => Ok(GroupBy::ClientProviderModel),
"workspace,model" | "workspace-model" => Ok(GroupBy::WorkspaceModel),
"session" | "session,model" | "session-model" => Ok(GroupBy::Session),
"client,session" | "client-session" | "client,session,model" | "client-session-model" => {
Ok(GroupBy::ClientSession)
}
_ => Err(format!(
"Invalid group-by value: '{}'. Valid options: model, client,model, client,provider,model, workspace,model, session,model, client,session,model",
s
)),
}
}
}
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct TokenBreakdown {
pub input: i64,
pub output: i64,
pub cache_read: i64,
pub cache_write: i64,
pub reasoning: i64,
}
impl TokenBreakdown {
pub fn add_assign_saturating(&mut self, other: &Self) {
self.input = self.input.saturating_add(other.input);
self.output = self.output.saturating_add(other.output);
self.cache_read = self.cache_read.saturating_add(other.cache_read);
self.cache_write = self.cache_write.saturating_add(other.cache_write);
self.reasoning = self.reasoning.saturating_add(other.reasoning);
}
pub fn total(&self) -> i64 {
self.input
.saturating_add(self.output)
.saturating_add(self.cache_read)
.saturating_add(self.cache_write)
.saturating_add(self.reasoning)
}
}
impl std::ops::AddAssign<&TokenBreakdown> for TokenBreakdown {
fn add_assign(&mut self, other: &TokenBreakdown) {
self.add_assign_saturating(other);
}
}
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ModelPerformance {
#[serde(rename = "msPer1KTokens")]
pub ms_per_1k_tokens: Option<f64>,
pub total_duration_ms: i64,
pub timed_tokens: i64,
pub sample_count: i32,
pub token_coverage: f64,
}
impl ModelPerformance {
pub fn record_message(&mut self, token_total: i64, duration_ms: Option<i64>) {
let Some(duration_ms) = duration_ms else {
return;
};
if duration_ms <= 0 || token_total <= 0 {
return;
}
self.total_duration_ms = self.total_duration_ms.saturating_add(duration_ms);
self.timed_tokens = self.timed_tokens.saturating_add(token_total);
self.sample_count = self.sample_count.saturating_add(1);
}
pub fn finalize(&mut self, total_tokens: i64) {
self.ms_per_1k_tokens = if self.timed_tokens > 0 && self.total_duration_ms > 0 {
Some(self.total_duration_ms as f64 * 1000.0 / self.timed_tokens as f64)
} else {
None
};
self.token_coverage = if total_tokens > 0 {
(self.timed_tokens as f64 / total_tokens as f64).clamp(0.0, 1.0)
} else {
0.0
};
}
pub fn from_totals(total_duration_ms: i64, timed_tokens: i64, sample_count: i32) -> Self {
let mut performance = Self {
total_duration_ms,
timed_tokens,
sample_count,
..Self::default()
};
performance.finalize(timed_tokens);
performance
}
}
#[derive(Debug, Clone)]
pub struct ParsedMessage {
pub client: String,
pub model_id: String,
pub provider_id: String,
pub session_id: String,
pub workspace_key: Option<String>,
pub workspace_label: Option<String>,
pub timestamp: i64,
pub date: String,
pub input: i64,
pub output: i64,
pub cache_read: i64,
pub cache_write: i64,
pub reasoning: i64,
pub duration_ms: Option<i64>,
pub message_count: i32,
pub agent: Option<String>,
pub cost: f64,
pub cost_source: CostSource,
}
pub struct ParsedMessages {
pub messages: Vec<ParsedMessage>,
pub counts: ClientCounts,
pub processing_time_ms: u32,
}
impl Clone for ParsedMessages {
fn clone(&self) -> Self {
let mut counts = ClientCounts::new();
for client in ClientId::iter() {
counts.set(client, self.counts.get(client));
}
Self {
messages: self.messages.clone(),
counts,
processing_time_ms: self.processing_time_ms,
}
}
}
impl std::fmt::Debug for ParsedMessages {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut debug = f.debug_struct("ParsedMessages");
debug.field("messages", &self.messages);
for client in ClientId::iter() {
debug.field(client.as_str(), &self.counts.get(client));
}
debug.field("processing_time_ms", &self.processing_time_ms);
debug.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct DevinDesktopLookupSnapshot {
db_paths: Vec<PathBuf>,
related_files: Vec<message_cache::RelatedFileFingerprint>,
}
type DevinDesktopLookupCache = Mutex<
HashMap<DevinDesktopLookupSnapshot, Arc<OnceLock<sessions::devin::DevinDesktopSessionLookup>>>,
>;
fn devin_desktop_lookup_cell_for_snapshot(
lookup_cache: &DevinDesktopLookupCache,
db_paths: &[PathBuf],
fingerprint: &message_cache::SourceFingerprint,
) -> Arc<OnceLock<sessions::devin::DevinDesktopSessionLookup>> {
let snapshot = DevinDesktopLookupSnapshot {
db_paths: db_paths.to_vec(),
related_files: fingerprint.related_files.clone(),
};
let mut lookups = lookup_cache
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
Arc::clone(
lookups
.entry(snapshot)
.or_insert_with(|| Arc::new(OnceLock::new())),
)
}
#[derive(Debug, Clone, Default)]
pub struct LocalParseOptions {
pub home_dir: Option<String>,
pub use_env_roots: bool,
pub clients: Option<Vec<String>>,
pub since: Option<String>,
pub until: Option<String>,
pub year: Option<String>,
pub scanner_settings: scanner::ScannerSettings,
}
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct DailyTotals {
pub tokens: i64,
pub cost: f64,
pub messages: i32,
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ClientContribution {
pub client: String,
pub model_id: String,
pub provider_id: String,
pub tokens: TokenBreakdown,
pub cost: f64,
pub messages: i32,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct DailyContribution {
pub date: String,
pub totals: DailyTotals,
pub intensity: u8,
pub token_breakdown: TokenBreakdown,
pub clients: Vec<ClientContribution>,
#[serde(skip_serializing_if = "Option::is_none")]
pub active_time_ms: Option<i64>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
pub struct SessionContribution {
pub session_id: String,
pub client: String,
pub provider: String,
pub model: String,
pub totals: DailyTotals,
pub token_breakdown: TokenBreakdown,
pub clients: Vec<ClientContribution>,
pub first_seen: i64,
pub last_seen: i64,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct YearSummary {
pub year: String,
pub total_tokens: i64,
pub total_cost: f64,
pub range_start: String,
pub range_end: String,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct DataSummary {
pub total_tokens: i64,
pub total_cost: f64,
pub total_days: i32,
pub active_days: i32,
pub average_per_day: f64,
pub max_cost_in_single_day: f64,
pub clients: Vec<String>,
pub models: Vec<String>,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct GraphMeta {
pub generated_at: String,
pub version: String,
pub date_range_start: String,
pub date_range_end: String,
pub processing_time_ms: u32,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct GraphResult {
pub meta: GraphMeta,
pub summary: DataSummary,
pub years: Vec<YearSummary>,
pub contributions: Vec<DailyContribution>,
#[serde(skip_serializing_if = "Option::is_none")]
pub time_metrics: Option<sessionize::TimeMetrics>,
}
#[derive(Debug, Clone, Default)]
pub struct ReportOptions {
pub home_dir: Option<String>,
pub use_env_roots: bool,
pub clients: Option<Vec<String>>,
pub since: Option<String>,
pub until: Option<String>,
pub year: Option<String>,
pub group_by: GroupBy,
pub scanner_settings: scanner::ScannerSettings,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct ModelUsage {
pub client: String,
pub merged_clients: Option<String>,
pub workspace_key: Option<String>,
pub workspace_label: Option<String>,
pub session_id: Option<String>,
pub model: String,
pub provider: String,
pub input: i64,
pub output: i64,
pub cache_read: i64,
pub cache_write: i64,
pub reasoning: i64,
pub message_count: i32,
pub cost: f64,
pub performance: ModelPerformance,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct MonthlyUsage {
pub month: String,
pub models: Vec<String>,
pub input: i64,
pub output: i64,
pub cache_read: i64,
pub cache_write: i64,
pub message_count: i32,
pub cost: f64,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct ModelReport {
pub entries: Vec<ModelUsage>,
pub total_input: i64,
pub total_output: i64,
pub total_cache_read: i64,
pub total_cache_write: i64,
pub total_messages: i32,
pub total_cost: f64,
pub processing_time_ms: u32,
}
const UNKNOWN_WORKSPACE_LABEL: &str = "Unknown workspace";
const UNKNOWN_WORKSPACE_GROUP_KEY: &str = "\0unknown-workspace";
#[derive(Debug, Clone, serde::Serialize)]
pub struct MonthlyReport {
pub entries: Vec<MonthlyUsage>,
pub total_cost: f64,
pub processing_time_ms: u32,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct HourlyUsage {
pub hour: String,
pub clients: Vec<String>,
pub models: Vec<String>,
pub input: i64,
pub output: i64,
pub cache_read: i64,
pub cache_write: i64,
pub message_count: i32,
pub turn_count: i32,
pub reasoning: i64,
pub cost: f64,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct HourlyReport {
pub entries: Vec<HourlyUsage>,
pub total_cost: f64,
pub processing_time_ms: u32,
}
pub fn get_home_dir_string(home_dir_option: &Option<String>) -> Result<String, String> {
home_dir_option
.clone()
.or_else(|| std::env::var("HOME").ok())
.or_else(|| dirs::home_dir().map(|p| p.to_string_lossy().into_owned()))
.ok_or_else(|| {
"HOME directory not specified and could not determine home directory".to_string()
})
}
#[allow(dead_code)]
fn parse_all_messages_with_pricing(
home_dir: &str,
clients: &[String],
pricing: Option<&pricing::PricingService>,
) -> Vec<UnifiedMessage> {
parse_all_messages_with_pricing_with_env_strategy(
home_dir,
clients,
pricing,
true,
&scanner::ScannerSettings::default(),
)
}
fn parse_all_messages_with_pricing_with_env_strategy(
home_dir: &str,
clients: &[String],
pricing: Option<&pricing::PricingService>,
use_env_roots: bool,
scanner_settings: &scanner::ScannerSettings,
) -> Vec<UnifiedMessage> {
#[derive(Debug)]
struct CachedParseOutcome {
messages: Vec<UnifiedMessage>,
cache_entry: Option<message_cache::CachedSourceEntry>,
invalidate_cache: bool,
}
fn apply_pricing_to_messages(
messages: &mut [UnifiedMessage],
pricing: Option<&pricing::PricingService>,
) {
for message in messages {
message.refresh_derived_fields();
apply_pricing_if_available(message, pricing);
}
}
fn cached_messages(
cached: &message_cache::CachedSourceEntry,
pricing: Option<&pricing::PricingService>,
) -> Vec<UnifiedMessage> {
let mut messages = cached.messages.clone();
apply_pricing_to_messages(&mut messages, pricing);
messages
}
type CodexSourceOutcome = (CachedParseOutcome, sessions::codex::CodexTurnCoverage);
fn parse_full_log_source(
path: &Path,
pricing: Option<&pricing::PricingService>,
is_headless: bool,
) -> CodexSourceOutcome {
let fallback_timestamp = sessions::utils::file_modified_timestamp_ms(path);
let parsed = sessions::codex::parse_codex_file_incremental(
path,
0,
sessions::codex::CodexParseState::default(),
);
let turn_coverage = parsed.state.turn_coverage.clone();
let messages = finalize_codex_messages(
parsed.messages.clone(),
pricing,
is_headless,
&parsed.fallback_timestamp_indices,
fallback_timestamp,
);
if !parsed.parse_succeeded || parsed.unresolved_model_events {
return (
CachedParseOutcome {
messages,
cache_entry: None,
invalidate_cache: false,
},
turn_coverage,
);
}
let cache_entry = build_codex_cache_entry(
path,
parsed.messages,
parsed.consumed_offset,
parsed.state,
parsed.fallback_timestamp_indices,
);
(
CachedParseOutcome {
messages,
cache_entry,
invalidate_cache: false,
},
turn_coverage,
)
}
fn finalize_codex_messages(
mut messages: Vec<UnifiedMessage>,
pricing: Option<&pricing::PricingService>,
is_headless: bool,
fallback_timestamp_indices: &[usize],
fallback_timestamp: i64,
) -> Vec<UnifiedMessage> {
for index in fallback_timestamp_indices {
if let Some(message) = messages.get_mut(*index) {
message.set_timestamp(fallback_timestamp);
}
}
apply_pricing_to_messages(&mut messages, pricing);
for message in &mut messages {
apply_headless_agent(message, is_headless);
}
messages
}
fn build_codex_cache_entry(
path: &Path,
raw_messages: Vec<UnifiedMessage>,
consumed_offset: u64,
state: sessions::codex::CodexParseState,
fallback_timestamp_indices: Vec<usize>,
) -> Option<message_cache::CachedSourceEntry> {
let fingerprint = message_cache::SourceFingerprint::from_path(path)?;
if fingerprint.size != consumed_offset {
return None;
}
let codex_incremental = message_cache::build_codex_incremental_cache_with_prefix_hash(
path,
consumed_offset,
state,
fingerprint.content_hash,
)?;
Some(message_cache::CachedSourceEntry::new(
message_cache::CacheIdentity::for_client(ClientId::Codex),
path,
fingerprint,
raw_messages,
fallback_timestamp_indices,
Some(codex_incremental),
))
}
fn load_or_parse_source_with_fingerprint_and_policy<F, FingerprintFn>(
identity: message_cache::CacheIdentity,
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
fingerprint_from_path: FingerprintFn,
parse: F,
) -> CachedParseOutcome
where
F: Fn(&Path, Option<&message_cache::SourceFingerprint>) -> (Vec<UnifiedMessage>, bool),
FingerprintFn: Fn(
&Path,
Option<&message_cache::SourceFingerprint>,
) -> Option<message_cache::FingerprintStatus>,
{
let cached = source_cache.get(identity, path);
let Some(fingerprint_status) =
fingerprint_from_path(path, cached.map(|entry| &entry.fingerprint))
else {
let (mut messages, _) = parse(path, None);
apply_pricing_to_messages(&mut messages, pricing);
return CachedParseOutcome {
messages,
cache_entry: None,
invalidate_cache: false,
};
};
let fingerprint = match fingerprint_status {
message_cache::FingerprintStatus::Unchanged => {
let Some(cached) = cached else {
unreachable!("an uncached source always builds a complete fingerprint")
};
if !cached.messages.is_empty() {
return CachedParseOutcome {
messages: cached_messages(cached, pricing),
cache_entry: None,
invalidate_cache: false,
};
}
cached.fingerprint.clone()
}
message_cache::FingerprintStatus::Changed(fingerprint) => fingerprint,
};
if let Some(cached) = cached {
if cached.fingerprint == fingerprint && !cached.messages.is_empty() {
return CachedParseOutcome {
messages: cached_messages(cached, pricing),
cache_entry: None,
invalidate_cache: false,
};
}
}
let (mut messages, cacheable) = parse(path, Some(&fingerprint));
let cache_entry = if messages.is_empty() || !cacheable {
None
} else {
Some(message_cache::CachedSourceEntry::new(
identity,
path,
fingerprint,
messages.clone(),
Vec::new(),
None,
))
};
apply_pricing_to_messages(&mut messages, pricing);
CachedParseOutcome {
messages,
cache_entry,
invalidate_cache: !cacheable,
}
}
fn load_or_parse_source_with_fingerprint<F, FingerprintFn>(
identity: message_cache::CacheIdentity,
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
fingerprint_from_path: FingerprintFn,
parse: F,
) -> CachedParseOutcome
where
F: Fn(&Path) -> Vec<UnifiedMessage>,
FingerprintFn: Fn(
&Path,
Option<&message_cache::SourceFingerprint>,
) -> Option<message_cache::FingerprintStatus>,
{
load_or_parse_source_with_fingerprint_and_policy(
identity,
path,
source_cache,
pricing,
fingerprint_from_path,
|path, _| (parse(path), true),
)
}
fn load_or_parse_source_with_fingerprint_context<F, FingerprintFn>(
identity: message_cache::CacheIdentity,
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
fingerprint_from_path: FingerprintFn,
parse: F,
) -> CachedParseOutcome
where
F: Fn(&Path, Option<&message_cache::SourceFingerprint>) -> Vec<UnifiedMessage>,
FingerprintFn: Fn(
&Path,
Option<&message_cache::SourceFingerprint>,
) -> Option<message_cache::FingerprintStatus>,
{
load_or_parse_source_with_fingerprint_and_policy(
identity,
path,
source_cache,
pricing,
fingerprint_from_path,
|path, fingerprint| (parse(path, fingerprint), true),
)
}
fn load_or_parse_source<F>(
identity: message_cache::CacheIdentity,
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
parse: F,
) -> CachedParseOutcome
where
F: Fn(&Path) -> Vec<UnifiedMessage>,
{
load_or_parse_source_with_fingerprint(
identity,
path,
source_cache,
pricing,
message_cache::SourceFingerprint::check_path_samples_only,
parse,
)
}
fn load_or_parse_opencode_sqlite_source(
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
) -> CachedParseOutcome {
let identity = message_cache::CacheIdentity::for_client(ClientId::OpenCode);
fn finish(
identity: message_cache::CacheIdentity,
path: &Path,
fingerprint: Option<message_cache::SourceFingerprint>,
scan: sessions::opencode_schema::OpenCodeSchemaScan,
pricing: Option<&pricing::PricingService>,
) -> CachedParseOutcome {
let mut messages = scan.messages;
let cache_entry = match fingerprint {
Some(fingerprint) if !messages.is_empty() => Some(
message_cache::CachedSourceEntry::new(
identity,
path,
fingerprint,
messages.clone(),
Vec::new(),
None,
)
.with_opencode_incremental(scan.incremental),
),
_ => None,
};
apply_pricing_to_messages(&mut messages, pricing);
CachedParseOutcome {
messages,
cache_entry,
invalidate_cache: false,
}
}
let mut cached = source_cache.get(identity, path).cloned();
let Some(fingerprint_status) = message_cache::SourceFingerprint::check_sqlite_path(
path,
cached.as_ref().map(|entry| &entry.fingerprint),
) else {
return finish(
identity,
path,
None,
sessions::opencode::scan_opencode_sqlite(path),
pricing,
);
};
let fingerprint = match fingerprint_status {
message_cache::FingerprintStatus::Unchanged => {
let Some(entry) = cached.take() else {
unreachable!("an uncached source always builds a complete fingerprint")
};
if !entry.messages.is_empty() {
return CachedParseOutcome {
messages: cached_messages(&entry, pricing),
cache_entry: None,
invalidate_cache: false,
};
}
let fingerprint = entry.fingerprint.clone();
cached = Some(entry);
fingerprint
}
message_cache::FingerprintStatus::Changed(fingerprint) => fingerprint,
};
if let Some(entry) = cached {
if entry.fingerprint == fingerprint && !entry.messages.is_empty() {
return CachedParseOutcome {
messages: cached_messages(&entry, pricing),
cache_entry: None,
invalidate_cache: false,
};
}
if let (Some(state), false) = (
entry.opencode_incremental.as_ref(),
entry.messages.is_empty(),
) {
let state = state.clone();
if let Some(scan) =
sessions::opencode::rescan_opencode_sqlite(path, &state, entry.messages)
{
return finish(identity, path, Some(fingerprint), scan, pricing);
}
}
}
finish(
identity,
path,
Some(fingerprint),
sessions::opencode::scan_opencode_sqlite(path),
pricing,
)
}
fn load_or_parse_sqlite_source<F>(
identity: message_cache::CacheIdentity,
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
parse: F,
) -> CachedParseOutcome
where
F: Fn(&Path) -> Vec<UnifiedMessage>,
{
load_or_parse_source_with_fingerprint(
identity,
path,
source_cache,
pricing,
message_cache::SourceFingerprint::check_sqlite_path,
parse,
)
}
fn load_or_parse_codex_source(
path: &Path,
source_cache: &message_cache::SourceMessageCache,
pricing: Option<&pricing::PricingService>,
headless_roots: &[PathBuf],
) -> CodexSourceOutcome {
let identity = message_cache::CacheIdentity::for_client(ClientId::Codex);
let is_headless = is_headless_path(path, headless_roots);
let cached = source_cache.get(identity, path).cloned();
if cached.is_none() {
return parse_full_log_source(path, pricing, is_headless);
}
let Some(fingerprint_status) = message_cache::SourceFingerprint::check_path(
path,
cached.as_ref().map(|entry| &entry.fingerprint),
) else {
return parse_full_log_source(path, pricing, is_headless);
};
let fingerprint = match fingerprint_status {
message_cache::FingerprintStatus::Unchanged => cached
.as_ref()
.expect("an uncached source always builds a complete fingerprint")
.fingerprint
.clone(),
message_cache::FingerprintStatus::Changed(fingerprint) => fingerprint,
};
let fallback_timestamp = sessions::utils::file_modified_timestamp_ms(path);
if let Some(cached) = cached {
let reparse_from_start = |invalidate_cache: bool| {
let (mut outcome, turn_coverage) =
parse_full_log_source(path, pricing, is_headless);
outcome.invalidate_cache = invalidate_cache && outcome.cache_entry.is_none();
(outcome, turn_coverage)
};
if cached.fingerprint == fingerprint {
if message_cache::codex_cache_entry_matches_fingerprint(&cached, &fingerprint) {
let turn_coverage = cached
.codex_incremental
.as_ref()
.map(|incremental| incremental.state.turn_coverage.clone())
.unwrap_or_default();
return (
CachedParseOutcome {
messages: finalize_codex_messages(
cached.messages,
pricing,
is_headless,
&cached.fallback_timestamp_indices,
fallback_timestamp,
),
cache_entry: None,
invalidate_cache: false,
},
turn_coverage,
);
}
return reparse_from_start(true);
}
if let Some(codex_incremental) = cached.codex_incremental.as_ref() {
if fingerprint.size > codex_incremental.consumed_offset
&& message_cache::codex_prefix_matches(path, codex_incremental)
{
let parsed = sessions::codex::parse_codex_file_incremental(
path,
codex_incremental.consumed_offset,
codex_incremental.state.clone(),
);
if parsed.parse_succeeded && !parsed.unresolved_model_events {
let mut raw_messages = cached.messages.clone();
let mut fallback_timestamp_indices =
cached.fallback_timestamp_indices.clone();
let existing_len = raw_messages.len();
fallback_timestamp_indices.extend(
parsed
.fallback_timestamp_indices
.iter()
.map(|index| existing_len + index),
);
raw_messages.extend(parsed.messages);
let turn_coverage = parsed.state.turn_coverage.clone();
let cache_entry = build_codex_cache_entry(
path,
raw_messages.clone(),
parsed.consumed_offset,
parsed.state,
fallback_timestamp_indices.clone(),
);
if let Some(cache_entry) = cache_entry {
let messages = finalize_codex_messages(
raw_messages,
pricing,
is_headless,
&fallback_timestamp_indices,
fallback_timestamp,
);
return (
CachedParseOutcome {
messages,
cache_entry: Some(cache_entry),
invalidate_cache: false,
},
turn_coverage,
);
}
}
}
}
return reparse_from_start(true);
}
unreachable!("uncached Codex sources return before fingerprint validation")
}
let scan_result = scanner::scan_all_clients_with_scanner_settings(
home_dir,
clients,
use_env_roots,
scanner_settings,
);
let headless_roots = scanner::headless_roots_with_env_strategy(home_dir, use_env_roots);
let mut source_cache = message_cache::SourceMessageCache::load();
source_cache.prune_missing_files();
let mut all_messages: Vec<UnifiedMessage> = Vec::new();
let include_all = clients.is_empty();
let requested: HashSet<&str> = clients.iter().map(String::as_str).collect();
let include_synthetic = include_all || clients.iter().any(|c| c == "synthetic");
let include_devin_cli = include_synthetic || clients.iter().any(|c| c == "devin-cli");
let include_devin_desktop = include_synthetic || clients.iter().any(|c| c == "devin-desktop");
let mut opencode_seen: HashSet<String> = HashSet::new();
let mut opencode_sqlite_locations: HashMap<String, HashSet<String>> = HashMap::new();
for db_path in &scan_result.opencode_dbs {
let CachedParseOutcome {
messages,
cache_entry,
..
} = load_or_parse_opencode_sqlite_source(db_path, &source_cache, pricing);
for message in messages {
if let Some(key) = message.dedup_key.as_ref() {
opencode_sqlite_locations
.entry(message.session_id.clone())
.or_default()
.insert(key.clone());
if !opencode_seen.insert(key.clone()) {
continue;
}
}
all_messages.push(message);
}
if let Some(entry) = cache_entry {
source_cache.insert(entry);
}
}
let opencode_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::OpenCode)
.par_iter()
.filter(|path| !opencode_json_superseded_by_sqlite(path, &opencode_sqlite_locations))
.filter_map(|path| {
Some(load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::OpenCode),
path,
&source_cache,
pricing,
|path| {
sessions::opencode::parse_opencode_file(path)
.into_iter()
.collect()
},
))
})
.collect();
for outcome in opencode_outcomes {
all_messages.extend(outcome.messages.into_iter().filter(|message| {
message
.dedup_key
.as_ref()
.is_none_or(|key| opencode_seen.insert(key.clone()))
}));
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mut micode_seen: HashSet<String> = HashSet::new();
for db_path in &scan_result.micode_dbs {
let CachedParseOutcome {
messages,
cache_entry,
..
} = load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::MiMoCode),
db_path,
&source_cache,
None,
sessions::micode::parse_micode_sqlite,
);
all_messages.extend(
messages
.into_iter()
.map(|mut message| {
if message.cost <= 0.0 {
apply_pricing_if_available(&mut message, pricing);
}
message
})
.filter(|message| {
message
.dedup_key
.as_ref()
.is_none_or(|key| micode_seen.insert(key.clone()))
}),
);
if let Some(entry) = cache_entry {
source_cache.insert(entry);
}
}
let claude_home = PathBuf::from(home_dir);
let claude_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Claude)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Claude),
path,
&source_cache,
pricing,
|path, cached| {
message_cache::SourceFingerprint::check_claude_code_path_with_home_samples_only(
path,
cached,
Some(&claude_home),
)
},
|path| sessions::claudecode::parse_claude_file_with_home(path, Some(&claude_home)),
)
})
.collect();
let mut claude_messages_raw: Vec<(String, UnifiedMessage)> = Vec::new();
for outcome in claude_outcomes {
claude_messages_raw.extend(outcome.messages.into_iter().map(|msg| {
let dedup_key = msg.dedup_key.clone().unwrap_or_default();
(dedup_key, msg)
}));
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mut seen_keys: HashSet<String> = HashSet::new();
let claude_messages: Vec<UnifiedMessage> = claude_messages_raw
.into_iter()
.filter(|(key, _)| key.is_empty() || seen_keys.insert(key.clone()))
.map(|(_, msg)| msg)
.collect();
all_messages.extend(claude_messages);
let mut openclaw_owned_rollouts: Vec<UnifiedMessage> = Vec::new();
let mut recorded_codex_turns = RecordedCodexTurns::default();
let codex_outcomes: Vec<(PathBuf, CodexSourceOutcome)> = scan_result
.get(ClientId::Codex)
.par_iter()
.map(|path| {
(
path.clone(),
load_or_parse_codex_source(path, &source_cache, pricing, &headless_roots),
)
})
.collect();
let mut codex_seen: HashSet<String> = HashSet::new();
for (path, (outcome, turn_coverage)) in codex_outcomes {
let mut owned_thread: Option<String> = None;
let mut counted_under_codex = false;
for message in outcome.messages {
if message.client == sessions::codex::OPENCLAW_CLIENT_ID {
if owned_thread.is_none() {
owned_thread = Some(message.session_id.clone());
}
openclaw_owned_rollouts.push(message);
} else if should_keep_deduped_message(&mut codex_seen, &message) {
counted_under_codex |= include_all
|| retain_for_requested_clients(
&message.client,
&message.model_id,
&message.provider_id,
&requested,
);
all_messages.push(message);
}
}
if let Some(thread) = owned_thread {
recorded_codex_turns.record(&thread, &turn_coverage);
} else if counted_under_codex {
if let Some(thread) = sessions::codex::thread_id_from_rollout_path(&path) {
recorded_codex_turns.record(&thread, &turn_coverage);
}
}
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
} else if outcome.invalidate_cache {
source_cache.remove(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
);
}
}
let copilot_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Copilot)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Copilot),
path,
&source_cache,
pricing,
sessions::copilot::parse_copilot_file,
)
})
.collect();
for outcome in copilot_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
if let Some(db_path) = &scan_result.copilot_desktop_db {
let otel_sessions: HashSet<String> = all_messages
.iter()
.filter(|message| message.client == "copilot")
.map(|message| message.session_id.clone())
.collect();
let desktop_msgs = sessions::copilot_desktop::parse_copilot_desktop_db(db_path);
all_messages.extend(
desktop_msgs
.into_iter()
.filter(|message| !otel_sessions.contains(&message.session_id))
.map(|mut message| {
apply_pricing_if_available(&mut message, pricing);
message
}),
);
}
{
let existing_dedup_keys: HashSet<String> = all_messages
.iter()
.filter(|m| m.client == "copilot")
.filter_map(|m| m.dedup_key.clone())
.collect();
let existing_copilot_session_timestamps: HashSet<(String, i64)> = all_messages
.iter()
.filter(|m| m.client == "copilot")
.map(|m| (m.session_id.clone(), m.timestamp))
.collect();
let vscode_msgs = sessions::copilot_vscode::parse_copilot_vscode_sessions(
&scan_result.copilot_vscode_sessions,
);
all_messages.extend(
vscode_msgs
.into_iter()
.filter(|m| {
let key_unique = m
.dedup_key
.as_deref()
.map(|k| !existing_dedup_keys.contains(k))
.unwrap_or(true);
let session_ts_unique = !existing_copilot_session_timestamps
.contains(&(m.session_id.clone(), m.timestamp));
key_unique && session_ts_unique
})
.map(|mut message| {
apply_pricing_if_available(&mut message, pricing);
message
}),
);
}
let gemini_outcomes: Vec<(PathBuf, CachedParseOutcome)> = scan_result
.get(ClientId::Gemini)
.par_iter()
.map(|path| {
let outcome = load_or_parse_source_with_fingerprint_and_policy(
message_cache::CacheIdentity::for_client(ClientId::Gemini),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_path_samples_only,
|path, _| {
let parsed = sessions::gemini::parse_gemini_file_with_cache_status(path);
(parsed.messages, parsed.cacheable)
},
);
(path.clone(), outcome)
})
.collect();
for (path, outcome) in gemini_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
} else if outcome.invalidate_cache {
source_cache.remove(
message_cache::CacheIdentity::for_client(ClientId::Gemini),
&path,
);
}
}
let cursor_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Cursor)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Cursor),
path,
&source_cache,
pricing,
sessions::cursor::parse_cursor_file,
)
})
.collect();
for outcome in cursor_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let warp_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Warp)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Warp),
path,
&source_cache,
pricing,
sessions::warp::parse_warp_file,
)
})
.collect();
for outcome in warp_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let grok_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Grok)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Grok),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_grok_path_samples_only,
sessions::grok::parse_grok_updates_file,
)
})
.collect();
for outcome in grok_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let opencodereview_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::OpenCodeReview)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::OpenCodeReview),
path,
&source_cache,
pricing,
sessions::opencodereview::parse_opencodereview_file,
)
})
.collect();
for outcome in opencodereview_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let jcode_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Jcode)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Jcode),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_jcode_path_samples_only,
sessions::jcode::parse_jcode_file,
)
})
.collect();
let mut jcode_seen: HashSet<String> = HashSet::new();
for outcome in jcode_outcomes {
all_messages.extend(
outcome
.messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut jcode_seen, message)),
);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let amp_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Amp)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Amp),
path,
&source_cache,
pricing,
sessions::amp::parse_amp_file,
)
})
.collect();
for outcome in amp_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let codebuff_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Codebuff)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Codebuff),
path,
&source_cache,
pricing,
sessions::codebuff::parse_codebuff_file,
)
})
.collect();
for outcome in codebuff_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let droid_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Droid)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Droid),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_droid_path_samples_only,
sessions::droid::parse_droid_file,
)
})
.collect();
for outcome in droid_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
{
let openclaw_identity = message_cache::CacheIdentity::for_client(ClientId::OpenClaw);
let mut openclaw_transcripts: Vec<&PathBuf> = Vec::new();
let mut openclaw_codex_rollouts: Vec<&PathBuf> = Vec::new();
for path in scan_result.get(ClientId::OpenClaw) {
match sessions::openclaw::classify_openclaw_jsonl(path) {
sessions::openclaw::OpenClawJsonlKind::Transcript => {
openclaw_transcripts.push(path)
}
sessions::openclaw::OpenClawJsonlKind::CodexRollout => {
openclaw_codex_rollouts.push(path)
}
sessions::openclaw::OpenClawJsonlKind::CodexHomeOther => {}
}
}
let rollout_outcomes: Vec<(PathBuf, CodexSourceOutcome)> = openclaw_codex_rollouts
.par_iter()
.map(|path| {
(
(*path).clone(),
load_or_parse_codex_source(path, &source_cache, pricing, &headless_roots),
)
})
.collect();
let mut rollout_messages = std::mem::take(&mut openclaw_owned_rollouts);
for (path, (outcome, turn_coverage)) in rollout_outcomes {
let thread_from_name = sessions::codex::thread_id_from_rollout_path(&path);
let mut recorded_thread: Option<String> = None;
for mut message in outcome.messages {
if message.client != sessions::codex::OPENCLAW_CLIENT_ID {
message.client = sessions::codex::OPENCLAW_CLIENT_ID.to_string();
if let Some(thread) = &thread_from_name {
message.session_id = thread.clone();
}
}
if recorded_thread.is_none() {
recorded_thread = Some(message.session_id.clone());
}
rollout_messages.push(message);
}
if let Some(thread) = recorded_thread {
recorded_codex_turns.record(&thread, &turn_coverage);
}
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
} else if outcome.invalidate_cache {
source_cache.remove(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
);
}
}
let openclaw_db_outcomes: Vec<(PathBuf, CachedParseOutcome)> = scan_result
.openclaw_dbs
.par_iter()
.map(|db_path| {
(
db_path.clone(),
load_or_parse_source_with_fingerprint_and_policy(
openclaw_identity,
db_path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_sqlite_path,
|path, _| {
let scan = sessions::openclaw::scan_openclaw_sqlite(path);
(scan.messages, scan.complete)
},
),
)
})
.collect();
let openclaw_jsonl_outcomes: Vec<(PathBuf, CachedParseOutcome)> = openclaw_transcripts
.par_iter()
.map(|path| {
(
(*path).clone(),
load_or_parse_source(
openclaw_identity,
path,
&source_cache,
pricing,
sessions::openclaw::parse_openclaw_transcript,
),
)
})
.collect();
let mut openclaw_seen: HashSet<String> = HashSet::new();
let mut mirrored_thread_sessions: HashMap<String, String> = HashMap::new();
for (path, outcome) in openclaw_db_outcomes
.into_iter()
.chain(openclaw_jsonl_outcomes)
{
for message in outcome.messages {
if let Some(mirror) = message
.dedup_key
.as_deref()
.and_then(sessions::openclaw::codex_mirror_turn_from_dedup_key)
{
mirrored_thread_sessions
.entry(mirror.thread.to_string())
.or_insert_with(|| message.session_id.clone());
if recorded_codex_turns.covers(mirror.thread, mirror.turn) {
continue;
}
}
if should_keep_deduped_message(&mut openclaw_seen, &message) {
all_messages.push(message);
}
}
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
} else if outcome.invalidate_cache {
source_cache.remove(openclaw_identity, &path);
}
}
for mut message in rollout_messages {
if let Some(session_id) = mirrored_thread_sessions.get(&message.session_id).cloned() {
message.session_id = session_id;
}
if should_keep_deduped_message(&mut openclaw_seen, &message) {
all_messages.push(message);
}
}
}
let pi_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Pi)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Pi),
path,
&source_cache,
pricing,
sessions::pi::parse_pi_file,
)
})
.collect();
for outcome in pi_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mut commandcode_seen: HashSet<String> = HashSet::new();
let commandcode_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::CommandCode)
.par_iter()
.flat_map(|path| {
sessions::commandcode::parse_commandcode_file(path)
.into_iter()
.map(|mut msg| {
if msg.cost <= 0.0 {
apply_pricing_if_available(&mut msg, pricing);
}
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(
commandcode_messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut commandcode_seen, message)),
);
let mut gjc_seen: HashSet<String> = HashSet::new();
let gjc_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::Gjc)
.par_iter()
.flat_map(|path| {
sessions::gjc::parse_gjc_file(path)
.into_iter()
.map(|mut msg| {
if msg.cost <= 0.0 {
apply_pricing_if_available(&mut msg, pricing);
}
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(
gjc_messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut gjc_seen, message)),
);
let mut junie_seen: HashSet<String> = HashSet::new();
let junie_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::Junie)
.par_iter()
.flat_map(|path| {
sessions::junie::parse_junie_file(path)
.into_iter()
.map(|mut msg| {
if msg.cost <= 0.0 {
apply_pricing_if_available(&mut msg, pricing);
}
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(
junie_messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut junie_seen, message)),
);
if let Some(db_path) = &scan_result.zcode_db {
let CachedParseOutcome {
messages,
cache_entry,
..
} = load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::Zcode),
db_path,
&source_cache,
pricing,
sessions::zcode::parse_zcode_sqlite,
);
all_messages.extend(messages);
if let Some(entry) = cache_entry {
source_cache.insert(entry);
}
}
let zcode_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::Zcode)
.par_iter()
.flat_map(|path| {
sessions::zcode::parse_zcode_file(path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(zcode_messages);
let kimi_paths = scan_result.get(ClientId::Kimi);
let kimi_outcomes: Vec<CachedParseOutcome> = kimi_paths
.par_iter()
.map(|path| {
let parse: fn(&Path) -> Vec<UnifiedMessage> = if sessions::kimi::is_kimi_code_path(path)
{
sessions::kimi::parse_kimi_code_file
} else {
sessions::kimi::parse_kimi_file
};
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Kimi),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_kimi_path_samples_only,
parse,
)
})
.collect();
let mut kimi_lane = Vec::with_capacity(kimi_outcomes.len());
for (path, outcome) in kimi_paths.iter().zip(kimi_outcomes) {
kimi_lane.push((path.clone(), outcome.messages));
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
sessions::kimi::apply_code_workspaces(&mut kimi_lane);
for (_, messages) in kimi_lane {
all_messages.extend(messages);
}
let qwen_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Qwen)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Qwen),
path,
&source_cache,
pricing,
sessions::qwen::parse_qwen_file,
)
})
.collect();
for outcome in qwen_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let roocode_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::RooCode)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::RooCode),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_roo_path_samples_only,
sessions::roocode::parse_roocode_file,
)
})
.collect();
for outcome in roocode_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let kilocode_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::KiloCode)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::KiloCode),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_roo_path_samples_only,
sessions::kilocode::parse_kilocode_file,
)
})
.collect();
for outcome in kilocode_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let cline_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Cline)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Cline),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_roo_path_samples_only,
sessions::cline::parse_cline_file,
)
})
.collect();
for outcome in cline_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let senpi_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Senpi)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Senpi),
path,
&source_cache,
pricing,
sessions::senpi::parse_senpi_file,
)
})
.collect();
for outcome in senpi_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let augment_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Augment)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Augment),
path,
&source_cache,
pricing,
sessions::augment::parse_augment_file,
)
})
.collect();
for outcome in augment_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let kimchi_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Kimchi)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Kimchi),
path,
&source_cache,
pricing,
sessions::kimchi::parse_kimchi_file,
)
})
.collect();
for outcome in kimchi_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let reasonix_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Reasonix)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Reasonix),
path,
&source_cache,
pricing,
sessions::reasonix::parse_reasonix_file,
)
})
.collect();
for outcome in reasonix_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let prime_agent_files: Vec<(
Vec<UnifiedMessage>,
sessions::prime_agent::PrimeFileAccounting,
)> = scan_result
.get(ClientId::PrimeAgent)
.par_iter()
.map(|path| sessions::prime_agent::parse_prime_agent_file_with_accounting(path))
.collect();
let mut prime_agent_messages = Vec::new();
let mut prime_agent_accounting = Vec::new();
for (file_messages, file_accounting) in prime_agent_files {
prime_agent_messages.extend(file_messages);
prime_agent_accounting.push(file_accounting);
}
let mut prime_agent_messages = sessions::prime_agent::reconcile_prime_agent_messages(
prime_agent_messages,
&prime_agent_accounting,
);
apply_pricing_to_messages(&mut prime_agent_messages, pricing);
all_messages.extend(prime_agent_messages);
let freebuff_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Freebuff)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Freebuff),
path,
&source_cache,
pricing,
sessions::freebuff::parse_freebuff_file,
)
})
.collect();
for outcome in freebuff_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let cherrystudio_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::CherryStudio)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::CherryStudio),
path,
&source_cache,
pricing,
sessions::cherrystudio::parse_cherrystudio_file,
)
})
.collect();
for outcome in cherrystudio_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let dsh_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Dsh)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Dsh),
path,
&source_cache,
pricing,
sessions::dsh::parse_dsh_file,
)
})
.collect();
for outcome in dsh_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let lmstudio_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::LmStudio)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::LmStudio),
path,
&source_cache,
pricing,
sessions::lmstudio::parse_lmstudio_file,
)
})
.collect();
let mut lmstudio_seen = HashSet::new();
for outcome in lmstudio_outcomes {
all_messages.extend(
outcome
.messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut lmstudio_seen, message)),
);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let unsloth_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Unsloth)
.par_iter()
.map(|db_path| {
load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::Unsloth),
db_path,
&source_cache,
pricing,
sessions::unsloth::parse_unsloth_sqlite,
)
})
.collect();
let mut unsloth_seen = HashSet::new();
for outcome in unsloth_outcomes {
all_messages.extend(
outcome
.messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut unsloth_seen, message)),
);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mcode_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Mcode)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Mcode),
path,
&source_cache,
pricing,
sessions::mcode::parse_mcode_file,
)
})
.collect();
for outcome in mcode_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let fx_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Fx)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Fx),
path,
&source_cache,
pricing,
sessions::fx::parse_fx_file,
)
})
.collect();
for outcome in fx_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let omp_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Omp)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Omp),
path,
&source_cache,
pricing,
sessions::omp::parse_omp_file,
)
})
.collect();
for outcome in omp_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mux_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Mux)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::Mux),
path,
&source_cache,
pricing,
sessions::mux::parse_mux_file,
)
})
.collect();
for outcome in mux_outcomes {
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
if let Some(db_path) = &scan_result.kilo_db {
let kilo_messages: Vec<UnifiedMessage> = sessions::kilo::parse_kilo_sqlite(db_path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect();
all_messages.extend(kilo_messages);
}
let mut hermes_seen: HashSet<String> = HashSet::new();
for db_path in scan_result.hermes_db_paths() {
let hermes_messages = parse_hermes_sqlite_with_pricing(&db_path, pricing);
all_messages.extend(
hermes_messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut hermes_seen, message)),
);
}
if let Some(db_path) = &scan_result.goose_db {
let goose_messages: Vec<UnifiedMessage> = sessions::goose::parse_goose_sqlite(db_path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect();
all_messages.extend(goose_messages);
}
let mut devin_cli_session_ids: HashSet<String> = HashSet::new();
if include_devin_cli {
let devin_cli_outcomes: Vec<CachedParseOutcome> = scan_result
.devin_dbs
.par_iter()
.map(|db_path| {
load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::DevinCli),
db_path,
&source_cache,
pricing,
sessions::devin::parse_devin_cli_sqlite,
)
})
.collect();
let mut devin_cli_seen = HashSet::new();
for outcome in devin_cli_outcomes {
for message in outcome
.messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut devin_cli_seen, message))
{
devin_cli_session_ids.insert(message.session_id.clone());
all_messages.push(message);
}
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
}
for db_path in scan_result.zed_db_paths() {
let outcome = load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::Zed),
&db_path,
&source_cache,
pricing,
sessions::zed::parse_zed_sqlite,
);
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let kiro_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::Kiro)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint(
message_cache::CacheIdentity::for_client(ClientId::Kiro),
path,
&source_cache,
pricing,
message_cache::SourceFingerprint::check_kiro_path_samples_only,
sessions::kiro::parse_kiro_file,
)
})
.collect();
let mut kiro_file_messages: Vec<UnifiedMessage> = Vec::new();
for outcome in kiro_outcomes {
kiro_file_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
all_messages.extend(sessions::kiro::suppress_snapshots_covered_by_executions(
kiro_file_messages,
));
if let Some(db_path) = &scan_result.kiro_db {
let kiro_db_messages: Vec<UnifiedMessage> = sessions::kiro::parse_kiro_sqlite(db_path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect();
all_messages.extend(kiro_db_messages);
}
for source in &scan_result.crush_dbs {
let crush_messages: Vec<UnifiedMessage> =
sessions::crush::parse_crush_sqlite(&source.db_path)
.into_iter()
.map(|mut msg| {
msg.set_workspace(source.workspace_key.clone(), source.workspace_label.clone());
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect();
all_messages.extend(crush_messages);
}
let antigravity_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::Antigravity)
.par_iter()
.flat_map(|path| {
sessions::antigravity::parse_antigravity_file(path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(antigravity_messages);
let antigravity_cli_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::AntigravityCli)
.par_iter()
.flat_map(|path| {
sessions::antigravity_cli::parse_antigravity_cli_file(path)
.into_iter()
.map(|mut msg| {
apply_pricing_if_available(&mut msg, pricing);
msg
})
.collect::<Vec<_>>()
})
.collect();
all_messages.extend(antigravity_cli_messages);
let trae_messages: Vec<UnifiedMessage> = scan_result
.get(ClientId::Trae)
.par_iter()
.flat_map(|path| sessions::trae::parse_trae_file("trae", path))
.collect();
let deduped_trae_messages = dedupe_latest_trae_messages(trae_messages);
all_messages.extend(deduped_trae_messages);
let codebuddy_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::CodeBuddy)
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::CodeBuddy),
path,
&source_cache,
pricing,
sessions::codebuddy::parse_codebuddy_file,
)
})
.collect();
let mut codebuddy_seen: HashSet<String> = HashSet::new();
for outcome in codebuddy_outcomes {
all_messages.extend(outcome.messages.into_iter().filter(|message| {
message
.dedup_key
.as_ref()
.is_none_or(|key| codebuddy_seen.insert(key.clone()))
}));
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
if include_devin_desktop {
let devin_desktop_lookups = DevinDesktopLookupCache::default();
let devin_desktop_outcomes: Vec<CachedParseOutcome> = scan_result
.get(ClientId::DevinDesktop)
.par_iter()
.map(|path| {
load_or_parse_source_with_fingerprint_context(
message_cache::CacheIdentity::for_client(ClientId::DevinDesktop),
path,
&source_cache,
pricing,
|path, cached| {
message_cache::SourceFingerprint::check_devin_desktop_path_samples_only(
path,
&scan_result.devin_dbs,
cached,
)
},
|path, fingerprint| {
if let Some(fingerprint) = fingerprint {
let lookup_cell = devin_desktop_lookup_cell_for_snapshot(
&devin_desktop_lookups,
&scan_result.devin_dbs,
fingerprint,
);
let lookup = lookup_cell.get_or_init(|| {
sessions::devin::load_devin_desktop_session_lookup(
&scan_result.devin_dbs,
)
});
sessions::devin::parse_devin_desktop_ndjson_with_lookup(path, lookup)
} else {
sessions::devin::parse_devin_desktop_ndjson_with_lookup(
path,
&sessions::devin::load_devin_desktop_session_lookup(
&scan_result.devin_dbs,
),
)
}
},
)
})
.collect();
for outcome in devin_desktop_outcomes {
all_messages.extend(
outcome
.messages
.into_iter()
.filter(|message| !devin_cli_session_ids.contains(&message.session_id)),
);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
}
let (workbuddy_detailed_paths, workbuddy_fallback_paths) =
partition_workbuddy_paths(scan_result.get(ClientId::WorkBuddy));
let workbuddy_detailed_outcomes: Vec<CachedParseOutcome> = workbuddy_detailed_paths
.par_iter()
.map(|path| {
load_or_parse_source(
message_cache::CacheIdentity::for_client(ClientId::WorkBuddy),
path,
&source_cache,
pricing,
sessions::workbuddy::parse_workbuddy_file,
)
})
.collect();
let workbuddy_fallback_outcomes: Vec<CachedParseOutcome> = workbuddy_fallback_paths
.par_iter()
.map(|path| {
load_or_parse_sqlite_source(
message_cache::CacheIdentity::for_client(ClientId::WorkBuddy),
path,
&source_cache,
pricing,
sessions::workbuddy::parse_workbuddy_file,
)
})
.collect();
let mut workbuddy_detailed_messages = Vec::new();
for outcome in workbuddy_detailed_outcomes {
workbuddy_detailed_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
let mut workbuddy_fallback_messages = Vec::new();
for outcome in workbuddy_fallback_outcomes {
workbuddy_fallback_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
all_messages.extend(merge_workbuddy_messages(
workbuddy_detailed_messages,
workbuddy_fallback_messages,
));
if include_synthetic {
if let Some(db_path) = &scan_result.synthetic_db {
let outcome = load_or_parse_sqlite_source(
message_cache::CacheIdentity::synthetic(),
db_path,
&source_cache,
pricing,
sessions::synthetic::parse_octofriend_sqlite,
);
all_messages.extend(outcome.messages);
if let Some(entry) = outcome.cache_entry {
source_cache.insert(entry);
}
}
}
if !include_all {
let requested: HashSet<&str> = clients.iter().map(String::as_str).collect();
all_messages.retain(|msg| {
retain_for_requested_clients(&msg.client, &msg.model_id, &msg.provider_id, &requested)
});
}
if include_synthetic {
for msg in &mut all_messages {
sessions::synthetic::normalize_synthetic_gateway_fields(
&mut msg.model_id,
&mut msg.provider_id,
);
}
}
source_cache.save_if_dirty();
all_messages
}
fn dedupe_latest_trae_messages(mut messages: Vec<UnifiedMessage>) -> Vec<UnifiedMessage> {
let mut latest_by_session: HashMap<String, UnifiedMessage> = HashMap::new();
for message in messages.drain(..) {
let session_id = message.session_id.clone();
match latest_by_session.get_mut(&session_id) {
Some(existing) => {
let should_replace = message.timestamp > existing.timestamp
|| (message.timestamp == existing.timestamp
&& message.dedup_key.as_ref().is_some_and(|key| {
existing
.dedup_key
.as_ref()
.is_none_or(|existing_key| key > existing_key)
}));
if should_replace {
*existing = message;
}
}
None => {
let _ = latest_by_session.insert(session_id, message);
}
}
}
let mut deduped: Vec<UnifiedMessage> = latest_by_session.into_values().collect();
deduped.sort_unstable_by(|a, b| {
a.session_id
.cmp(&b.session_id)
.then_with(|| a.timestamp.cmp(&b.timestamp))
});
deduped
}
fn partition_workbuddy_paths(paths: &[PathBuf]) -> (Vec<&PathBuf>, Vec<&PathBuf>) {
paths
.iter()
.partition(|path| sessions::workbuddy::is_detailed_workbuddy_source(path))
}
fn merge_workbuddy_messages(
detailed_messages: Vec<UnifiedMessage>,
fallback_messages: Vec<UnifiedMessage>,
) -> Vec<UnifiedMessage> {
let detailed_sessions: HashSet<String> = detailed_messages
.iter()
.filter(|message| !message.session_id.is_empty())
.map(|message| message.session_id.clone())
.collect();
let mut seen: HashSet<String> = HashSet::new();
let mut merged: Vec<UnifiedMessage> = detailed_messages
.into_iter()
.filter(|message| should_keep_deduped_message(&mut seen, message))
.collect();
merged.extend(fallback_messages.into_iter().filter(|message| {
!detailed_sessions.contains(&message.session_id)
&& should_keep_deduped_message(&mut seen, message)
}));
merged
}
fn filter_unified_messages(
messages: Vec<UnifiedMessage>,
options: &LocalParseOptions,
) -> Vec<UnifiedMessage> {
let mut filtered = messages;
if let Some(year) = &options.year {
let year_prefix = format!("{}-", year);
filtered.retain(|m| m.date.starts_with(&year_prefix));
}
if let Some(since) = &options.since {
filtered.retain(|m| m.date.as_str() >= since.as_str());
}
if let Some(until) = &options.until {
filtered.retain(|m| m.date.as_str() <= until.as_str());
}
filtered
}
fn report_workspace_bucket(msg: &UnifiedMessage) -> (String, Option<String>, String) {
match (&msg.workspace_key, &msg.workspace_label) {
(Some(key), Some(label)) => (key.clone(), Some(key.clone()), label.clone()),
(Some(key), None) => (
key.clone(),
Some(key.clone()),
sessions::workspace_label_from_key(key)
.unwrap_or_else(|| UNKNOWN_WORKSPACE_LABEL.to_string()),
),
_ => (
UNKNOWN_WORKSPACE_GROUP_KEY.to_string(),
None,
UNKNOWN_WORKSPACE_LABEL.to_string(),
),
}
}
fn aggregate_model_usage_entries(
messages: Vec<UnifiedMessage>,
group_by: &GroupBy,
) -> Vec<ModelUsage> {
let mut model_map: HashMap<String, ModelUsage> = HashMap::new();
for msg in messages {
let normalized = model_name_for_grouping(&msg.client, &msg.provider_id, &msg.model_id);
let (workspace_group_key, workspace_key, workspace_label) = report_workspace_bucket(&msg);
let key = match group_by {
GroupBy::Model => normalized.clone(),
GroupBy::ClientModel => format!("{}:{}", msg.client, normalized),
GroupBy::ClientProviderModel => {
format!("{}:{}:{}", msg.client, msg.provider_id, normalized)
}
GroupBy::WorkspaceModel => format!("{}:{}", workspace_group_key, normalized),
GroupBy::Session => format!("{}:{}", msg.session_id, normalized),
GroupBy::ClientSession => {
format!("{}:{}:{}", msg.client, msg.session_id, normalized)
}
};
let merge_clients = matches!(group_by, GroupBy::Model | GroupBy::WorkspaceModel);
let session_grouped = matches!(group_by, GroupBy::Session | GroupBy::ClientSession);
let entry = model_map.entry(key).or_insert_with(|| ModelUsage {
client: msg.client.clone(),
merged_clients: if merge_clients {
Some(msg.client.clone())
} else {
None
},
workspace_key: if matches!(group_by, GroupBy::WorkspaceModel) {
workspace_key.clone()
} else {
None
},
workspace_label: if matches!(group_by, GroupBy::WorkspaceModel) {
Some(workspace_label.clone())
} else {
None
},
session_id: if session_grouped {
Some(msg.session_id.clone())
} else {
None
},
model: normalized.clone(),
provider: msg.provider_id.clone(),
input: 0,
output: 0,
cache_read: 0,
cache_write: 0,
reasoning: 0,
message_count: 0,
cost: 0.0,
performance: ModelPerformance::default(),
});
if merge_clients {
if !entry.client.split(", ").any(|s| s == msg.client) {
entry.client = format!("{}, {}", entry.client, msg.client);
}
if let Some(merged_clients) = &mut entry.merged_clients {
if !merged_clients.split(", ").any(|s| s == msg.client) {
*merged_clients = format!("{}, {}", merged_clients, msg.client);
}
}
}
if *group_by != GroupBy::ClientProviderModel
&& !entry.provider.split(", ").any(|p| p == msg.provider_id)
{
entry.provider = format!("{}, {}", entry.provider, msg.provider_id);
}
entry.input = entry.input.saturating_add(msg.tokens.input);
entry.output = entry.output.saturating_add(msg.tokens.output);
entry.cache_read = entry.cache_read.saturating_add(msg.tokens.cache_read);
entry.cache_write = entry.cache_write.saturating_add(msg.tokens.cache_write);
entry.reasoning = entry.reasoning.saturating_add(msg.tokens.reasoning);
entry.message_count += msg.message_count.max(0);
entry.cost += msg.cost;
entry
.performance
.record_message(positive_token_total(&msg.tokens), msg.duration_ms);
}
let mut entries: Vec<ModelUsage> = model_map
.into_values()
.map(|mut entry| {
let total_tokens = entry
.input
.max(0)
.saturating_add(entry.output.max(0))
.saturating_add(entry.cache_read.max(0))
.saturating_add(entry.cache_write.max(0))
.saturating_add(entry.reasoning.max(0));
entry.performance.finalize(total_tokens);
let mut providers: Vec<&str> = entry.provider.split(", ").collect();
providers.sort_unstable();
providers.dedup();
entry.provider = providers.join(", ");
entry
})
.collect();
entries.sort_by(|a, b| match (a.cost.is_nan(), b.cost.is_nan()) {
(true, true) => std::cmp::Ordering::Equal,
(true, false) => std::cmp::Ordering::Greater,
(false, true) => std::cmp::Ordering::Less,
(false, false) => b
.cost
.partial_cmp(&a.cost)
.unwrap_or(std::cmp::Ordering::Equal),
});
entries
}
fn positive_token_total(tokens: &TokenBreakdown) -> i64 {
tokens
.input
.max(0)
.saturating_add(tokens.output.max(0))
.saturating_add(tokens.cache_read.max(0))
.saturating_add(tokens.cache_write.max(0))
.saturating_add(tokens.reasoning.max(0))
}
fn model_report_token_totals(entries: &[ModelUsage]) -> (i64, i64, i64, i64) {
entries.iter().fold(
(0, 0, 0, 0),
|(input, output, cache_read, cache_write), entry| {
(
input.saturating_add(entry.input),
output.saturating_add(entry.output),
cache_read.saturating_add(entry.cache_read),
cache_write.saturating_add(entry.cache_write),
)
},
)
}
pub async fn get_model_report(options: ReportOptions) -> Result<ModelReport, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::ALL
.iter()
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let pricing = load_pricing_for_local_parse().await;
let all_messages = parse_all_messages_with_pricing_with_env_strategy(
&home_dir,
&clients,
pricing.as_deref(),
options.use_env_roots,
&options.scanner_settings,
);
let filtered = filter_messages_for_report(all_messages, &options);
let entries = aggregate_model_usage_entries(filtered, &options.group_by);
let (total_input, total_output, total_cache_read, total_cache_write) =
model_report_token_totals(&entries);
let total_messages: i32 = entries.iter().map(|e| e.message_count).sum();
let total_cost: f64 = entries.iter().map(|e| e.cost).sum::<f64>() + 0.0;
Ok(ModelReport {
entries,
total_input,
total_output,
total_cache_read,
total_cache_write,
total_messages,
total_cost,
processing_time_ms: start.elapsed().as_millis() as u32,
})
}
#[derive(Default)]
struct MonthAggregator {
models: HashSet<String>,
input: i64,
output: i64,
cache_read: i64,
cache_write: i64,
message_count: i32,
cost: f64,
}
pub async fn get_monthly_report(options: ReportOptions) -> Result<MonthlyReport, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::ALL
.iter()
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let pricing = load_pricing_for_local_parse().await;
let all_messages = parse_all_messages_with_pricing_with_env_strategy(
&home_dir,
&clients,
pricing.as_deref(),
options.use_env_roots,
&options.scanner_settings,
);
let filtered = filter_messages_for_report(all_messages, &options);
let mut month_map: HashMap<String, MonthAggregator> = HashMap::new();
for msg in filtered {
let month = if msg.date.len() >= 7 {
msg.date[..7].to_string()
} else {
continue;
};
let entry = month_map.entry(month).or_default();
entry.models.insert(model_name_for_grouping(
&msg.client,
&msg.provider_id,
&msg.model_id,
));
entry.input = entry.input.saturating_add(msg.tokens.input);
entry.output = entry.output.saturating_add(msg.tokens.output);
entry.cache_read = entry.cache_read.saturating_add(msg.tokens.cache_read);
entry.cache_write = entry.cache_write.saturating_add(msg.tokens.cache_write);
entry.message_count += msg.message_count.max(0);
entry.cost += msg.cost;
}
let mut entries: Vec<MonthlyUsage> = month_map
.into_iter()
.map(|(month, agg)| MonthlyUsage {
month,
models: agg.models.into_iter().collect(),
input: agg.input,
output: agg.output,
cache_read: agg.cache_read,
cache_write: agg.cache_write,
message_count: agg.message_count,
cost: agg.cost,
})
.collect();
entries.sort_by(|a, b| a.month.cmp(&b.month));
let total_cost: f64 = entries.iter().map(|e| e.cost).sum::<f64>() + 0.0;
Ok(MonthlyReport {
entries,
total_cost,
processing_time_ms: start.elapsed().as_millis() as u32,
})
}
#[derive(Default)]
struct HourAggregator {
clients: HashSet<String>,
models: HashSet<String>,
input: i64,
output: i64,
cache_read: i64,
cache_write: i64,
reasoning: i64,
message_count: i32,
turn_count: i32,
cost: f64,
}
pub async fn get_hourly_report(options: ReportOptions) -> Result<HourlyReport, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::ALL
.iter()
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let pricing = load_pricing_for_local_parse().await;
let all_messages = parse_all_messages_with_pricing_with_env_strategy(
&home_dir,
&clients,
pricing.as_deref(),
options.use_env_roots,
&options.scanner_settings,
);
let filtered = filter_messages_for_report(all_messages, &options);
let mut hour_map: HashMap<String, HourAggregator> = HashMap::new();
for msg in filtered {
let hour_key = if msg.timestamp > 0 {
crate::bucket_tz::bucket_timezone()
.date_hour_of_ms(msg.timestamp)
.unwrap_or_else(|| format!("{} 00:00", msg.date))
} else {
format!("{} 00:00", msg.date)
};
let entry = hour_map.entry(hour_key).or_default();
entry.clients.insert(msg.client.clone());
entry.models.insert(model_name_for_grouping(
&msg.client,
&msg.provider_id,
&msg.model_id,
));
entry.input = entry.input.saturating_add(msg.tokens.input);
entry.output = entry.output.saturating_add(msg.tokens.output);
entry.cache_read = entry.cache_read.saturating_add(msg.tokens.cache_read);
entry.cache_write = entry.cache_write.saturating_add(msg.tokens.cache_write);
entry.reasoning = entry.reasoning.saturating_add(msg.tokens.reasoning);
entry.message_count += msg.message_count.max(0);
if msg.is_turn_start {
entry.turn_count += 1;
}
entry.cost += msg.cost;
}
let mut entries: Vec<HourlyUsage> = hour_map
.into_iter()
.map(|(hour, agg)| HourlyUsage {
hour,
clients: {
let mut v: Vec<String> = agg.clients.into_iter().collect();
v.sort();
v
},
models: {
let mut v: Vec<String> = agg.models.into_iter().collect();
v.sort();
v
},
input: agg.input,
output: agg.output,
cache_read: agg.cache_read,
cache_write: agg.cache_write,
message_count: agg.message_count,
turn_count: agg.turn_count,
reasoning: agg.reasoning,
cost: agg.cost,
})
.collect();
entries.sort_by(|a, b| a.hour.cmp(&b.hour));
let total_cost: f64 = entries.iter().map(|e| e.cost).sum::<f64>() + 0.0;
Ok(HourlyReport {
entries,
total_cost,
processing_time_ms: start.elapsed().as_millis() as u32,
})
}
async fn generate_graph_with_loaded_pricing(
options: ReportOptions,
pricing: Option<&pricing::PricingService>,
) -> Result<GraphResult, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::ALL
.iter()
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let all_messages = parse_all_messages_with_pricing_with_env_strategy(
&home_dir,
&clients,
pricing,
options.use_env_roots,
&options.scanner_settings,
);
let filtered = filter_messages_for_report(all_messages, &options);
let intervals = sessionize::sessionize(&filtered, sessionize::DEFAULT_IDLE_GAP_MS);
let time_metrics =
sessionize::compute_time_metrics(&intervals, sessionize::DEFAULT_IDLE_GAP_MS);
let daily_active_time = sessionize::compute_daily_active_time(&intervals);
let contributions = aggregator::aggregate_by_date(filtered);
let processing_time_ms = start.elapsed().as_millis() as u32;
let mut result = aggregator::generate_graph_result(contributions, processing_time_ms);
result.time_metrics = Some(time_metrics);
for contribution in &mut result.contributions {
if let Some(&ms) = daily_active_time.get(&contribution.date) {
contribution.active_time_ms = Some(ms);
}
}
Ok(result)
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct TimeMetricsReport {
pub metrics: sessionize::TimeMetrics,
pub processing_time_ms: u32,
}
pub async fn get_time_metrics_report(options: ReportOptions) -> Result<TimeMetricsReport, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::ALL
.iter()
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let all_messages = parse_all_messages_with_pricing_with_env_strategy(
&home_dir,
&clients,
None,
options.use_env_roots,
&options.scanner_settings,
);
let filtered = filter_messages_for_report(all_messages, &options);
let intervals = sessionize::sessionize(&filtered, sessionize::DEFAULT_IDLE_GAP_MS);
let metrics = sessionize::compute_time_metrics(&intervals, sessionize::DEFAULT_IDLE_GAP_MS);
Ok(TimeMetricsReport {
metrics,
processing_time_ms: start.elapsed().as_millis() as u32,
})
}
pub async fn generate_graph(options: ReportOptions) -> Result<GraphResult, String> {
let pricing = pricing::PricingService::get_or_init().await?;
generate_graph_with_loaded_pricing(options, Some(&pricing)).await
}
pub async fn generate_local_graph_report(options: ReportOptions) -> Result<GraphResult, String> {
let pricing = load_pricing_for_local_parse().await;
generate_graph_with_loaded_pricing(options, pricing.as_deref()).await
}
fn filter_messages_for_report(
messages: Vec<UnifiedMessage>,
options: &ReportOptions,
) -> Vec<UnifiedMessage> {
let mut filtered = messages;
if let Some(year) = &options.year {
let year_prefix = format!("{}-", year);
filtered.retain(|m| m.date.starts_with(&year_prefix));
}
if let Some(since) = &options.since {
filtered.retain(|m| m.date.as_str() >= since.as_str());
}
if let Some(until) = &options.until {
filtered.retain(|m| m.date.as_str() <= until.as_str());
}
filtered
}
fn is_headless_path(path: &Path, headless_roots: &[PathBuf]) -> bool {
headless_roots.iter().any(|root| path.starts_with(root))
}
fn apply_headless_agent(message: &mut UnifiedMessage, is_headless: bool) {
if is_headless && message.agent.is_none() {
message.agent = Some("headless".to_string());
}
}
fn pricing_multiplier(message: &UnifiedMessage) -> f64 {
if message.client == "zed"
&& message
.provider_id
.eq_ignore_ascii_case(sessions::zed::ZED_HOSTED_PROVIDER)
{
1.1
} else {
1.0
}
}
fn apply_pricing_if_available(
message: &mut UnifiedMessage,
pricing: Option<&pricing::PricingService>,
) {
if message.has_authoritative_cost() {
return;
}
let Some(pricing) = pricing else {
return;
};
let calculated_cost = pricing.calculate_cost_with_provider(
&message.model_id,
Some(&message.provider_id),
&message.tokens,
) * pricing_multiplier(message);
if calculated_cost > 0.0 {
message.cost = calculated_cost;
message.mark_estimated_cost();
}
}
fn parse_hermes_sqlite_with_pricing(
db_path: &Path,
pricing: Option<&pricing::PricingService>,
) -> Vec<UnifiedMessage> {
sessions::hermes::parse_hermes_sqlite(db_path)
.into_iter()
.map(|mut msg| {
if msg.cost <= 0.0 {
apply_pricing_if_available(&mut msg, pricing);
}
msg
})
.collect()
}
fn select_local_parse_pricing<F>(
fresh: Result<Arc<pricing::PricingService>, String>,
stale: F,
) -> Option<Arc<pricing::PricingService>>
where
F: FnOnce() -> Option<pricing::PricingService>,
{
fresh.ok().or_else(|| stale().map(Arc::new))
}
async fn load_pricing_for_local_parse() -> Option<Arc<pricing::PricingService>> {
if std::env::var("TOKMESH_PRICING_CACHE_ONLY")
.map(|value| matches!(value.as_str(), "1" | "true" | "TRUE" | "yes" | "YES"))
.unwrap_or(false)
{
return pricing::PricingService::load_cached_any_age().map(Arc::new);
}
select_local_parse_pricing(
pricing::PricingService::get_or_init().await,
pricing::PricingService::load_cached_any_age,
)
}
fn resolve_local_parse_request(
options: &LocalParseOptions,
) -> Result<(String, Vec<String>), String> {
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::iter()
.filter(|c| c.parse_local())
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
Ok((home_dir, clients))
}
fn parse_local_unified_messages_resolved(
options: LocalParseOptions,
home_dir: &str,
clients: &[String],
pricing: Option<&pricing::PricingService>,
) -> Result<Vec<UnifiedMessage>, String> {
let messages = parse_all_messages_with_pricing_with_env_strategy(
home_dir,
clients,
pricing,
options.use_env_roots,
&options.scanner_settings,
);
Ok(filter_unified_messages(messages, &options))
}
fn opencode_json_superseded_by_sqlite(
path: &Path,
sqlite_locations: &HashMap<String, HashSet<String>>,
) -> bool {
let Some(message_id) = path.file_stem().and_then(|stem| stem.to_str()) else {
return false;
};
let Some(session_id) = path
.parent()
.and_then(Path::file_name)
.and_then(|name| name.to_str())
else {
return false;
};
sqlite_locations
.get(session_id)
.is_some_and(|message_ids| message_ids.contains(message_id))
}
#[derive(Debug, Default)]
struct RecordedCodexTurns {
turns: HashMap<String, HashSet<String>>,
whole_threads: HashSet<String>,
}
impl RecordedCodexTurns {
fn record(&mut self, thread: &str, coverage: &sessions::codex::CodexTurnCoverage) {
if coverage.without_turn_id || coverage.turn_ids.is_empty() {
self.whole_threads.insert(thread.to_string());
} else {
self.turns
.entry(thread.to_string())
.or_default()
.extend(coverage.turn_ids.iter().cloned());
}
}
fn covers(&self, thread: &str, turn: Option<&str>) -> bool {
self.whole_threads.contains(thread)
|| turn.is_some_and(|turn| {
self.turns
.get(thread)
.is_some_and(|turns| turns.contains(turn))
})
}
}
pub fn parse_local_clients(options: LocalParseOptions) -> Result<ParsedMessages, String> {
let start = Instant::now();
let home_dir = get_home_dir_string(&options.home_dir)?;
let clients: Vec<String> = options.clients.clone().unwrap_or_else(|| {
let mut clients: Vec<String> = ClientId::iter()
.filter(|c| c.parse_local())
.map(|c| c.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
clients
});
let include_all = clients.is_empty();
let requested: HashSet<&str> = clients.iter().map(String::as_str).collect();
let include_synthetic = include_all || clients.iter().any(|c| c == "synthetic");
let include_devin_cli = include_synthetic || clients.iter().any(|c| c == "devin-cli");
let include_devin_desktop = include_synthetic || clients.iter().any(|c| c == "devin-desktop");
let scan_result = scanner::scan_all_clients_with_scanner_settings(
&home_dir,
&clients,
options.use_env_roots,
&options.scanner_settings,
);
let headless_roots =
scanner::headless_roots_with_env_strategy(&home_dir, options.use_env_roots);
let mut messages: Vec<ParsedMessage> = Vec::new();
let mut counts = ClientCounts::new();
let opencode_count: i32 = {
let mut seen: HashSet<String> = HashSet::new();
let mut count: i32 = 0;
for db_path in &scan_result.opencode_dbs {
let sqlite_msgs: Vec<(String, ParsedMessage)> =
sessions::opencode::parse_opencode_sqlite(db_path)
.into_iter()
.filter_map(|msg| {
let key = msg.dedup_key.clone().unwrap_or_default();
if !key.is_empty() && !seen.insert(key.clone()) {
return None;
}
Some((key, unified_to_parsed(&msg)))
})
.collect();
count += sqlite_msgs.len() as i32;
for (_key, parsed) in sqlite_msgs {
messages.push(parsed);
}
}
let json_msgs: Vec<(String, ParsedMessage)> = scan_result
.get(ClientId::OpenCode)
.par_iter()
.filter_map(|path| {
let msg = sessions::opencode::parse_opencode_file(path)?;
let key = msg.dedup_key.clone().unwrap_or_default();
Some((key, unified_to_parsed(&msg)))
})
.collect();
let deduped: Vec<ParsedMessage> = json_msgs
.into_iter()
.filter(|(key, _)| key.is_empty() || seen.insert(key.clone()))
.map(|(_, msg)| msg)
.collect();
count += deduped.len() as i32;
messages.extend(deduped);
count
};
counts.set(ClientId::OpenCode, opencode_count);
let mut micode_seen: HashSet<String> = HashSet::new();
let micode_msgs: Vec<ParsedMessage> = scan_result
.micode_dbs
.iter()
.flat_map(|db_path| sessions::micode::parse_micode_sqlite(db_path))
.filter(|msg| should_keep_deduped_message(&mut micode_seen, msg))
.map(|msg| unified_to_parsed(&msg))
.collect();
let micode_count = summed_parsed_message_count(&micode_msgs);
counts.set(ClientId::MiMoCode, micode_count);
messages.extend(micode_msgs);
let claude_home = PathBuf::from(&home_dir);
let claude_msgs_raw: Vec<(String, ParsedMessage)> = scan_result
.get(ClientId::Claude)
.par_iter()
.map_init(std::collections::HashMap::new, |parent_cache, path| {
sessions::claudecode::parse_claude_file_with_cache_and_home(
path,
parent_cache,
Some(&claude_home),
)
.into_iter()
.map(|msg| {
let dedup_key = msg.dedup_key.clone().unwrap_or_default();
(dedup_key, unified_to_parsed(&msg))
})
.collect::<Vec<_>>()
})
.flatten()
.collect();
let mut seen_keys: HashSet<String> = HashSet::new();
let claude_msgs: Vec<ParsedMessage> = claude_msgs_raw
.into_iter()
.filter(|(key, _)| key.is_empty() || seen_keys.insert(key.clone()))
.map(|(_, msg)| msg)
.collect();
let claude_count = claude_msgs.len() as i32;
counts.set(ClientId::Claude, claude_count);
messages.extend(claude_msgs);
let codex_files: Vec<(
PathBuf,
Vec<UnifiedMessage>,
sessions::codex::CodexTurnCoverage,
)> = scan_result
.get(ClientId::Codex)
.par_iter()
.map(|path| {
let is_headless = is_headless_path(path, &headless_roots);
let parsed = sessions::codex::parse_codex_file_incremental(
path,
0,
sessions::codex::CodexParseState::default(),
);
let messages = parsed
.messages
.into_iter()
.map(|mut msg| {
apply_headless_agent(&mut msg, is_headless);
msg
})
.collect::<Vec<_>>();
(path.clone(), messages, parsed.state.turn_coverage)
})
.collect();
let mut openclaw_owned_rollouts: Vec<UnifiedMessage> = Vec::new();
let mut recorded_codex_turns = RecordedCodexTurns::default();
let mut codex_seen: HashSet<String> = HashSet::new();
let mut codex_msgs: Vec<ParsedMessage> = Vec::new();
for (path, file_messages, turn_coverage) in codex_files {
let mut owned_thread: Option<String> = None;
let mut counted_under_codex = false;
for message in file_messages {
if message.client == sessions::codex::OPENCLAW_CLIENT_ID {
if owned_thread.is_none() {
owned_thread = Some(message.session_id.clone());
}
openclaw_owned_rollouts.push(message);
} else if should_keep_deduped_message(&mut codex_seen, &message)
&& (include_all
|| retain_for_requested_clients(
&message.client,
&message.model_id,
&message.provider_id,
&requested,
))
{
counted_under_codex = true;
codex_msgs.push(unified_to_parsed(&message));
}
}
if let Some(thread) = owned_thread {
recorded_codex_turns.record(&thread, &turn_coverage);
} else if counted_under_codex {
if let Some(thread) = sessions::codex::thread_id_from_rollout_path(&path) {
recorded_codex_turns.record(&thread, &turn_coverage);
}
}
}
let codex_count = codex_msgs.len() as i32;
counts.set(ClientId::Codex, codex_count);
messages.extend(codex_msgs);
let mut copilot_unified_msgs: Vec<_> = scan_result
.get(ClientId::Copilot)
.par_iter()
.flat_map(|path| {
sessions::copilot::parse_copilot_file(path)
.into_iter()
.collect::<Vec<_>>()
})
.collect();
if let Some(db_path) = &scan_result.copilot_desktop_db {
let otel_sessions: HashSet<String> = copilot_unified_msgs
.iter()
.map(|message| message.session_id.clone())
.collect();
copilot_unified_msgs.extend(
sessions::copilot_desktop::parse_copilot_desktop_db(db_path)
.into_iter()
.filter(|message| !otel_sessions.contains(&message.session_id)),
);
}
{
let existing_dedup_keys: HashSet<String> = copilot_unified_msgs
.iter()
.filter_map(|m| m.dedup_key.clone())
.collect();
let existing_copilot_session_timestamps: HashSet<(String, i64)> = copilot_unified_msgs
.iter()
.map(|m| (m.session_id.clone(), m.timestamp))
.collect();
copilot_unified_msgs.extend(
sessions::copilot_vscode::parse_copilot_vscode_sessions(
&scan_result.copilot_vscode_sessions,
)
.into_iter()
.filter(|m| {
let key_unique = m
.dedup_key
.as_deref()
.map(|k| !existing_dedup_keys.contains(k))
.unwrap_or(true);
let session_ts_unique = !existing_copilot_session_timestamps
.contains(&(m.session_id.clone(), m.timestamp));
key_unique && session_ts_unique
}),
);
}
let copilot_msgs: Vec<ParsedMessage> =
copilot_unified_msgs.iter().map(unified_to_parsed).collect();
let copilot_count = copilot_msgs.len() as i32;
counts.set(ClientId::Copilot, copilot_count);
messages.extend(copilot_msgs);
let gemini_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Gemini)
.par_iter()
.flat_map(|path| {
sessions::gemini::parse_gemini_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let gemini_count = gemini_msgs.len() as i32;
counts.set(ClientId::Gemini, gemini_count);
messages.extend(gemini_msgs);
let amp_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Amp)
.par_iter()
.flat_map(|path| {
sessions::amp::parse_amp_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let amp_count = amp_msgs.len() as i32;
counts.set(ClientId::Amp, amp_count);
messages.extend(amp_msgs);
let codebuff_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Codebuff)
.par_iter()
.flat_map(|path| {
sessions::codebuff::parse_codebuff_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let codebuff_count = codebuff_msgs.len() as i32;
counts.set(ClientId::Codebuff, codebuff_count);
messages.extend(codebuff_msgs);
let droid_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Droid)
.par_iter()
.flat_map(|path| {
sessions::droid::parse_droid_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let droid_count = droid_msgs.len() as i32;
counts.set(ClientId::Droid, droid_count);
messages.extend(droid_msgs);
let include_openclaw = include_all || clients.iter().any(|c| c == "openclaw");
if !include_openclaw {
openclaw_owned_rollouts.clear();
}
let openclaw_msgs: Vec<ParsedMessage> = {
let mut openclaw_transcripts: Vec<&PathBuf> = Vec::new();
let mut openclaw_codex_rollouts: Vec<&PathBuf> = Vec::new();
for path in scan_result.get(ClientId::OpenClaw) {
match sessions::openclaw::classify_openclaw_jsonl(path) {
sessions::openclaw::OpenClawJsonlKind::Transcript => {
openclaw_transcripts.push(path)
}
sessions::openclaw::OpenClawJsonlKind::CodexRollout => {
openclaw_codex_rollouts.push(path)
}
sessions::openclaw::OpenClawJsonlKind::CodexHomeOther => {}
}
}
let mut rollout_messages: Vec<UnifiedMessage> = openclaw_owned_rollouts;
let location_rollouts: Vec<(Vec<UnifiedMessage>, sessions::codex::CodexTurnCoverage)> =
openclaw_codex_rollouts
.par_iter()
.map(|path| {
let thread_from_name = sessions::codex::thread_id_from_rollout_path(path);
let parsed = sessions::codex::parse_codex_file_incremental(
path,
0,
sessions::codex::CodexParseState::default(),
);
let messages = parsed
.messages
.into_iter()
.map(|mut message| {
if message.client != sessions::codex::OPENCLAW_CLIENT_ID {
message.client = sessions::codex::OPENCLAW_CLIENT_ID.to_string();
if let Some(thread) = &thread_from_name {
message.session_id = thread.clone();
}
}
message
})
.collect::<Vec<_>>();
(messages, parsed.state.turn_coverage)
})
.collect();
for (messages, turn_coverage) in location_rollouts {
if let Some(first) = messages.first() {
recorded_codex_turns.record(&first.session_id, &turn_coverage);
}
rollout_messages.extend(messages);
}
let sqlite_messages: Vec<UnifiedMessage> = scan_result
.openclaw_dbs
.par_iter()
.flat_map(|db_path| sessions::openclaw::parse_openclaw_sqlite(db_path))
.collect();
let jsonl_messages: Vec<UnifiedMessage> = openclaw_transcripts
.par_iter()
.flat_map(|path| sessions::openclaw::parse_openclaw_transcript(path))
.collect();
let mut seen: HashSet<String> = HashSet::new();
let mut mirrored_thread_sessions: HashMap<String, String> = HashMap::new();
let mut kept: Vec<UnifiedMessage> = Vec::new();
for message in sqlite_messages.into_iter().chain(jsonl_messages) {
if let Some(mirror) = message
.dedup_key
.as_deref()
.and_then(sessions::openclaw::codex_mirror_turn_from_dedup_key)
{
mirrored_thread_sessions
.entry(mirror.thread.to_string())
.or_insert_with(|| message.session_id.clone());
if recorded_codex_turns.covers(mirror.thread, mirror.turn) {
continue;
}
}
if should_keep_deduped_message(&mut seen, &message) {
kept.push(message);
}
}
for mut message in rollout_messages {
if let Some(session_id) = mirrored_thread_sessions.get(&message.session_id).cloned() {
message.session_id = session_id;
}
if should_keep_deduped_message(&mut seen, &message) {
kept.push(message);
}
}
kept.iter().map(unified_to_parsed).collect()
};
let openclaw_count = openclaw_msgs.len() as i32;
counts.set(ClientId::OpenClaw, openclaw_count);
messages.extend(openclaw_msgs);
let pi_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Pi)
.par_iter()
.flat_map(|path| {
sessions::pi::parse_pi_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let pi_count = pi_msgs.len() as i32;
counts.set(ClientId::Pi, pi_count);
messages.extend(pi_msgs);
let commandcode_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::CommandCode)
.par_iter()
.flat_map(|path| sessions::commandcode::parse_commandcode_file(path))
.collect();
let mut commandcode_seen: HashSet<String> = HashSet::new();
let commandcode_msgs: Vec<ParsedMessage> = commandcode_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut commandcode_seen, message))
.map(|msg| unified_to_parsed(&msg))
.collect();
let commandcode_count = summed_parsed_message_count(&commandcode_msgs);
counts.set(ClientId::CommandCode, commandcode_count);
messages.extend(commandcode_msgs);
let gjc_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::Gjc)
.par_iter()
.flat_map(|path| sessions::gjc::parse_gjc_file(path))
.collect();
let mut gjc_seen: HashSet<String> = HashSet::new();
let gjc_msgs: Vec<ParsedMessage> = gjc_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut gjc_seen, message))
.map(|message| unified_to_parsed(&message))
.collect();
let gjc_count = gjc_msgs.len() as i32;
counts.set(ClientId::Gjc, gjc_count);
messages.extend(gjc_msgs);
let junie_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::Junie)
.par_iter()
.flat_map(|path| sessions::junie::parse_junie_file(path))
.collect();
let mut junie_seen: HashSet<String> = HashSet::new();
let junie_msgs: Vec<ParsedMessage> = junie_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut junie_seen, message))
.map(|message| unified_to_parsed(&message))
.collect();
let junie_count = summed_parsed_message_count(&junie_msgs);
counts.set(ClientId::Junie, junie_count);
messages.extend(junie_msgs);
let mut zcode_msgs: Vec<ParsedMessage> = scan_result
.zcode_db
.as_ref()
.map(|db_path| {
sessions::zcode::parse_zcode_sqlite(db_path)
.into_iter()
.map(|message| unified_to_parsed(&message))
.collect::<Vec<_>>()
})
.unwrap_or_default();
zcode_msgs.extend(
scan_result
.get(ClientId::Zcode)
.par_iter()
.flat_map(|path| sessions::zcode::parse_zcode_file(path))
.map(|message| unified_to_parsed(&message))
.collect::<Vec<_>>(),
);
let zcode_count = summed_parsed_message_count(&zcode_msgs);
counts.set(ClientId::Zcode, zcode_count);
messages.extend(zcode_msgs);
let opencodereview_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::OpenCodeReview)
.par_iter()
.flat_map(|path| {
sessions::opencodereview::parse_opencodereview_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let opencodereview_count = summed_parsed_message_count(&opencodereview_msgs);
counts.set(ClientId::OpenCodeReview, opencodereview_count);
messages.extend(opencodereview_msgs);
let mut kimi_sources: Vec<(PathBuf, Vec<UnifiedMessage>)> = scan_result
.get(ClientId::Kimi)
.par_iter()
.map(|path| {
let messages = if sessions::kimi::is_kimi_code_path(path) {
sessions::kimi::parse_kimi_code_file(path)
} else {
sessions::kimi::parse_kimi_file(path)
};
(path.clone(), messages)
})
.collect();
sessions::kimi::apply_code_workspaces(&mut kimi_sources);
let kimi_msgs: Vec<ParsedMessage> = kimi_sources
.into_iter()
.flat_map(|(_, messages)| {
messages
.into_iter()
.map(|message| unified_to_parsed(&message))
})
.collect();
counts.set(ClientId::Kimi, summed_parsed_message_count(&kimi_msgs));
messages.extend(kimi_msgs);
let qwen_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Qwen)
.par_iter()
.flat_map(|path| {
sessions::qwen::parse_qwen_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let qwen_count = qwen_msgs.len() as i32;
counts.set(ClientId::Qwen, qwen_count);
messages.extend(qwen_msgs);
let roocode_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::RooCode)
.par_iter()
.flat_map(|path| {
sessions::roocode::parse_roocode_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let roocode_count = roocode_msgs.len() as i32;
counts.set(ClientId::RooCode, roocode_count);
messages.extend(roocode_msgs);
let kilocode_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::KiloCode)
.par_iter()
.flat_map(|path| {
sessions::kilocode::parse_kilocode_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let kilocode_count = summed_parsed_message_count(&kilocode_msgs);
counts.set(ClientId::KiloCode, kilocode_count);
messages.extend(kilocode_msgs);
let cline_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Cline)
.par_iter()
.flat_map(|path| {
sessions::cline::parse_cline_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let cline_count = summed_parsed_message_count(&cline_msgs);
counts.set(ClientId::Cline, cline_count);
messages.extend(cline_msgs);
let senpi_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Senpi)
.par_iter()
.flat_map(|path| {
sessions::senpi::parse_senpi_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let senpi_count = summed_parsed_message_count(&senpi_msgs);
counts.set(ClientId::Senpi, senpi_count);
messages.extend(senpi_msgs);
let augment_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Augment)
.par_iter()
.flat_map(|path| {
sessions::augment::parse_augment_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let augment_count = summed_parsed_message_count(&augment_msgs);
counts.set(ClientId::Augment, augment_count);
messages.extend(augment_msgs);
let kimchi_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Kimchi)
.par_iter()
.flat_map(|path| {
sessions::kimchi::parse_kimchi_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let kimchi_count = summed_parsed_message_count(&kimchi_msgs);
counts.set(ClientId::Kimchi, kimchi_count);
messages.extend(kimchi_msgs);
let reasonix_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Reasonix)
.par_iter()
.flat_map(|path| {
sessions::reasonix::parse_reasonix_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let reasonix_count = summed_parsed_message_count(&reasonix_msgs);
counts.set(ClientId::Reasonix, reasonix_count);
messages.extend(reasonix_msgs);
let prime_agent_files: Vec<(
Vec<UnifiedMessage>,
sessions::prime_agent::PrimeFileAccounting,
)> = scan_result
.get(ClientId::PrimeAgent)
.par_iter()
.map(|path| sessions::prime_agent::parse_prime_agent_file_with_accounting(path))
.collect();
let mut prime_agent_msgs_raw = Vec::new();
let mut prime_agent_accounting = Vec::new();
for (file_messages, file_accounting) in prime_agent_files {
prime_agent_msgs_raw.extend(file_messages);
prime_agent_accounting.push(file_accounting);
}
let prime_agent_msgs: Vec<ParsedMessage> =
sessions::prime_agent::reconcile_prime_agent_messages(
prime_agent_msgs_raw,
&prime_agent_accounting,
)
.into_iter()
.map(|message| unified_to_parsed(&message))
.collect();
let prime_agent_count = summed_parsed_message_count(&prime_agent_msgs);
counts.set(ClientId::PrimeAgent, prime_agent_count);
messages.extend(prime_agent_msgs);
let freebuff_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Freebuff)
.par_iter()
.flat_map(|path| {
sessions::freebuff::parse_freebuff_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let freebuff_count = summed_parsed_message_count(&freebuff_msgs);
counts.set(ClientId::Freebuff, freebuff_count);
messages.extend(freebuff_msgs);
let cherrystudio_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::CherryStudio)
.par_iter()
.flat_map(|path| {
sessions::cherrystudio::parse_cherrystudio_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let cherrystudio_count = summed_parsed_message_count(&cherrystudio_msgs);
counts.set(ClientId::CherryStudio, cherrystudio_count);
messages.extend(cherrystudio_msgs);
let dsh_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Dsh)
.par_iter()
.flat_map(|path| {
sessions::dsh::parse_dsh_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let dsh_count = summed_parsed_message_count(&dsh_msgs);
counts.set(ClientId::Dsh, dsh_count);
messages.extend(dsh_msgs);
let mcode_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Mcode)
.par_iter()
.flat_map(|path| {
sessions::mcode::parse_mcode_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let mcode_count = summed_parsed_message_count(&mcode_msgs);
counts.set(ClientId::Mcode, mcode_count);
messages.extend(mcode_msgs);
let omp_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Omp)
.par_iter()
.flat_map(|path| {
sessions::omp::parse_omp_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let omp_count = summed_parsed_message_count(&omp_msgs);
counts.set(ClientId::Omp, omp_count);
messages.extend(omp_msgs);
let lmstudio_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::LmStudio)
.par_iter()
.flat_map(|path| sessions::lmstudio::parse_lmstudio_file(path))
.collect();
let mut lmstudio_seen: HashSet<String> = HashSet::new();
let lmstudio_msgs: Vec<ParsedMessage> = lmstudio_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut lmstudio_seen, message))
.map(|message| unified_to_parsed(&message))
.collect();
let lmstudio_count = summed_parsed_message_count(&lmstudio_msgs);
counts.set(ClientId::LmStudio, lmstudio_count);
messages.extend(lmstudio_msgs);
let unsloth_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::Unsloth)
.par_iter()
.flat_map(|path| sessions::unsloth::parse_unsloth_sqlite(path))
.collect();
let mut unsloth_seen: HashSet<String> = HashSet::new();
let unsloth_msgs: Vec<ParsedMessage> = unsloth_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut unsloth_seen, message))
.map(|message| unified_to_parsed(&message))
.collect();
let unsloth_count = summed_parsed_message_count(&unsloth_msgs);
counts.set(ClientId::Unsloth, unsloth_count);
messages.extend(unsloth_msgs);
let mux_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Mux)
.par_iter()
.flat_map(|path| {
sessions::mux::parse_mux_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let mux_count = summed_parsed_message_count(&mux_msgs);
counts.set(ClientId::Mux, mux_count);
messages.extend(mux_msgs);
let fx_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Fx)
.par_iter()
.flat_map(|path| {
sessions::fx::parse_fx_file(path)
.into_iter()
.map(|mut msg| {
msg.refresh_derived_fields();
unified_to_parsed(&msg)
})
.collect::<Vec<_>>()
})
.collect();
let fx_count = summed_parsed_message_count(&fx_msgs);
counts.set(ClientId::Fx, fx_count);
messages.extend(fx_msgs);
let _kilo_count: i32 = if let Some(db_path) = &scan_result.kilo_db {
let kilo_msgs: Vec<ParsedMessage> = sessions::kilo::parse_kilo_sqlite(db_path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let count = summed_parsed_message_count(&kilo_msgs);
counts.set(ClientId::Kilo, count);
messages.extend(kilo_msgs);
count
} else {
0
};
let hermes_db_paths = scan_result.hermes_db_paths();
if !hermes_db_paths.is_empty() {
let mut hermes_seen: HashSet<String> = HashSet::new();
let hermes_msgs: Vec<ParsedMessage> = hermes_db_paths
.iter()
.flat_map(|db_path| sessions::hermes::parse_hermes_sqlite(db_path))
.filter(|msg| should_keep_deduped_message(&mut hermes_seen, msg))
.map(|msg| unified_to_parsed(&msg))
.collect();
let count = summed_parsed_message_count(&hermes_msgs);
counts.set(ClientId::Hermes, count);
messages.extend(hermes_msgs);
}
if let Some(db_path) = &scan_result.goose_db {
let goose_msgs: Vec<ParsedMessage> = sessions::goose::parse_goose_sqlite(db_path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let count = summed_parsed_message_count(&goose_msgs);
counts.set(ClientId::Goose, count);
messages.extend(goose_msgs);
}
let zed_db_paths = scan_result.zed_db_paths();
if !zed_db_paths.is_empty() {
let zed_msgs: Vec<ParsedMessage> = zed_db_paths
.iter()
.flat_map(|db_path| sessions::zed::parse_zed_sqlite(db_path))
.map(|msg| unified_to_parsed(&msg))
.collect();
let count = summed_parsed_message_count(&zed_msgs);
counts.set(ClientId::Zed, count);
messages.extend(zed_msgs);
}
let kiro_unified: Vec<UnifiedMessage> = scan_result
.get(ClientId::Kiro)
.par_iter()
.flat_map(|path| sessions::kiro::parse_kiro_file(path))
.collect();
let kiro_msgs: Vec<ParsedMessage> =
sessions::kiro::suppress_snapshots_covered_by_executions(kiro_unified)
.iter()
.map(unified_to_parsed)
.collect();
let kiro_count = summed_parsed_message_count(&kiro_msgs);
counts.set(ClientId::Kiro, kiro_count);
messages.extend(kiro_msgs);
if let Some(db_path) = &scan_result.kiro_db {
let kiro_db_msgs: Vec<ParsedMessage> = sessions::kiro::parse_kiro_sqlite(db_path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let kiro_db_count = summed_parsed_message_count(&kiro_db_msgs);
counts.add(ClientId::Kiro, kiro_db_count);
messages.extend(kiro_db_msgs);
}
let crush_msgs: Vec<ParsedMessage> = scan_result
.crush_dbs
.par_iter()
.flat_map(|source| {
sessions::crush::parse_crush_sqlite(&source.db_path)
.into_iter()
.map(|mut msg| {
msg.set_workspace(source.workspace_key.clone(), source.workspace_label.clone());
unified_to_parsed(&msg)
})
.collect::<Vec<_>>()
})
.collect();
let crush_count = summed_parsed_message_count(&crush_msgs);
counts.set(ClientId::Crush, crush_count);
messages.extend(crush_msgs);
let antigravity_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Antigravity)
.par_iter()
.flat_map(|path| {
sessions::antigravity::parse_antigravity_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let antigravity_count = antigravity_msgs.len() as i32;
counts.set(ClientId::Antigravity, antigravity_count);
messages.extend(antigravity_msgs);
let antigravity_cli_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::AntigravityCli)
.par_iter()
.flat_map(|path| {
sessions::antigravity_cli::parse_antigravity_cli_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let antigravity_cli_count = antigravity_cli_msgs.len() as i32;
counts.set(ClientId::AntigravityCli, antigravity_cli_count);
messages.extend(antigravity_cli_msgs);
let trae_msgs: Vec<ParsedMessage> = {
let unique_trae_messages = dedupe_latest_trae_messages(
scan_result
.get(ClientId::Trae)
.par_iter()
.flat_map(|path| sessions::trae::parse_trae_file("trae", path))
.collect(),
);
unique_trae_messages
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect()
};
let trae_count = trae_msgs.len() as i32;
counts.set(ClientId::Trae, trae_count);
messages.extend(trae_msgs);
let warp_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Warp)
.par_iter()
.flat_map(|path| {
sessions::warp::parse_warp_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let warp_count = summed_parsed_message_count(&warp_msgs);
counts.set(ClientId::Warp, warp_count);
messages.extend(warp_msgs);
let mut devin_cli_seen = HashSet::new();
let devin_cli_messages: Vec<UnifiedMessage> = if include_devin_cli {
scan_result
.devin_dbs
.iter()
.flat_map(|db_path| sessions::devin::parse_devin_cli_sqlite(db_path))
.filter(|message| should_keep_deduped_message(&mut devin_cli_seen, message))
.collect()
} else {
Vec::new()
};
let cli_session_ids: HashSet<String> = devin_cli_messages
.iter()
.map(|message| message.session_id.clone())
.collect();
let devin_desktop_messages_raw: Vec<UnifiedMessage> = if include_devin_desktop {
let devin_desktop_lookup =
sessions::devin::load_devin_desktop_session_lookup(&scan_result.devin_dbs);
scan_result
.get(ClientId::DevinDesktop)
.par_iter()
.flat_map(|path| {
sessions::devin::parse_devin_desktop_ndjson_with_lookup(path, &devin_desktop_lookup)
})
.collect()
} else {
Vec::new()
};
let devin_desktop_raw_count: i32 = devin_desktop_messages_raw
.iter()
.map(|msg| msg.message_count.max(0))
.sum();
let devin_desktop_messages: Vec<UnifiedMessage> = devin_desktop_messages_raw
.into_iter()
.filter(|message| !cli_session_ids.contains(&message.session_id))
.collect();
let devin_cli_parsed: Vec<ParsedMessage> = devin_cli_messages
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let devin_desktop_parsed: Vec<ParsedMessage> = devin_desktop_messages
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let devin_cli_count = summed_parsed_message_count(&devin_cli_parsed);
let devin_desktop_count = devin_desktop_raw_count;
counts.set(ClientId::DevinCli, devin_cli_count);
counts.set(ClientId::DevinDesktop, devin_desktop_count);
messages.extend(devin_cli_parsed);
messages.extend(devin_desktop_parsed);
let codebuddy_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::CodeBuddy)
.par_iter()
.flat_map(|path| sessions::codebuddy::parse_codebuddy_file(path))
.collect();
let mut codebuddy_seen: HashSet<String> = HashSet::new();
let codebuddy_msgs: Vec<ParsedMessage> = codebuddy_msgs_raw
.into_iter()
.filter(|message| {
message
.dedup_key
.as_ref()
.is_none_or(|key| codebuddy_seen.insert(key.clone()))
})
.map(|msg| unified_to_parsed(&msg))
.collect();
let codebuddy_count = summed_parsed_message_count(&codebuddy_msgs);
counts.set(ClientId::CodeBuddy, codebuddy_count);
messages.extend(codebuddy_msgs);
let (workbuddy_detailed_paths, workbuddy_fallback_paths) =
partition_workbuddy_paths(scan_result.get(ClientId::WorkBuddy));
let workbuddy_detailed_messages: Vec<UnifiedMessage> = workbuddy_detailed_paths
.par_iter()
.flat_map(|path| sessions::workbuddy::parse_workbuddy_file(path))
.collect();
let workbuddy_fallback_messages: Vec<UnifiedMessage> = workbuddy_fallback_paths
.par_iter()
.flat_map(|path| sessions::workbuddy::parse_workbuddy_file(path))
.collect();
let workbuddy_msgs: Vec<ParsedMessage> =
merge_workbuddy_messages(workbuddy_detailed_messages, workbuddy_fallback_messages)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
let workbuddy_count = summed_parsed_message_count(&workbuddy_msgs);
counts.set(ClientId::WorkBuddy, workbuddy_count);
messages.extend(workbuddy_msgs);
let grok_msgs: Vec<ParsedMessage> = scan_result
.get(ClientId::Grok)
.par_iter()
.flat_map(|path| {
sessions::grok::parse_grok_updates_file(path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect::<Vec<_>>()
})
.collect();
let grok_count = summed_parsed_message_count(&grok_msgs);
counts.set(ClientId::Grok, grok_count);
messages.extend(grok_msgs);
let jcode_msgs_raw: Vec<UnifiedMessage> = scan_result
.get(ClientId::Jcode)
.par_iter()
.flat_map(|path| sessions::jcode::parse_jcode_file(path))
.collect();
let mut jcode_seen: HashSet<String> = HashSet::new();
let jcode_msgs: Vec<ParsedMessage> = jcode_msgs_raw
.into_iter()
.filter(|message| should_keep_deduped_message(&mut jcode_seen, message))
.map(|msg| unified_to_parsed(&msg))
.collect();
let jcode_count = summed_parsed_message_count(&jcode_msgs);
counts.set(ClientId::Jcode, jcode_count);
messages.extend(jcode_msgs);
if include_synthetic {
if let Some(db_path) = &scan_result.synthetic_db {
let synthetic_msgs: Vec<ParsedMessage> =
sessions::synthetic::parse_octofriend_sqlite(db_path)
.into_iter()
.map(|msg| unified_to_parsed(&msg))
.collect();
messages.extend(synthetic_msgs);
}
}
if !include_all {
messages.retain(|msg| {
retain_for_requested_clients(&msg.client, &msg.model_id, &msg.provider_id, &requested)
});
}
if include_synthetic {
for msg in &mut messages {
sessions::synthetic::normalize_synthetic_gateway_fields(
&mut msg.model_id,
&mut msg.provider_id,
);
}
}
let filtered = filter_parsed_messages(messages, &options);
Ok(ParsedMessages {
messages: filtered,
counts,
processing_time_ms: start.elapsed().as_millis() as u32,
})
}
#[doc(hidden)]
pub async fn parse_local_unified_messages_with_pricing(
options: LocalParseOptions,
pricing: Option<&pricing::PricingService>,
) -> Result<Vec<UnifiedMessage>, String> {
let (home_dir, clients) = resolve_local_parse_request(&options)?;
parse_local_unified_messages_resolved(options, &home_dir, &clients, pricing)
}
pub async fn parse_local_unified_messages(
options: LocalParseOptions,
) -> Result<Vec<UnifiedMessage>, String> {
let (home_dir, clients) = resolve_local_parse_request(&options)?;
let pricing = load_pricing_for_local_parse().await;
parse_local_unified_messages_resolved(options, &home_dir, &clients, pricing.as_deref())
}
fn unified_to_parsed(msg: &UnifiedMessage) -> ParsedMessage {
ParsedMessage {
client: msg.client.clone(),
model_id: msg.model_id.clone(),
provider_id: msg.provider_id.clone(),
session_id: msg.session_id.clone(),
workspace_key: msg.workspace_key.clone(),
workspace_label: msg.workspace_label.clone(),
timestamp: msg.timestamp,
date: msg.date.clone(),
input: msg.tokens.input,
output: msg.tokens.output,
cache_read: msg.tokens.cache_read,
cache_write: msg.tokens.cache_write,
reasoning: msg.tokens.reasoning,
duration_ms: msg.duration_ms,
message_count: msg.message_count,
agent: msg.agent.clone(),
cost: msg.cost,
cost_source: msg.cost_source,
}
}
fn should_keep_deduped_message(seen_keys: &mut HashSet<String>, message: &UnifiedMessage) -> bool {
message
.dedup_key
.as_ref()
.is_none_or(|key| seen_keys.insert(key.clone()))
}
fn summed_parsed_message_count(messages: &[ParsedMessage]) -> i32 {
messages
.iter()
.map(|msg| msg.message_count.max(0))
.sum::<i32>()
}
fn filter_parsed_messages(
messages: Vec<ParsedMessage>,
options: &LocalParseOptions,
) -> Vec<ParsedMessage> {
let mut filtered = messages;
if let Some(year) = &options.year {
let year_prefix = format!("{}-", year);
filtered.retain(|m| m.date.starts_with(&year_prefix));
}
if let Some(since) = &options.since {
filtered.retain(|m| m.date.as_str() >= since.as_str());
}
if let Some(until) = &options.until {
filtered.retain(|m| m.date.as_str() <= until.as_str());
}
filtered
}
pub fn parsed_to_unified(msg: &ParsedMessage, cost: f64) -> UnifiedMessage {
UnifiedMessage {
client: msg.client.clone(),
model_id: msg.model_id.clone(),
provider_id: msg.provider_id.clone(),
session_id: msg.session_id.clone(),
workspace_key: msg.workspace_key.clone(),
workspace_label: msg.workspace_label.clone(),
timestamp: msg.timestamp,
date: msg.date.clone(),
tokens: TokenBreakdown {
input: msg.input,
output: msg.output,
cache_read: msg.cache_read,
cache_write: msg.cache_write,
reasoning: msg.reasoning,
},
cost,
cost_source: CostSource::Unknown,
duration_ms: msg.duration_ms,
message_count: msg.message_count,
agent: msg.agent.clone(),
dedup_key: None,
session_title: None,
is_turn_start: false,
}
}
#[cfg(test)]
mod tests {
use super::{
aggregate_model_usage_entries, apply_pricing_if_available, dedupe_latest_trae_messages,
filter_messages_for_report, generate_graph_with_loaded_pricing, message_cache,
normalize_model_for_grouping, parse_all_messages_with_pricing_with_env_strategy,
parse_local_clients, parsed_to_unified, pricing, retain_for_requested_clients, scanner,
select_local_parse_pricing, unified_to_parsed, ClientId, GroupBy, LocalParseOptions,
ReportOptions, TokenBreakdown, UnifiedMessage, UNKNOWN_WORKSPACE_LABEL,
};
use std::collections::{HashMap, HashSet};
use std::io::Write;
use std::str::FromStr;
use std::sync::Arc;
fn parse_all_messages_with_pricing(
home_dir: &str,
clients: &[String],
pricing: Option<&pricing::PricingService>,
) -> Vec<UnifiedMessage> {
parse_all_messages_with_pricing_with_env_strategy(
home_dir,
clients,
pricing,
false,
&scanner::ScannerSettings::default(),
)
}
#[test]
fn token_total_saturates_on_overlarge_buckets() {
let t = TokenBreakdown {
input: i64::MAX,
output: i64::MAX,
cache_read: i64::MAX,
cache_write: 0,
reasoning: 0,
};
assert_eq!(t.total(), i64::MAX);
assert_eq!(super::positive_token_total(&t), i64::MAX);
}
#[test]
fn model_aggregation_saturates_overflowing_token_folds() {
let make = || {
UnifiedMessage::new_with_dedup(
"antigravity-cli",
"gemini-3-pro",
"antigravity",
"session-overflow",
1_733_011_200_000,
TokenBreakdown {
input: i64::MAX,
output: 0,
cache_read: i64::MAX,
cache_write: 0,
reasoning: 0,
},
0.0,
None,
)
};
let entries = aggregate_model_usage_entries(vec![make(), make()], &GroupBy::Model);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].input, i64::MAX);
assert_eq!(entries[0].cache_read, i64::MAX);
}
#[test]
fn model_report_totals_saturate_across_groups() {
let make = |model: &str| {
UnifiedMessage::new_with_dedup(
"antigravity-cli",
model,
"antigravity",
"session-overflow",
1_733_011_200_000,
TokenBreakdown {
input: i64::MAX,
output: 0,
cache_read: i64::MAX,
cache_write: 0,
reasoning: 0,
},
0.0,
None,
)
};
let entries = aggregate_model_usage_entries(
vec![make("gemini-3-pro"), make("claude-opus-4-6")],
&GroupBy::Model,
);
assert_eq!(entries.len(), 2);
let (total_input, _total_output, total_cache_read, _total_cache_write) =
super::model_report_token_totals(&entries);
assert_eq!(total_input, i64::MAX);
assert_eq!(total_cache_read, i64::MAX);
}
fn make_workspace_message(
client: &str,
model_id: &str,
provider_id: &str,
session_id: &str,
cost: f64,
workspace_key: Option<&str>,
workspace_label: Option<&str>,
) -> UnifiedMessage {
let mut msg = UnifiedMessage::new(
client,
model_id,
provider_id,
session_id,
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
cost,
);
msg.set_workspace(
workspace_key.map(str::to_string),
workspace_label.map(str::to_string),
);
msg
}
fn make_workbuddy_message(
session_id: &str,
timestamp: i64,
input: i64,
dedup_key: &str,
) -> UnifiedMessage {
let mut msg = UnifiedMessage::new(
"workbuddy",
"glm-5.2",
"zai",
session_id,
timestamp,
TokenBreakdown {
input,
output: 0,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
msg.dedup_key = Some(dedup_key.to_string());
msg
}
fn make_trae_message(
session_id: &str,
timestamp: i64,
dedup_key: Option<&str>,
cost: f64,
) -> UnifiedMessage {
UnifiedMessage::new_with_dedup(
"trae",
"gpt-5.2",
"openai",
session_id,
timestamp,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
cost,
dedup_key.map(str::to_string),
)
}
#[test]
fn workbuddy_fallback_dedups_by_session_not_date() {
const DAY1: i64 = 1_782_883_200_000;
const DAY2: i64 = 1_782_969_600_000;
let detailed = vec![make_workbuddy_message(
"sess-A",
DAY1,
100,
"workbuddy:detailed-A",
)];
let fallback = vec![
make_workbuddy_message("sess-A", DAY2, 5000, "workbuddy:fallback-A"),
make_workbuddy_message("sess-B", DAY1, 2000, "workbuddy:fallback-B"),
];
let merged = super::merge_workbuddy_messages(detailed, fallback);
assert_eq!(merged.len(), 2);
assert!(merged
.iter()
.any(|message| message.dedup_key.as_deref() == Some("workbuddy:detailed-A")));
assert!(merged
.iter()
.any(|message| message.dedup_key.as_deref() == Some("workbuddy:fallback-B")));
assert!(!merged
.iter()
.any(|message| message.dedup_key.as_deref() == Some("workbuddy:fallback-A")));
}
#[test]
fn workbuddy_fallback_kept_when_no_detailed_messages() {
let fallback = vec![
make_workbuddy_message("sess-A", 1_782_883_200_000, 1000, "workbuddy:fallback-A"),
make_workbuddy_message("sess-B", 1_782_969_600_000, 2000, "workbuddy:fallback-B"),
];
let merged = super::merge_workbuddy_messages(Vec::new(), fallback);
assert_eq!(merged.len(), 2);
}
#[allow(clippy::too_many_arguments)]
fn build_opencode_sqlite_payload(
created_ms: f64,
completed_ms: f64,
input: i64,
output: i64,
reasoning: i64,
cache_read: i64,
cache_write: i64,
cost: f64,
) -> String {
format!(
r#"{{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"cost": {cost},
"tokens": {{
"input": {input},
"output": {output},
"reasoning": {reasoning},
"cache": {{ "read": {cache_read}, "write": {cache_write} }}
}},
"time": {{ "created": {created_ms}, "completed": {completed_ms} }},
"mode": "build"
}}"#
)
}
fn create_opencode_sqlite_db(db_path: &std::path::Path) -> rusqlite::Connection {
let conn = rusqlite::Connection::open(db_path).unwrap();
conn.execute_batch(
"CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.unwrap();
conn
}
fn create_hermes_sqlite_db(db_path: &std::path::Path) -> rusqlite::Connection {
let conn = rusqlite::Connection::open(db_path).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
source TEXT NOT NULL,
model TEXT,
started_at REAL NOT NULL,
message_count INTEGER DEFAULT 0,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
cache_read_tokens INTEGER DEFAULT 0,
cache_write_tokens INTEGER DEFAULT 0,
reasoning_tokens INTEGER DEFAULT 0,
billing_provider TEXT,
estimated_cost_usd REAL,
actual_cost_usd REAL
);
CREATE TABLE session_model_usage (
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
model TEXT NOT NULL,
billing_provider TEXT NOT NULL DEFAULT '',
billing_base_url TEXT NOT NULL DEFAULT '',
billing_mode TEXT NOT NULL DEFAULT '',
task TEXT NOT NULL DEFAULT '',
api_call_count INTEGER NOT NULL DEFAULT 0,
input_tokens INTEGER NOT NULL DEFAULT 0,
output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
cache_write_tokens INTEGER NOT NULL DEFAULT 0,
reasoning_tokens INTEGER NOT NULL DEFAULT 0,
estimated_cost_usd REAL NOT NULL DEFAULT 0,
actual_cost_usd REAL NOT NULL DEFAULT 0,
cost_status TEXT,
cost_source TEXT,
first_seen REAL,
last_seen REAL,
PRIMARY KEY (session_id, model, billing_provider, billing_base_url, billing_mode, task)
);",
)
.unwrap();
conn
}
fn create_zed_sqlite_db(db_path: &std::path::Path) -> rusqlite::Connection {
let conn = rusqlite::Connection::open(db_path).unwrap();
conn.execute_batch(
"CREATE TABLE threads (
id TEXT PRIMARY KEY,
summary TEXT NOT NULL,
updated_at TEXT NOT NULL,
data_type TEXT NOT NULL,
data BLOB NOT NULL
);",
)
.unwrap();
conn
}
fn insert_zed_thread(conn: &rusqlite::Connection, id: &str, model: &str) {
let payload = format!(
r#"{{
"version": "0.3.0",
"title": "Test thread",
"updated_at": "2026-05-01T12:30:00Z",
"request_token_usage": {{
"turn-1": {{
"input_tokens": 42,
"output_tokens": 7,
"cache_creation_input_tokens": 3,
"cache_read_input_tokens": 5
}}
}},
"model": {{
"provider": "zed.dev",
"model": "{model}"
}},
"imported": false
}}"#
);
conn.execute(
"INSERT INTO threads (id, summary, updated_at, data_type, data) VALUES (?1, ?2, ?3, ?4, ?5)",
rusqlite::params![id, "Test thread", "2026-05-01T12:30:00Z", "json", payload.as_bytes()],
)
.unwrap();
}
fn insert_hermes_session(
conn: &rusqlite::Connection,
id: &str,
model: &str,
message_count: i64,
input_tokens: i64,
output_tokens: i64,
actual_cost_usd: f64,
) {
conn.execute(
"INSERT INTO sessions (
id, source, model, started_at, message_count,
input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens,
billing_provider, estimated_cost_usd, actual_cost_usd
) VALUES (?1, 'cli', ?2, 1775001102.0, ?3, ?4, ?5, 0, 0, 0, 'anthropic', NULL, ?6)",
rusqlite::params![
id,
model,
message_count,
input_tokens,
output_tokens,
actual_cost_usd
],
)
.unwrap();
conn.execute(
"INSERT INTO session_model_usage (
session_id, model, billing_provider, billing_base_url, billing_mode, task,
input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens,
estimated_cost_usd, actual_cost_usd
) VALUES (?1, ?2, 'anthropic', '', '', '', ?3, ?4, 0, 0, 0, 0, ?5)",
rusqlite::params![
id,
model,
input_tokens,
output_tokens,
actual_cost_usd
],
)
.unwrap();
}
#[test]
fn test_model_suffixes_resolve_only_at_identity_boundaries() {
let fast = UnifiedMessage::new(
"opencode",
"gpt-5.6-sol-fast",
"openai",
"s1",
1_700_000_000_000,
TokenBreakdown::default(),
0.0,
);
assert_eq!(fast.model_id, "gpt-5.6-sol-fast");
assert_eq!(
super::canonical_model_id_for_client(&fast.client, &fast.model_id),
"gpt-5.6-sol"
);
assert_eq!(
super::model_name_for_grouping(&fast.client, &fast.provider_id, &fast.model_id),
"gpt-5.6-sol"
);
let build = UnifiedMessage::new(
"grok",
"grok-4.5-build",
"xai",
"s1",
1_700_000_000_000,
TokenBreakdown::default(),
0.0,
);
assert_eq!(build.model_id, "grok-4.5-build");
assert_eq!(
super::canonical_model_id_for_client(&build.client, &build.model_id),
"grok-4.5"
);
assert_eq!(
super::model_name_for_grouping(&build.client, &build.provider_id, &build.model_id),
"grok-4.5"
);
let build_contributions = crate::aggregate_by_date(vec![build]);
assert_eq!(build_contributions[0].clients[0].model_id, "grok-4.5");
let base = UnifiedMessage::new(
"opencode",
"gpt-5.6-sol",
"openai",
"s1",
1_700_000_000_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.1,
);
let stripped_fast = UnifiedMessage::new(
"opencode",
"gpt-5.6-sol-fast",
"openai",
"s1",
1_700_000_100_000,
TokenBreakdown {
input: 20,
output: 10,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.2,
);
let entries = aggregate_model_usage_entries(vec![base, stripped_fast], &GroupBy::Model);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].model, "gpt-5.6-sol");
assert_eq!(entries[0].input, 30);
assert_eq!(entries[0].output, 15);
}
#[test]
fn test_opencode_fast_fallback_pricing_uses_raw_identity_before_canonical_submit() {
let mut litellm = HashMap::new();
for model_id in ["gpt-5.6-sol-fast", "openai/gpt-5.6-sol-fast"] {
litellm.insert(
model_id.to_string(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
}
for model_id in ["gpt-5.6-sol", "openai/gpt-5.6-sol"] {
litellm.insert(
model_id.to_string(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
}
let pricing = pricing::PricingService::new(litellm, HashMap::new());
for (raw_model_id, canonical_model_id) in [
("gpt-5.6-sol-fast", "gpt-5.6-sol"),
("openai/gpt-5.6-sol-fast", "openai/gpt-5.6-sol"),
] {
let mut message = UnifiedMessage::new(
"opencode",
raw_model_id,
"openai",
"session-1",
1_700_000_000_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
assert_eq!(message.model_id, raw_model_id);
apply_pricing_if_available(&mut message, Some(&pricing));
assert!((message.cost - 0.2).abs() < 1e-12);
assert_eq!(message.cost_source, crate::CostSource::Estimated);
assert_eq!(message.model_id, raw_model_id);
assert_eq!(
super::model_name_for_grouping(
&message.client,
&message.provider_id,
&message.model_id,
),
canonical_model_id
);
assert_eq!(
super::canonical_model_id_for_client(&message.client, &message.model_id),
canonical_model_id
);
let contributions = crate::aggregate_by_date(vec![message]);
assert_eq!(contributions.len(), 1);
assert_eq!(contributions[0].clients.len(), 1);
assert_eq!(contributions[0].clients[0].model_id, canonical_model_id);
}
}
#[test]
fn test_normalize_model_for_grouping() {
assert_eq!(
normalize_model_for_grouping("claude-opus-4-5-20251101"),
"claude-opus-4-5"
);
assert_eq!(
normalize_model_for_grouping("claude-sonnet-4-5-20250929"),
"claude-sonnet-4-5"
);
assert_eq!(
normalize_model_for_grouping("claude-sonnet-4-20250514"),
"claude-sonnet-4"
);
assert_eq!(
normalize_model_for_grouping("claude-opus-4.5"),
"claude-opus-4-5"
);
assert_eq!(
normalize_model_for_grouping("claude-sonnet-4.5"),
"claude-sonnet-4-5"
);
assert_eq!(
normalize_model_for_grouping("claude-opus-4.6"),
"claude-opus-4-6"
);
assert_eq!(
normalize_model_for_grouping("anthropic/claude-4-6-sonnet"),
"claude-sonnet-4-6"
);
assert_eq!(
normalize_model_for_grouping("anthropic/claude-4-5-haiku"),
"claude-haiku-4-5"
);
assert_eq!(
normalize_model_for_grouping("anthropic/claude-4-6-opus"),
"claude-opus-4-6"
);
assert_eq!(normalize_model_for_grouping("gpt-5.2"), "gpt-5.2");
assert_eq!(normalize_model_for_grouping("gpt-5.4(xhigh)"), "gpt-5.4");
assert_eq!(normalize_model_for_grouping("gpt-5.4(high)"), "gpt-5.4");
assert_eq!(normalize_model_for_grouping("gpt-5.4(minimal)"), "gpt-5.4");
assert_eq!(normalize_model_for_grouping("gpt-5.4(auto)"), "gpt-5.4");
assert_eq!(normalize_model_for_grouping("gpt-5.4(none)"), "gpt-5.4");
assert_eq!(
normalize_model_for_grouping("gpt-5.4(weirdgarbage)"),
"gpt-5.4(weirdgarbage)"
);
assert_eq!(
normalize_model_for_grouping("claude-sonnet-4.5(high)"),
"claude-sonnet-4-5"
);
assert_eq!(
normalize_model_for_grouping("gemini-3-pro(auto)"),
"gemini-3-pro"
);
assert_eq!(
normalize_model_for_grouping("gemini-2.5-pro"),
"gemini-2.5-pro"
);
assert_eq!(
normalize_model_for_grouping("claude-opus-4-5-high"),
"claude-opus-4-5-high"
);
assert_eq!(
normalize_model_for_grouping("claude-opus-4-5-thinking-high"),
"claude-opus-4-5-thinking-high"
);
assert_eq!(
normalize_model_for_grouping("claude-sonnet-4-5-high"),
"claude-sonnet-4-5-high"
);
assert_eq!(
normalize_model_for_grouping("claude-4-sonnet"),
"claude-4-sonnet"
);
assert_eq!(
normalize_model_for_grouping("claude-4-opus-thinking"),
"claude-4-opus-thinking"
);
assert_eq!(normalize_model_for_grouping("big-pickle"), "big-pickle");
assert_eq!(normalize_model_for_grouping("grok-code"), "grok-code");
assert_eq!(
normalize_model_for_grouping("claude-opus-4.5-20251101"),
"claude-opus-4-5"
);
}
#[test]
fn test_group_by_from_str_valid_values() {
assert_eq!(GroupBy::from_str("model").unwrap(), GroupBy::Model);
assert_eq!(
GroupBy::from_str("client,model").unwrap(),
GroupBy::ClientModel
);
assert_eq!(
GroupBy::from_str("client-model").unwrap(),
GroupBy::ClientModel
);
assert_eq!(
GroupBy::from_str("client,provider,model").unwrap(),
GroupBy::ClientProviderModel
);
assert_eq!(
GroupBy::from_str("client-provider-model").unwrap(),
GroupBy::ClientProviderModel
);
assert_eq!(
GroupBy::from_str("workspace,model").unwrap(),
GroupBy::WorkspaceModel
);
assert_eq!(
GroupBy::from_str("workspace-model").unwrap(),
GroupBy::WorkspaceModel
);
assert_eq!(GroupBy::from_str("session").unwrap(), GroupBy::Session);
assert_eq!(
GroupBy::from_str("session,model").unwrap(),
GroupBy::Session
);
assert_eq!(
GroupBy::from_str("session-model").unwrap(),
GroupBy::Session
);
assert_eq!(
GroupBy::from_str("client,session").unwrap(),
GroupBy::ClientSession
);
assert_eq!(
GroupBy::from_str("client,session,model").unwrap(),
GroupBy::ClientSession
);
assert_eq!(
GroupBy::from_str("client-session-model").unwrap(),
GroupBy::ClientSession
);
assert!(GroupBy::from_str("unknown").is_err());
}
#[test]
fn test_group_by_default_is_client_model() {
assert_eq!(GroupBy::default(), GroupBy::ClientModel);
}
#[test]
fn test_group_by_display_round_trips_with_from_str() {
let variants = [
GroupBy::Model,
GroupBy::ClientModel,
GroupBy::ClientProviderModel,
GroupBy::WorkspaceModel,
GroupBy::Session,
GroupBy::ClientSession,
];
for variant in variants {
let rendered = variant.to_string();
let parsed = GroupBy::from_str(&rendered).unwrap();
assert_eq!(parsed, variant);
}
}
#[test]
fn test_group_by_from_str_whitespace_handling() {
assert_eq!(
GroupBy::from_str("client, model").unwrap(),
GroupBy::ClientModel
);
assert_eq!(GroupBy::from_str(" model ").unwrap(), GroupBy::Model);
assert_eq!(
GroupBy::from_str("client , provider , model").unwrap(),
GroupBy::ClientProviderModel
);
assert_eq!(
GroupBy::from_str("workspace, model").unwrap(),
GroupBy::WorkspaceModel
);
}
#[test]
fn test_model_usage_performance_uses_only_timed_positive_token_messages() {
let mut timed = make_workspace_message(
"opencode",
"gpt-5.4",
"openai",
"session-1",
0.0,
None,
None,
);
timed.tokens = TokenBreakdown {
input: 100,
output: 50,
cache_read: 25,
cache_write: 0,
reasoning: 25,
};
timed.duration_ms = Some(400);
let mut untimed = make_workspace_message(
"opencode",
"gpt-5.4",
"openai",
"session-2",
0.0,
None,
None,
);
untimed.tokens = TokenBreakdown {
input: 300,
output: 0,
cache_read: 0,
cache_write: 0,
reasoning: 0,
};
let entries = aggregate_model_usage_entries(vec![timed, untimed], &GroupBy::ClientModel);
assert_eq!(entries.len(), 1);
let performance = &entries[0].performance;
assert_eq!(performance.total_duration_ms, 400);
assert_eq!(performance.timed_tokens, 200);
assert_eq!(performance.sample_count, 1);
assert_eq!(performance.ms_per_1k_tokens, Some(2000.0));
assert!((performance.token_coverage - 0.4).abs() < f64::EPSILON);
}
#[test]
fn test_model_usage_performance_is_null_without_duration_samples() {
let entries = aggregate_model_usage_entries(
vec![make_workspace_message(
"claude",
"claude-sonnet-4-5",
"anthropic",
"session-1",
0.0,
None,
None,
)],
&GroupBy::ClientModel,
);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].performance.ms_per_1k_tokens, None);
assert_eq!(entries[0].performance.total_duration_ms, 0);
assert_eq!(entries[0].performance.timed_tokens, 0);
assert_eq!(entries[0].performance.token_coverage, 0.0);
}
#[test]
fn test_workspace_model_grouping_merges_same_workspace_and_model() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-1",
1.25,
Some("/repo-a"),
Some("repo-a"),
),
make_workspace_message(
"qwen",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-2",
2.75,
Some("/repo-a"),
Some("repo-a"),
),
],
&GroupBy::WorkspaceModel,
);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].model, "claude-sonnet-4-5");
assert_eq!(entries[0].workspace_key.as_deref(), Some("/repo-a"));
assert_eq!(entries[0].workspace_label.as_deref(), Some("repo-a"));
assert_eq!(entries[0].cost, 4.0);
assert_eq!(entries[0].message_count, 2);
assert_eq!(entries[0].merged_clients.as_deref(), Some("claude, qwen"));
}
#[test]
fn test_model_grouping_merges_anthropic_prefixed_claude_variant_with_canonical_model() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"anthropic/claude-4-6-sonnet",
"anthropic",
"session-1",
1.25,
Some("/repo-a"),
Some("repo-a"),
),
make_workspace_message(
"claude",
"claude-sonnet-4-6",
"anthropic",
"session-2",
2.75,
Some("/repo-b"),
Some("repo-b"),
),
],
&GroupBy::ClientModel,
);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].model, "claude-sonnet-4-6");
assert_eq!(entries[0].input, 20);
assert_eq!(entries[0].output, 10);
assert_eq!(entries[0].cost, 4.0);
assert_eq!(entries[0].message_count, 2);
}
#[test]
fn test_workspace_model_grouping_separates_different_workspaces() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-1",
1.0,
Some("/repo-a"),
Some("repo-a"),
),
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-2",
2.0,
Some("/repo-b"),
Some("repo-b"),
),
],
&GroupBy::WorkspaceModel,
);
assert_eq!(entries.len(), 2);
let labels: HashSet<_> = entries
.iter()
.map(|entry| entry.workspace_label.as_deref().unwrap())
.collect();
assert_eq!(labels, HashSet::from(["repo-a", "repo-b"]));
}
#[test]
fn test_workspace_model_grouping_uses_unknown_bucket_without_workspace_metadata() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-1",
1.0,
None,
None,
),
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-2",
"2.0".parse().unwrap(),
None,
None,
),
],
&GroupBy::WorkspaceModel,
);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].workspace_key, None);
assert_eq!(
entries[0].workspace_label.as_deref(),
Some(UNKNOWN_WORKSPACE_LABEL)
);
assert_eq!(entries[0].message_count, 2);
assert_eq!(entries[0].cost, 3.0);
}
#[test]
fn test_parsed_round_trip_preserves_workspace_metadata() {
let mut unified = UnifiedMessage::new(
"qwen",
"qwen3.5-plus",
"qwen",
"session-1",
1_742_390_400_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 2,
cache_write: 0,
reasoning: 1,
},
1.25,
);
unified.set_workspace(
Some("//server/share/demo-workspace".to_string()),
Some("demo-workspace".to_string()),
);
unified.duration_ms = Some(2500);
let parsed = unified_to_parsed(&unified);
let round_tripped = parsed_to_unified(&parsed, 2.5);
assert_eq!(
round_tripped.workspace_key.as_deref(),
Some("//server/share/demo-workspace")
);
assert_eq!(
round_tripped.workspace_label.as_deref(),
Some("demo-workspace")
);
assert_eq!(round_tripped.cost, 2.5);
assert_eq!(round_tripped.duration_ms, Some(2500));
}
#[test]
fn test_workspace_model_grouping_keeps_real_unknown_workspace_separate() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-1",
1.0,
Some("unknown-workspace"),
Some("unknown-workspace"),
),
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-2",
2.0,
None,
None,
),
],
&GroupBy::WorkspaceModel,
);
assert_eq!(entries.len(), 2);
assert!(entries.iter().any(|entry| {
entry.workspace_key.as_deref() == Some("unknown-workspace")
&& entry.workspace_label.as_deref() == Some("unknown-workspace")
&& (entry.cost - 1.0).abs() < f64::EPSILON
}));
assert!(entries.iter().any(|entry| {
entry.workspace_key.is_none()
&& entry.workspace_label.as_deref() == Some(UNKNOWN_WORKSPACE_LABEL)
&& (entry.cost - 2.0).abs() < f64::EPSILON
}));
}
#[test]
fn test_session_grouping_merges_same_session_and_model() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-shared",
1.25,
None,
None,
),
make_workspace_message(
"amp",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-shared",
2.75,
None,
None,
),
],
&GroupBy::Session,
);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].session_id.as_deref(), Some("session-shared"));
assert_eq!(entries[0].model, "claude-sonnet-4-5");
assert!((entries[0].cost - 4.0).abs() < f64::EPSILON);
assert_eq!(entries[0].message_count, 2);
assert!(entries[0].workspace_key.is_none());
assert!(entries[0].workspace_label.is_none());
assert!(entries[0].merged_clients.is_none());
}
#[test]
fn test_session_grouping_separates_different_sessions() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message("codex", "gpt-5", "openai", "session-a", 1.0, None, None),
make_workspace_message("codex", "gpt-5", "openai", "session-b", 2.0, None, None),
],
&GroupBy::Session,
);
assert_eq!(entries.len(), 2);
let session_ids: HashSet<_> = entries
.iter()
.map(|e| e.session_id.as_deref().unwrap())
.collect();
assert_eq!(session_ids, HashSet::from(["session-a", "session-b"]));
}
#[test]
fn test_client_session_grouping_keeps_clients_separate() {
let entries = aggregate_model_usage_entries(
vec![
make_workspace_message(
"claude",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-shared",
1.0,
None,
None,
),
make_workspace_message(
"amp",
"claude-sonnet-4-5-20250929",
"anthropic",
"session-shared",
3.0,
None,
None,
),
],
&GroupBy::ClientSession,
);
assert_eq!(entries.len(), 2);
for entry in &entries {
assert_eq!(entry.session_id.as_deref(), Some("session-shared"));
assert!(entry.merged_clients.is_none());
}
let by_client: HashSet<_> = entries.iter().map(|e| e.client.as_str()).collect();
assert_eq!(by_client, HashSet::from(["claude", "amp"]));
}
#[test]
fn test_non_session_grouping_does_not_populate_session_id() {
for group_by in &[
GroupBy::Model,
GroupBy::ClientModel,
GroupBy::ClientProviderModel,
GroupBy::WorkspaceModel,
] {
let entries = aggregate_model_usage_entries(
vec![make_workspace_message(
"codex",
"gpt-5",
"openai",
"session-x",
1.0,
None,
None,
)],
group_by,
);
assert_eq!(entries.len(), 1);
assert!(
entries[0].session_id.is_none(),
"session_id leaked into {:?} grouping",
group_by
);
}
}
#[test]
fn test_retain_for_requested_clients_keeps_original_client_matches() {
let requested: HashSet<&str> = HashSet::from(["opencode"]);
assert!(retain_for_requested_clients(
"opencode",
"gpt-4o",
"anthropic",
&requested
));
assert!(!retain_for_requested_clients(
"claude",
"gpt-4o",
"anthropic",
&requested
));
}
#[test]
fn test_retain_for_requested_clients_accepts_synthetic_gateway_traffic() {
let requested: HashSet<&str> = HashSet::from(["synthetic"]);
assert!(retain_for_requested_clients(
"opencode",
"hf:deepseek-ai/DeepSeek-V3-0324",
"unknown",
&requested
));
assert!(retain_for_requested_clients(
"synthetic",
"deepseek-v3-0324",
"synthetic",
&requested
));
assert!(!retain_for_requested_clients(
"opencode",
"gpt-4o",
"anthropic",
&requested
));
}
#[test]
fn test_retain_for_requested_clients_preserves_kilo_split() {
let kilocode_only: HashSet<&str> = HashSet::from(["kilocode"]);
assert!(retain_for_requested_clients(
"kilocode",
"gpt-5",
"openai",
&kilocode_only
));
assert!(!retain_for_requested_clients(
"kilo",
"gpt-5",
"openai",
&kilocode_only
));
let kilo_only: HashSet<&str> = HashSet::from(["kilo"]);
assert!(retain_for_requested_clients(
"kilo", "gpt-5", "openai", &kilo_only
));
assert!(!retain_for_requested_clients(
"kilocode", "gpt-5", "openai", &kilo_only
));
}
#[test]
fn test_cursor_parse_path_preserves_reported_zero_cost() {
let temp_dir = tempfile::TempDir::new().unwrap();
let cursor_cache_dir = temp_dir.path().join(".config/tokmesh/cursor-cache");
std::fs::create_dir_all(&cursor_cache_dir).unwrap();
let csv = r#"Date,Kind,Model,Max Mode,Input (w/ Cache Write),Input (w/o Cache Write),Cache Read,Output Tokens,Total Tokens,Cost
"2026-03-04T12:00:00.000Z","Included","Composer 1.5","No","1200","1000","5000","2000","8000","0""#;
std::fs::write(cursor_cache_dir.join("usage.csv"), csv).unwrap();
let pricing = pricing::PricingService::new(HashMap::new(), HashMap::new());
let messages = parse_all_messages_with_pricing(
temp_dir.path().to_str().unwrap(),
&["cursor".to_string()],
Some(&pricing),
);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].client, "cursor");
assert_eq!(messages[0].model_id, "Composer 1.5");
assert_eq!(messages[0].cost, 0.0);
assert_eq!(messages[0].cost_source, crate::CostSource::ProviderReported);
}
#[test]
#[serial_test::serial]
fn test_micode_authoritative_cost_is_not_repriced_on_first_parse_or_cache_hit() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let micode_dir = source_home.path().join(".local/share/mimocode");
std::fs::create_dir_all(&micode_dir).unwrap();
let db_path = micode_dir.join("mimocode.db");
let conn = rusqlite::Connection::open(&db_path).unwrap();
conn.execute_batch(
"CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.unwrap();
let data_json = r#"{
"role": "assistant",
"modelID": "mimo-v2.5-pro",
"providerID": "mimo",
"cost": 0.05,
"tokens": { "input": 1000, "output": 500, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
"time": { "created": 1700000000000.0 }
}"#;
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params!["msg_auth_cost", "ses_1", data_json],
)
.unwrap();
drop(conn);
let mut litellm = HashMap::new();
litellm.insert(
"mimo-v2.5-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let first = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["micode".to_string()],
Some(&pricing),
);
assert_eq!(first.len(), 1);
assert!(
(first[0].cost - 0.05).abs() < 1e-9,
"authoritative cost must survive the first parse, got {}",
first[0].cost
);
let second = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["micode".to_string()],
Some(&pricing),
);
assert_eq!(second.len(), 1);
assert!(
(second[0].cost - 0.05).abs() < 1e-9,
"authoritative cost must survive the cache hit, got {}",
second[0].cost
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
fn write_kimi_repeated_status_fixture(source_home: &std::path::Path) {
let session_dir = source_home.join(".kimi/sessions/group-1/session-1");
std::fs::create_dir_all(&session_dir).unwrap();
std::fs::write(
session_dir.join("wire.jsonl"),
r#"{"type": "metadata", "protocol_version": "1.3"}
{"timestamp": 1770983410.0, "message": {"type": "StatusUpdate", "payload": {"token_usage": {"input_other": 10, "output": 1, "input_cache_read": 0, "input_cache_creation": 0}, "message_id": "msg-progressive"}}}
{"timestamp": 1770983420.0, "message": {"type": "StatusUpdate", "payload": {"token_usage": {"input_other": 20, "output": 2, "input_cache_read": 0, "input_cache_creation": 0}, "message_id": "msg-progressive"}}}
{"timestamp": 1770983430.0, "message": {"type": "StatusUpdate", "payload": {"token_usage": {"input_other": 5, "output": 1, "input_cache_read": 0, "input_cache_creation": 0}, "message_id": "msg-distinct"}}}
{"timestamp": 1770983440.0, "message": {"type": "StatusUpdate", "payload": {"token_usage": {"input_other": 7, "output": 1, "input_cache_read": 0, "input_cache_creation": 0}}}}
{"timestamp": 1770983450.0, "message": {"type": "StatusUpdate", "payload": {"token_usage": {"input_other": 8, "output": 1, "input_cache_read": 0, "input_cache_creation": 0}}}}"#,
)
.unwrap();
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_kimi_deduplicates_repeated_status_updates() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_kimi_repeated_status_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["kimi".to_string()],
None,
);
assert_eq!(messages.len(), 4);
assert_eq!(messages.iter().map(|m| m.tokens.input).sum::<i64>(), 40);
assert_eq!(messages.iter().map(|m| m.tokens.output).sum::<i64>(), 5);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_local_clients_kimi_deduplicates_repeated_status_updates() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_kimi_repeated_status_fixture(source_home.path());
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(source_home.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["kimi".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Kimi), 4);
assert_eq!(parsed.messages.len(), 4);
assert_eq!(parsed.messages.iter().map(|m| m.input).sum::<i64>(), 40);
assert_eq!(parsed.messages.iter().map(|m| m.output).sum::<i64>(), 5);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
#[cfg(any(target_os = "macos", target_os = "windows"))]
fn test_parse_local_clients_kimi_work_uses_kimi_code_parser() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let work_root = if cfg!(target_os = "macos") {
source_home
.path()
.join("Library")
.join("Application Support")
} else {
source_home.path().join("AppData").join("Roaming")
};
let wire = work_root
.join("kimi-desktop")
.join("daimon-share")
.join("daimon")
.join("runtime")
.join("kimi-code")
.join("home")
.join("sessions")
.join("wd_workspace_c107cac82a87")
.join("conv-4e171339d10b9954d0fc24da")
.join("agents")
.join("main")
.join("wire.jsonl");
std::fs::create_dir_all(wire.parent().unwrap()).unwrap();
std::fs::write(
&wire,
r#"{"type":"usage.record","model":"k2d6-agent","usage":{"inputOther":100,"output":10,"inputCacheRead":20,"inputCacheCreation":0},"usageScope":"turn","time":1780319377010}"#,
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(source_home.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["kimi".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Kimi), 1);
assert_eq!(parsed.messages.len(), 1);
let message = &parsed.messages[0];
assert_eq!(message.session_id, "conv-4e171339d10b9954d0fc24da");
assert_eq!(message.model_id, "k2d6-agent");
assert_eq!(message.input, 100);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_source_cache_refreshes_stale_date_on_cache_hit() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let message_dir = source_home
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
let path = message_dir.join("msg_001.json");
std::fs::write(
&path,
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let fingerprint = message_cache::SourceFingerprint::from_path(&path).unwrap();
let mut stale_message = UnifiedMessage::new(
"opencode",
"accounts/fireworks/models/deepseek-v3-0324",
"fireworks",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
stale_message.date = "1900-01-01".to_string();
let mut cache = message_cache::SourceMessageCache::default();
cache.insert(message_cache::CachedSourceEntry::new(
message_cache::CacheIdentity::for_client(ClientId::OpenCode),
&path,
fingerprint,
vec![stale_message],
Vec::new(),
None,
));
cache.save_if_dirty();
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(messages.len(), 1);
assert_ne!(messages[0].date, "1900-01-01");
assert_eq!(
messages[0].date,
UnifiedMessage::new(
"opencode",
"accounts/fireworks/models/deepseek-v3-0324",
"fireworks",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
)
.date
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[cfg(unix)]
#[test]
#[serial_test::serial]
fn test_empty_parse_results_are_not_cached_for_optional_file_sources() {
use std::os::unix::fs::PermissionsExt;
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let message_dir = source_home
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
let path = message_dir.join("msg_001.json");
std::fs::write(
&path,
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let mut permissions = std::fs::metadata(&path).unwrap().permissions();
permissions.set_mode(0o000);
std::fs::set_permissions(&path, permissions).unwrap();
let first_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert!(first_messages.is_empty());
let cache = message_cache::SourceMessageCache::load();
assert!(cache
.get(
message_cache::CacheIdentity::for_client(ClientId::OpenCode),
&path,
)
.is_none());
let mut readable_permissions = std::fs::metadata(&path).unwrap().permissions();
readable_permissions.set_mode(0o644);
std::fs::set_permissions(&path, readable_permissions).unwrap();
let second_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(second_messages.len(), 1);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_empty_cache_hits_are_reparsed_for_optional_file_sources() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let message_dir = source_home
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
let path = message_dir.join("msg_001.json");
std::fs::write(
&path,
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let fingerprint = message_cache::SourceFingerprint::from_path(&path).unwrap();
let mut cache = message_cache::SourceMessageCache::default();
cache.insert(message_cache::CachedSourceEntry::new(
message_cache::CacheIdentity::for_client(ClientId::OpenCode),
&path,
fingerprint,
Vec::new(),
Vec::new(),
None,
));
cache.save_if_dirty();
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(messages.len(), 1);
let loaded = message_cache::SourceMessageCache::load();
let repaired_entry = loaded
.get(
message_cache::CacheIdentity::for_client(ClientId::OpenCode),
&path,
)
.unwrap();
assert_eq!(repaired_entry.messages.len(), 1);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_sqlite_source_cache_invalidates_on_wal_change() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let db_dir = source_home.path().join(".local/share/opencode");
std::fs::create_dir_all(&db_dir).unwrap();
let db_path = db_dir.join("opencode.db");
let conn = rusqlite::Connection::open(&db_path).unwrap();
let journal_mode: String = conn
.query_row("PRAGMA journal_mode=WAL;", [], |row| row.get(0))
.unwrap();
assert_eq!(journal_mode.to_lowercase(), "wal");
conn.execute_batch(
"PRAGMA wal_autocheckpoint=0;
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.unwrap();
let row_one = r#"{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"tokens": { "input": 100, "output": 50, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
"time": { "created": 1700000000000.0 }
}"#;
let row_two = r#"{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"tokens": { "input": 120, "output": 60, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
"time": { "created": 1700000001000.0 }
}"#;
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params!["msg-1", "session-1", row_one],
)
.unwrap();
let first_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(first_messages.len(), 1);
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params!["msg-2", "session-1", row_two],
)
.unwrap();
assert!(db_path.with_extension("db-wal").exists());
let refreshed_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(refreshed_messages.len(), 2);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_dedups_across_channel_suffixed_opencode_dbs() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let db_dir = source_home.path().join(".local/share/opencode");
std::fs::create_dir_all(&db_dir).unwrap();
let schema = "PRAGMA journal_mode=WAL;
PRAGMA wal_autocheckpoint=0;
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);";
let row = |input: u64, ts: u64| {
format!(
r#"{{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"tokens": {{ "input": {input}, "output": 10, "reasoning": 0, "cache": {{ "read": 0, "write": 0 }} }},
"time": {{ "created": {ts}.0 }}
}}"#
)
};
let default_db = db_dir.join("opencode.db");
let conn = rusqlite::Connection::open(&default_db).unwrap();
conn.execute_batch(schema).unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"shared-msg",
"session-shared",
row(100, 1_700_000_000_000u64)
],
)
.unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"latest-only",
"session-latest",
row(200, 1_700_000_001_000u64)
],
)
.unwrap();
drop(conn);
let stable_db = db_dir.join("opencode-stable.db");
let conn = rusqlite::Connection::open(&stable_db).unwrap();
conn.execute_batch(schema).unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"shared-msg",
"session-shared",
row(100, 1_700_000_000_000u64)
],
)
.unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"stable-only",
"session-stable",
row(300, 1_700_000_002_000u64)
],
)
.unwrap();
drop(conn);
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(
messages.len(),
3,
"expected 3 unique messages (shared + latest-only + stable-only), got {}",
messages.len()
);
let mut ids: Vec<String> = messages
.iter()
.filter_map(|m| m.dedup_key.clone())
.collect();
ids.sort();
assert_eq!(ids, vec!["latest-only", "shared-msg", "stable-only"]);
let messages_warm = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(
messages_warm.len(),
3,
"warm cache must also dedup shared message across channel dbs"
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_opencode_sqlite_deduplicates_forked_history() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let db_dir = source_home.path().join(".local/share/opencode");
std::fs::create_dir_all(&db_dir).unwrap();
let db_path = db_dir.join("opencode.db");
let conn = create_opencode_sqlite_db(&db_path);
let msg_a = build_opencode_sqlite_payload(
1_700_000_000_000.0,
1_700_000_000_500.0,
100,
50,
0,
10,
5,
0.01,
);
let msg_b = build_opencode_sqlite_payload(
1_700_000_001_000.0,
1_700_000_001_500.0,
200,
80,
10,
20,
0,
0.02,
);
let msg_c = build_opencode_sqlite_payload(
1_700_000_002_000.0,
1_700_000_002_500.0,
300,
120,
15,
0,
0,
0.03,
);
for (id, session_id, payload) in [
("root_a", "root", msg_a.as_str()),
("root_b", "root", msg_b.as_str()),
("fork_a_copy", "fork", msg_a.as_str()),
("fork_b_copy", "fork", msg_b.as_str()),
("fork_c_new", "fork", msg_c.as_str()),
] {
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![id, session_id, payload],
)
.unwrap();
}
drop(conn);
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["opencode".to_string()],
None,
);
assert_eq!(messages.len(), 3);
assert_eq!(messages.iter().map(|m| m.tokens.input).sum::<i64>(), 600);
assert_eq!(messages.iter().map(|m| m.tokens.output).sum::<i64>(), 250);
assert_eq!(messages.iter().map(|m| m.cost).sum::<f64>(), 0.06);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_local_clients_opencode_sqlite_counts_deduplicated_forked_history() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let db_dir = source_home.path().join(".local/share/opencode");
std::fs::create_dir_all(&db_dir).unwrap();
let db_path = db_dir.join("opencode.db");
let conn = create_opencode_sqlite_db(&db_path);
let msg_a = build_opencode_sqlite_payload(
1_700_000_000_000.0,
1_700_000_000_500.0,
100,
50,
0,
10,
5,
0.01,
);
let msg_b = build_opencode_sqlite_payload(
1_700_000_001_000.0,
1_700_000_001_500.0,
200,
80,
10,
20,
0,
0.02,
);
let msg_c = build_opencode_sqlite_payload(
1_700_000_002_000.0,
1_700_000_002_500.0,
300,
120,
15,
0,
0,
0.03,
);
for (id, session_id, payload) in [
("root_a", "root", msg_a.as_str()),
("root_b", "root", msg_b.as_str()),
("fork_a_copy", "fork", msg_a.as_str()),
("fork_b_copy", "fork", msg_b.as_str()),
("fork_c_new", "fork", msg_c.as_str()),
] {
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![id, session_id, payload],
)
.unwrap();
}
drop(conn);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(source_home.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["opencode".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::OpenCode), 3);
assert_eq!(parsed.messages.len(), 3);
assert_eq!(parsed.messages.iter().map(|m| m.input).sum::<i64>(), 600);
assert_eq!(parsed.messages.iter().map(|m| m.output).sum::<i64>(), 250);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
fn write_codex_forked_history_fixture(source_home: &std::path::Path) {
let codex_dir = source_home.join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
std::fs::write(
codex_dir.join("parent.jsonl"),
concat!(
r#"{"timestamp":"2026-04-30T10:00:00Z","type":"session_meta","payload":{"id":"parent-session","source":"interactive","model_provider":"openai","cwd":"/Users/alice/root"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:00:01Z","type":"turn_context","payload":{"model":"gpt-5.2"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65},"last_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65}}}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:00:03Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"cached_input_tokens":20,"output_tokens":30,"total_tokens":130},"last_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65}}}}"#,
"\n"
),
)
.unwrap();
std::fs::write(
codex_dir.join("fork.jsonl"),
concat!(
r#"{"timestamp":"2026-04-30T10:01:00Z","type":"session_meta","payload":{"id":"fork-session","source":{"subagent":{"thread_spawn":{"parent_thread_id":"parent-session","depth":1}}},"model_provider":"openai","cwd":"/Users/alice/root-worktree"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:01:01Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"cached_input_tokens":20,"output_tokens":30,"total_tokens":130},"last_token_usage":{"input_tokens":100,"cached_input_tokens":20,"output_tokens":30,"total_tokens":130}}}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:01:02Z","type":"turn_context","payload":{"model":"gpt-5.2"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:01:03Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65},"last_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65}}}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:01:04Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"cached_input_tokens":20,"output_tokens":30,"total_tokens":130},"last_token_usage":{"input_tokens":50,"cached_input_tokens":10,"output_tokens":15,"total_tokens":65}}}}"#,
"\n",
r#"{"timestamp":"2026-04-30T10:01:05Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":110,"cached_input_tokens":22,"output_tokens":33,"total_tokens":143},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3,"total_tokens":13}}}}"#,
"\n"
),
)
.unwrap();
}
fn write_codex_parent_replay_fixture(source_home: &std::path::Path) {
let codex_dir = source_home.join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
std::fs::write(
codex_dir.join("parent.jsonl"),
concat!(
r#"{"timestamp":"2026-05-24T20:00:00Z","type":"session_meta","payload":{"id":"019e5b00-0000-7000-8000-000000000001","source":"vscode","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-05-24T20:00:01Z","type":"turn_context","payload":{"turn_id":"019e5b00-0001-7000-8000-000000000001","model":"gpt-5.5","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-05-24T20:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"output_tokens":10,"total_tokens":110},"last_token_usage":{"input_tokens":100,"output_tokens":10,"total_tokens":110}}}}"#,
"\n",
r#"{"timestamp":"2026-05-24T20:00:03Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":130,"output_tokens":13,"total_tokens":143},"last_token_usage":{"input_tokens":30,"output_tokens":3,"total_tokens":33}}}}"#,
"\n"
),
)
.unwrap();
for (filename, child_id, child_turn_id, timestamp) in [
(
"child-a.jsonl",
"019e5c03-1e99-7000-8000-000000000001",
"019e5c03-6425-7000-8000-000000000001",
"2026-05-24T21:00:00Z",
),
(
"child-b.jsonl",
"019e5c04-1e99-7000-8000-000000000001",
"019e5c04-6425-7000-8000-000000000001",
"2026-05-24T22:00:00Z",
),
] {
std::fs::write(
codex_dir.join(filename),
format!(
concat!(
r#"{{"timestamp":"{timestamp}","type":"session_meta","payload":{{"id":"{child_id}","forked_from_id":"019e5b00-0000-7000-8000-000000000001","source":{{"subagent":{{"thread_spawn":{{"parent_thread_id":"019e5b00-0000-7000-8000-000000000001","depth":1}}}}}},"model_provider":"openai","agent_nickname":"worker","cwd":"/repo"}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"session_meta","payload":{{"id":"019e5b00-0000-7000-8000-000000000001","source":"vscode","model_provider":"openai","cwd":"/repo"}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"turn_context","payload":{{"turn_id":"019e5b00-0001-7000-8000-000000000001","model":"gpt-5.5","cwd":"/repo"}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"event_msg","payload":{{"type":"token_count","info":{{"total_token_usage":{{"input_tokens":100,"output_tokens":10,"total_tokens":110}},"last_token_usage":{{"input_tokens":100,"output_tokens":10,"total_tokens":110}}}}}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"event_msg","payload":{{"type":"token_count","info":{{"total_token_usage":{{"input_tokens":130,"output_tokens":13,"total_tokens":143}},"last_token_usage":{{"input_tokens":30,"output_tokens":3,"total_tokens":33}}}}}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"event_msg","payload":{{"type":"task_started","turn_id":"{child_turn_id}"}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"turn_context","payload":{{"turn_id":"{child_turn_id}","model":"gpt-5.5","cwd":"/repo"}}}}"#,
"\n",
r#"{{"timestamp":"{timestamp}","type":"event_msg","payload":{{"type":"token_count","info":{{"total_token_usage":{{"input_tokens":140,"output_tokens":14,"total_tokens":154}},"last_token_usage":{{"input_tokens":10,"output_tokens":1,"total_tokens":11}}}}}}}}"#,
"\n",
),
timestamp = timestamp,
child_id = child_id,
child_turn_id = child_turn_id,
),
)
.unwrap();
}
}
fn write_codex_user_fork_replay_fixture(source_home: &std::path::Path) {
let sessions_dir = source_home.join(".codex/sessions/2026/01/02");
let archived_dir = source_home.join(".codex/archived_sessions");
std::fs::create_dir_all(&sessions_dir).unwrap();
std::fs::create_dir_all(&archived_dir).unwrap();
std::fs::write(
archived_dir.join("rollout-2026-01-02T03-04-05-11111111-1111-7111-8111-111111111111.jsonl"),
concat!(
r#"{"timestamp":"2026-01-02T03:04:05Z","type":"session_meta","payload":{"id":"11111111-1111-7111-8111-111111111111","source":"vscode","thread_source":"user","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:04:06Z","type":"turn_context","payload":{"turn_id":"11111111-3333-7333-8333-333333333333","model":"gpt-5.5","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:04:07Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1000,"cached_input_tokens":400,"output_tokens":100,"total_tokens":1100},"last_token_usage":{"input_tokens":1000,"cached_input_tokens":400,"output_tokens":100,"total_tokens":1100}}}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:04:08Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1200,"cached_input_tokens":450,"output_tokens":120,"total_tokens":1320},"last_token_usage":{"input_tokens":200,"cached_input_tokens":50,"output_tokens":20,"total_tokens":220}}}}"#,
"\n"
),
)
.unwrap();
std::fs::write(
sessions_dir.join("rollout-2026-01-02T03-10-00-22222222-2222-7222-8222-222222222222.jsonl"),
concat!(
r#"{"timestamp":"2026-01-02T03:10:00Z","type":"session_meta","payload":{"id":"22222222-2222-7222-8222-222222222222","forked_from_id":"11111111-1111-7111-8111-111111111111","source":"vscode","thread_source":"user","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:00Z","type":"session_meta","payload":{"id":"11111111-1111-7111-8111-111111111111","source":"vscode","thread_source":"user","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:00Z","type":"turn_context","payload":{"turn_id":"11111111-3333-7333-8333-333333333333","model":"gpt-5.5","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:00Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1000,"cached_input_tokens":400,"output_tokens":100,"total_tokens":1100},"last_token_usage":{"input_tokens":1000,"cached_input_tokens":400,"output_tokens":100,"total_tokens":1100}}}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:00Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1200,"cached_input_tokens":450,"output_tokens":120,"total_tokens":1320},"last_token_usage":{"input_tokens":200,"cached_input_tokens":50,"output_tokens":20,"total_tokens":220}}}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:30Z","type":"turn_context","payload":{"turn_id":"22222222-4444-7444-8444-444444444444","model":"gpt-5.5","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:30Z","type":"session_meta","payload":{"id":"22222222-2222-7222-8222-222222222222","forked_from_id":"11111111-1111-7111-8111-111111111111","source":"vscode","thread_source":"user","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-01-02T03:10:53Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1500,"cached_input_tokens":500,"output_tokens":150,"total_tokens":1650},"last_token_usage":{"input_tokens":300,"cached_input_tokens":50,"output_tokens":30,"total_tokens":330}}}}"#,
"\n"
),
)
.unwrap();
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_codex_deduplicates_forked_history() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_forked_history_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(messages.len(), 3);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.input)
.sum::<i64>(),
88
);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.cache_read)
.sum::<i64>(),
22
);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.output)
.sum::<i64>(),
33
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_codex_keeps_user_fork_own_turn() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_user_fork_replay_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
let session_ids: HashSet<_> = messages
.iter()
.map(|message| message.session_id.as_str())
.collect();
assert!(session_ids
.contains("rollout-2026-01-02T03-10-00-22222222-2222-7222-8222-222222222222"));
assert_eq!(messages.iter().map(|m| m.tokens.input).sum::<i64>(), 1000);
assert_eq!(
messages.iter().map(|m| m.tokens.cache_read).sum::<i64>(),
500
);
assert_eq!(messages.iter().map(|m| m.tokens.output).sum::<i64>(), 150);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
fn write_codex_sessions_and_archived_sessions_fixture(source_home: &std::path::Path) {
let sessions_dir = source_home.join(".codex/sessions");
let archived_dir = source_home.join(".codex/archived_sessions");
std::fs::create_dir_all(&sessions_dir).unwrap();
std::fs::create_dir_all(&archived_dir).unwrap();
std::fs::write(
sessions_dir.join("live-only.jsonl"),
concat!(
r#"{"timestamp":"2026-06-25T10:00:00Z","type":"session_meta","payload":{"id":"33333333-3333-7333-8333-333333333333","source":"interactive","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-06-25T10:00:01Z","type":"turn_context","payload":{"model":"gpt-5.5"}}"#,
"\n",
r#"{"timestamp":"2026-06-25T10:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":50,"output_tokens":5,"total_tokens":55},"last_token_usage":{"input_tokens":50,"output_tokens":5,"total_tokens":55}}}}"#,
"\n"
),
)
.unwrap();
std::fs::write(
archived_dir.join("archived-only.jsonl"),
concat!(
r#"{"timestamp":"2026-06-20T09:00:00Z","type":"session_meta","payload":{"id":"44444444-4444-7444-8444-444444444444","source":"interactive","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-06-20T09:00:01Z","type":"turn_context","payload":{"model":"gpt-5.5"}}"#,
"\n",
r#"{"timestamp":"2026-06-20T09:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":70,"output_tokens":7,"total_tokens":77},"last_token_usage":{"input_tokens":70,"output_tokens":7,"total_tokens":77}}}}"#,
"\n"
),
)
.unwrap();
let shared_content = concat!(
r#"{"timestamp":"2026-06-22T08:00:00Z","type":"session_meta","payload":{"id":"55555555-5555-7555-8555-555555555555","source":"interactive","model_provider":"openai","cwd":"/repo"}}"#,
"\n",
r#"{"timestamp":"2026-06-22T08:00:01Z","type":"turn_context","payload":{"model":"gpt-5.5"}}"#,
"\n",
r#"{"timestamp":"2026-06-22T08:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":30,"output_tokens":3,"total_tokens":33},"last_token_usage":{"input_tokens":30,"output_tokens":3,"total_tokens":33}}}}"#,
"\n"
);
std::fs::write(
sessions_dir.join("shared-in-sessions.jsonl"),
shared_content,
)
.unwrap();
std::fs::write(
archived_dir.join("shared-in-archived.jsonl"),
shared_content,
)
.unwrap();
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_codex_scans_archived_sessions_without_double_counting()
{
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_sessions_and_archived_sessions_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(
messages.len(),
3,
"archived_sessions must be scanned (live-only + archived-only), and a session \
present in both sessions/ and archived_sessions/ must be deduplicated to one \
message, not counted twice"
);
let session_ids: HashSet<_> = messages
.iter()
.map(|message| message.session_id.as_str())
.collect();
assert!(session_ids.contains("live-only"));
assert!(
session_ids.contains("archived-only"),
"archived_sessions/archived-only.jsonl must be scanned and parsed"
);
assert_eq!(messages.iter().map(|m| m.tokens.input).sum::<i64>(), 150);
assert_eq!(messages.iter().map(|m| m.tokens.output).sum::<i64>(), 15);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_codex_deduplicates_parent_replay_across_forks() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_parent_replay_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(messages.len(), 3);
assert_eq!(messages.iter().map(|m| m.tokens.input).sum::<i64>(), 140);
assert_eq!(messages.iter().map(|m| m.tokens.output).sum::<i64>(), 14);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
fn write_codex_twin_token_count_fixture(source_home: &std::path::Path) {
let codex_dir = source_home.join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
std::fs::write(
codex_dir.join("twin-deltas.jsonl"),
concat!(
r#"{"timestamp":"2026-04-30T11:00:00Z","type":"session_meta","payload":{"id":"twin-session","source":"interactive","model_provider":"openai","cwd":"/Users/alice/root"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T11:00:01Z","type":"turn_context","payload":{"model":"gpt-5.2"}}"#,
"\n",
r#"{"timestamp":"2026-04-30T11:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n",
r#"{"timestamp":"2026-04-30T11:00:03Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":20,"cached_input_tokens":4,"output_tokens":6},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
),
)
.unwrap();
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_with_pricing_codex_keeps_twin_token_counts_at_distinct_timestamps() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_twin_token_count_fixture(source_home.path());
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(
messages.len(),
2,
"two turns with identical token deltas at distinct timestamps must both survive dedup",
);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.input)
.sum::<i64>(),
16,
"input tokens normalize cache_read out of input: 2 turns × (10 - 2) = 16",
);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.output)
.sum::<i64>(),
6,
);
assert_eq!(
messages
.iter()
.map(|message| message.tokens.cache_read)
.sum::<i64>(),
4,
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_local_clients_codex_counts_deduplicated_forked_history() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
write_codex_forked_history_fixture(source_home.path());
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(source_home.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["codex".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Codex), 3);
assert_eq!(parsed.messages.len(), 3);
assert_eq!(
parsed
.messages
.iter()
.map(|message| message.input)
.sum::<i64>(),
88
);
assert_eq!(
parsed
.messages
.iter()
.map(|message| message.cache_read)
.sum::<i64>(),
22
);
assert_eq!(
parsed
.messages
.iter()
.map(|message| message.output)
.sum::<i64>(),
33
);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_codex_cache_reparses_from_zero_when_incremental_prefix_is_stale() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let codex_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
let path = codex_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
),
)
.unwrap();
let initial_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(initial_messages.len(), 1);
assert_eq!(initial_messages[0].model_id, "gpt-5.4");
assert!(message_cache::SourceMessageCache::load()
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.and_then(|entry| entry.codex_incremental.as_ref())
.is_some());
std::thread::sleep(std::time::Duration::from_millis(5));
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.5"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":15,"cached_input_tokens":3,"output_tokens":5},"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#,
"\n"
),
)
.unwrap();
let warm_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(warm_messages, fresh_messages);
assert_eq!(warm_messages.len(), 2);
assert!(warm_messages
.iter()
.all(|message| message.model_id == "gpt-5.5"));
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_source_cache_keeps_untimestamped_rows_in_sync_after_append() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let codex_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
let path = codex_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
),
)
.unwrap();
let first_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(first_messages.len(), 1);
std::thread::sleep(std::time::Duration::from_millis(5));
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
file.write_all(
concat!(
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":15,"cached_input_tokens":3,"output_tokens":5},"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#,
"\n"
)
.as_bytes(),
)
.unwrap();
file.flush().unwrap();
let warm_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(warm_messages, fresh_messages);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_source_cache_matches_cold_parse_after_malformed_json_append() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let codex_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&codex_dir).unwrap();
let path = codex_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":999""#,
"\n"
),
)
.unwrap();
let initial_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(initial_messages.len(), 1);
std::thread::sleep(std::time::Duration::from_millis(5));
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
file.write_all(
concat!(
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":15,"cached_input_tokens":3,"output_tokens":5},"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#,
"\n"
)
.as_bytes(),
)
.unwrap();
file.flush().unwrap();
let warm_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert!(message_cache::SourceMessageCache::load()
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.is_none());
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(warm_messages, fresh_messages);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_exact_hit_codex_cache_repairs_fallback_timestamps_without_incremental_state() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let session_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&session_dir).unwrap();
let path = session_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
),
)
.unwrap();
let expected = crate::sessions::codex::parse_codex_file(&path);
assert_eq!(expected.len(), 1);
let fingerprint = message_cache::SourceFingerprint::from_path(&path).unwrap();
let mut stale_message = expected[0].clone();
stale_message.timestamp = 0;
stale_message.date = "1900-01-01".to_string();
let mut cache = message_cache::SourceMessageCache::default();
cache.insert(message_cache::CachedSourceEntry::new(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
fingerprint,
vec![stale_message],
vec![0],
None,
));
cache.save_if_dirty();
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(messages, expected);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_codex_cache_repairs_fallback_timestamps_after_source_mtime_change() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let session_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&session_dir).unwrap();
let path = session_dir.join("session.jsonl");
let contents = concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
);
std::fs::write(&path, contents).unwrap();
let initial_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(initial_messages.len(), 1);
std::thread::sleep(std::time::Duration::from_millis(20));
std::fs::write(&path, contents).unwrap();
let warm_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(warm_messages, fresh_messages);
assert_ne!(warm_messages[0].timestamp, initial_messages[0].timestamp);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_full_log_parse_preserves_valid_messages_before_invalid_line_error() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let session_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&session_dir).unwrap();
let path = session_dir.join("session.jsonl");
let mut file = std::fs::File::create(&path).unwrap();
file.write_all(
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
)
.as_bytes(),
)
.unwrap();
file.write_all(&[0xff, b'\n']).unwrap();
file.flush().unwrap();
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].model_id, "gpt-5.4");
let cache = message_cache::SourceMessageCache::load();
assert!(cache
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.is_none());
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_codex_cache_does_not_persist_unknown_before_later_turn_context() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let session_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&session_dir).unwrap();
let path = session_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"session_meta","payload":{"source":"interactive","model_provider":"openai"}}"#,
"\n",
r#"{"timestamp":"2026-04-27T10:00:00Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#,
"\n"
),
)
.unwrap();
let initial_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(initial_messages.len(), 1);
assert_eq!(initial_messages[0].model_id, "unknown");
assert!(message_cache::SourceMessageCache::load()
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.is_none());
std::thread::sleep(std::time::Duration::from_millis(5));
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
file.write_all(
concat!(
r#"{"timestamp":"2026-04-27T10:00:04Z","type":"turn_context","payload":{"model":"gpt-5.5"}}"#,
"\n"
)
.as_bytes(),
)
.unwrap();
file.flush().unwrap();
let resumed_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(resumed_messages, fresh_messages);
assert_eq!(resumed_messages.len(), 1);
assert_eq!(resumed_messages[0].model_id, "gpt-5.5");
std::env::set_var("HOME", cache_home.path());
assert!(message_cache::SourceMessageCache::load()
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.is_some());
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_codex_cache_skips_non_newline_terminated_resume_prefix() {
let cache_home = tempfile::TempDir::new().unwrap();
let fresh_cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let session_dir = source_home.path().join(".codex/sessions");
std::fs::create_dir_all(&session_dir).unwrap();
let path = session_dir.join("session.jsonl");
std::fs::write(
&path,
concat!(
r#"{"type":"turn_context","payload":{"model":"gpt-5.4"}}"#,
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":3}}}}"#
),
)
.unwrap();
let initial_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(initial_messages.len(), 1);
assert!(message_cache::SourceMessageCache::load()
.get(
message_cache::CacheIdentity::for_client(ClientId::Codex),
&path,
)
.is_none());
std::thread::sleep(std::time::Duration::from_millis(5));
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
file.write_all(
concat!(
"\n",
r#"{"type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":15,"cached_input_tokens":3,"output_tokens":5},"last_token_usage":{"input_tokens":5,"cached_input_tokens":1,"output_tokens":2}}}}"#,
"\n"
)
.as_bytes(),
)
.unwrap();
file.flush().unwrap();
let warm_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
std::env::set_var("HOME", fresh_cache_home.path());
let fresh_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["codex".to_string()],
None,
);
assert_eq!(warm_messages, fresh_messages);
assert_eq!(warm_messages.len(), 2);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_source_cache_does_not_reuse_priced_cost_without_pricing_service() {
let temp_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", temp_home.path());
{
let cursor_cache_dir = source_home.path().join(".config/tokmesh/cursor-cache");
std::fs::create_dir_all(&cursor_cache_dir).unwrap();
let csv = r#"Date,Kind,Model,Max Mode,Input (w/ Cache Write),Input (w/o Cache Write),Cache Read,Output Tokens,Total Tokens,Cost
"2026-03-04T12:00:00.000Z","Included","Composer 1.5","No","1200","1000","5000","2000","8000","Included""#;
std::fs::write(cursor_cache_dir.join("usage.csv"), csv).unwrap();
let mut litellm = HashMap::new();
litellm.insert(
"Composer 1.5".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
cache_read_input_token_cost: Some(0.0005),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let repriced_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["cursor".to_string()],
Some(&pricing),
);
assert_eq!(repriced_messages.len(), 1);
assert!(repriced_messages[0].cost > 0.0);
let cached_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["cursor".to_string()],
None,
);
assert_eq!(cached_messages.len(), 1);
assert_eq!(cached_messages[0].cost, 0.0);
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
fn test_apply_pricing_if_available_keeps_existing_cost_without_pricing() {
let mut msg = UnifiedMessage::new_with_agent(
"roocode",
"gpt-4o",
"provider",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.42,
Some("planner".to_string()),
);
apply_pricing_if_available(&mut msg, None);
assert_eq!(msg.cost, 0.42);
}
#[test]
fn test_apply_pricing_if_available_overrides_cost_when_pricing_exists() {
let mut litellm = HashMap::new();
litellm.insert(
"gpt-4o".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"codex",
"gpt-4o",
"provider",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.02);
}
#[test]
fn test_apply_pricing_if_available_applies_zed_hosted_markup() {
let mut litellm = HashMap::new();
litellm.insert(
"claude-sonnet-4-5".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"zed",
"claude-sonnet-4-5",
crate::sessions::zed::ZED_HOSTED_PROVIDER,
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert!((msg.cost - 0.022).abs() < 1e-12);
}
#[test]
fn test_apply_pricing_if_available_skips_zed_markup_for_non_zed_client() {
let mut litellm = HashMap::new();
litellm.insert(
"claude-sonnet-4-5".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"claudecode",
"claude-sonnet-4-5",
crate::sessions::zed::ZED_HOSTED_PROVIDER,
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert!((msg.cost - 0.020).abs() < 1e-12);
}
#[test]
fn test_apply_pricing_if_available_skips_zed_markup_for_byok_provider() {
let mut litellm = HashMap::new();
litellm.insert(
"claude-sonnet-4-5".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"zed",
"claude-sonnet-4-5",
"anthropic",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert!((msg.cost - 0.020).abs() < 1e-12);
}
#[test]
fn test_apply_pricing_if_available_uses_reasoning_for_gemini() {
let mut litellm = HashMap::new();
litellm.insert(
"gemini-2.5-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"gemini",
"gemini-2.5-pro",
"google",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 7,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.034);
}
#[test]
fn test_apply_pricing_if_available_uses_cache_read_pricing_for_gemini() {
let mut litellm = HashMap::new();
litellm.insert(
"gemini-2.5-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
cache_read_input_token_cost: Some(0.0001),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"gemini",
"gemini-2.5-pro",
"google",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 7,
cache_write: 0,
reasoning: 3,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.0267);
}
#[test]
fn test_apply_pricing_if_available_uses_market_rate_for_free_variant() {
let mut openrouter = HashMap::new();
openrouter.insert(
"z-ai/glm-4.7".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(HashMap::new(), openrouter);
let mut msg = UnifiedMessage::new(
"opencode",
"glm-4.7-free",
"modal",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.02);
}
#[test]
fn test_apply_pricing_if_available_prefers_provider_aware_match() {
let mut litellm = HashMap::new();
litellm.insert(
"xai/grok-code-fast-1-0825".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
litellm.insert(
"azure_ai/grok-code-fast-1".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"opencode",
"grok-code",
"azure",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.2);
}
#[test]
fn test_apply_pricing_if_available_uses_nested_reseller_exact_match() {
let mut litellm = HashMap::new();
litellm.insert(
"gpt-4".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
..Default::default()
},
);
litellm.insert(
"azure/openai/gpt-4".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"opencode",
"gpt-4",
"azure",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.2);
}
#[test]
fn test_apply_pricing_if_available_keeps_scoped_fireworks_cost_without_exact_pricing() {
let mut litellm = HashMap::new();
litellm.insert(
"fireworks_ai/accounts/fireworks/models/deepseek-r1-0528-distill-qwen3-8b".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.0000002),
output_cost_per_token: Some(0.0000002),
..Default::default()
},
);
let mut openrouter = HashMap::new();
openrouter.insert(
"deepseek/deepseek-v4-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.000001),
output_cost_per_token: Some(0.000002),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, openrouter);
let mut msg = UnifiedMessage::new(
"opencode",
"accounts/fireworks/models/deepseek-v4-pro",
"fireworks",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.123,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.123);
}
#[test]
fn test_apply_pricing_if_available_prefers_provider_specific_exact_match_over_plain_exact() {
let mut litellm = HashMap::new();
litellm.insert(
"gemini-2.5-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
cache_creation_input_token_cost: None,
..Default::default()
},
);
let mut openrouter = HashMap::new();
openrouter.insert(
"google/gemini-2.5-pro".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.001),
output_cost_per_token: Some(0.002),
cache_creation_input_token_cost: Some(0.01),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, openrouter);
let mut msg = UnifiedMessage::new(
"opencode",
"gemini-2.5-pro",
"google",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 3,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.05);
}
#[test]
fn test_apply_pricing_if_available_normalizes_openai_codex_provider() {
let mut litellm = HashMap::new();
litellm.insert(
"openai/gpt-5.2-preview".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
litellm.insert(
"google/gpt-5.2-preview-max".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.1),
output_cost_per_token: Some(0.2),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"openclaw",
"gpt-5.2",
"openai-codex",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.2);
}
#[test]
fn test_apply_pricing_if_available_prices_claude_code_gpt_5_3_codex() {
let pricing = pricing::PricingService::new(HashMap::new(), HashMap::new());
let mut msg = UnifiedMessage::new(
"claude",
"gpt-5.3-codex",
"openai",
"session-1",
1_776_000_000_000,
TokenBreakdown {
input: 1_000_000,
output: 100_000,
cache_read: 50_000,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
let expected = 1.75 + 1.4 + 0.00875;
assert!((msg.cost - expected).abs() < 1e-12);
}
#[test]
fn test_apply_pricing_if_available_prices_claude_code_minimax_model() {
let mut litellm = HashMap::new();
litellm.insert(
"minimax/minimax-m2.1".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let mut msg = UnifiedMessage::new(
"claude",
"MiniMax-M2.1",
"minimax",
"session-1",
1_776_000_000_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
assert_eq!(msg.cost, 0.2);
}
#[test]
fn test_apply_pricing_if_available_prices_kimi_k2p6_alias() {
let mut openrouter = HashMap::new();
openrouter.insert(
"moonshotai/kimi-k2.6".into(),
pricing::ModelPricing {
input_cost_per_token: Some(9.5e-7),
output_cost_per_token: Some(0.000004),
..Default::default()
},
);
let pricing = pricing::PricingService::new(HashMap::new(), openrouter);
let mut msg = UnifiedMessage::new(
"kimi",
"k2p6",
"kimi-for-coding",
"session-1",
1_776_000_000_000,
TokenBreakdown {
input: 1_000_000,
output: 250_000,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(&pricing));
let expected = 1_000_000.0 * 9.5e-7 + 250_000.0 * 0.000004;
assert!((msg.cost - expected).abs() < 1e-12);
assert!(msg.cost > 0.0);
}
#[test]
fn test_select_local_parse_pricing_prefers_fresh_service_for_new_models() {
let mut fresh_litellm = HashMap::new();
fresh_litellm.insert(
"gpt-5.4".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.000002),
output_cost_per_token: Some(0.00001),
..Default::default()
},
);
let fresh = Arc::new(pricing::PricingService::new(fresh_litellm, HashMap::new()));
let stale = pricing::PricingService::new(HashMap::new(), HashMap::new());
let selected = select_local_parse_pricing(Ok(Arc::clone(&fresh)), || Some(stale)).unwrap();
let mut msg = UnifiedMessage::new(
"opencode",
"gpt-5.4",
"openai",
"session-1",
1_733_011_200_000,
TokenBreakdown {
input: 10,
output: 5,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.0,
);
apply_pricing_if_available(&mut msg, Some(selected.as_ref()));
assert!(msg.cost > 0.0);
}
#[test]
fn test_select_local_parse_pricing_falls_back_to_stale_cache_on_fetch_error() {
let mut stale_litellm = HashMap::new();
stale_litellm.insert(
"gpt-5.2".into(),
pricing::ModelPricing {
input_cost_per_token: Some(0.00000175),
output_cost_per_token: Some(0.000014),
..Default::default()
},
);
let stale = pricing::PricingService::new(stale_litellm, HashMap::new());
let selected =
select_local_parse_pricing(Err("network failed".to_string()), || Some(stale)).unwrap();
assert!(selected.lookup_with_source("gpt-5.2", None).is_some());
}
#[test]
fn test_select_local_parse_pricing_does_not_evaluate_stale_fallback_on_fresh_success() {
let fresh = Arc::new(pricing::PricingService::new(HashMap::new(), HashMap::new()));
let mut stale_called = false;
let selected = select_local_parse_pricing(Ok(Arc::clone(&fresh)), || {
stale_called = true;
None
})
.unwrap();
assert!(Arc::ptr_eq(&selected, &fresh));
assert!(!stale_called);
}
#[test]
fn test_dedupe_latest_trae_messages_keeps_latest_timestamp_for_session() {
let messages = vec![
make_trae_message(
"session-stable",
1_700_000_002_000,
Some("trae:session-stable:1_700_000_002"),
0.2,
),
make_trae_message(
"session-stable",
1_700_000_003_000,
Some("trae:session-stable:1_700_000_003"),
0.3,
),
make_trae_message(
"session-other",
1_700_000_001_000,
Some("trae:session-other:1_700_000_001"),
0.1,
),
];
let deduped = dedupe_latest_trae_messages(messages);
assert_eq!(deduped.len(), 2);
let stable = deduped
.iter()
.find(|msg| msg.session_id == "session-stable")
.expect("session-stable should remain after dedupe");
assert_eq!(stable.timestamp, 1_700_000_003_000);
assert_eq!(stable.cost, 0.3);
assert_eq!(
stable.dedup_key.as_deref(),
Some("trae:session-stable:1_700_000_003")
);
}
#[test]
fn test_dedupe_latest_trae_messages_tiebreaks_by_dedup_key() {
let messages = vec![
make_trae_message(
"session-stable",
1_700_000_010_000,
Some("dedupe-key-a"),
0.2,
),
make_trae_message(
"session-stable",
1_700_000_010_000,
Some("dedupe-key-z"),
0.4,
),
make_trae_message(
"session-stable",
1_700_000_009_000,
Some("dedupe-key-m"),
0.1,
),
];
let deduped = dedupe_latest_trae_messages(messages);
assert_eq!(deduped.len(), 1);
assert_eq!(deduped[0].timestamp, 1_700_000_010_000);
assert_eq!(deduped[0].dedup_key.as_deref(), Some("dedupe-key-z"));
assert_eq!(deduped[0].cost, 0.4);
}
#[test]
fn test_parse_all_messages_with_pricing_keeps_gateway_message_under_synthetic_filter() {
let temp_dir = tempfile::TempDir::new().unwrap();
let message_dir = temp_dir
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
std::fs::write(
message_dir.join("msg_001.json"),
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"hf:deepseek-ai/DeepSeek-V3-0324","providerID":"unknown","cost":0,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let pricing = pricing::PricingService::new(HashMap::new(), HashMap::new());
let messages = parse_all_messages_with_pricing(
temp_dir.path().to_str().unwrap(),
&["synthetic".to_string()],
Some(&pricing),
);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].client, "opencode");
assert_eq!(messages[0].model_id, "deepseek-v3-0324");
assert_eq!(messages[0].provider_id, "synthetic");
}
#[test]
fn test_parse_local_clients_preserves_gateway_message_client_counts() {
let temp_dir = tempfile::TempDir::new().unwrap();
let message_dir = temp_dir
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
std::fs::write(
message_dir.join("msg_001.json"),
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["opencode".to_string(), "synthetic".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::OpenCode), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "opencode");
assert_eq!(parsed.messages[0].model_id, "deepseek-v3-0324");
assert_eq!(parsed.messages[0].provider_id, "fireworks_ai");
}
#[test]
fn test_parse_all_messages_fireworks_provider_kept_under_synthetic_only_filter() {
let temp_dir = tempfile::TempDir::new().unwrap();
let message_dir = temp_dir
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
std::fs::write(
message_dir.join("msg_001.json"),
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0.1,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let pricing = pricing::PricingService::new(HashMap::new(), HashMap::new());
let messages = parse_all_messages_with_pricing(
temp_dir.path().to_str().unwrap(),
&["synthetic".to_string()],
Some(&pricing),
);
assert_eq!(
messages.len(),
1,
"fireworks gateway message must not be dropped when filtering for synthetic"
);
assert_eq!(messages[0].client, "opencode");
assert_eq!(messages[0].model_id, "deepseek-v3-0324");
assert_eq!(messages[0].provider_id, "fireworks_ai");
}
#[test]
fn test_parse_local_clients_fireworks_provider_kept_under_synthetic_only_filter() {
let temp_dir = tempfile::TempDir::new().unwrap();
let message_dir = temp_dir
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
std::fs::write(
message_dir.join("msg_001.json"),
r#"{"id":"msg-1","sessionID":"session-1","role":"assistant","modelID":"accounts/fireworks/models/deepseek-v3-0324","providerID":"fireworks","cost":0.1,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["synthetic".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(
parsed.messages.len(),
1,
"fireworks gateway message must not be dropped when filtering for synthetic only"
);
assert_eq!(parsed.messages[0].client, "opencode");
assert_eq!(parsed.messages[0].model_id, "deepseek-v3-0324");
assert_eq!(parsed.messages[0].provider_id, "fireworks_ai");
}
#[test]
fn test_opencode_embedded_cost_survives_repricing_while_missing_cost_reprices() {
let temp_dir = tempfile::TempDir::new().unwrap();
let message_dir = temp_dir
.path()
.join(".local/share/opencode/storage/message/project-1");
std::fs::create_dir_all(&message_dir).unwrap();
std::fs::write(
message_dir.join("msg_reported.json"),
r#"{"id":"msg-reported","sessionID":"session-1","role":"assistant","modelID":"gpt-4o","providerID":"openai","cost":0.05,"tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011200000}}"#,
)
.unwrap();
std::fs::write(
message_dir.join("msg_missing.json"),
r#"{"id":"msg-missing","sessionID":"session-1","role":"assistant","modelID":"gpt-4o","providerID":"openai","tokens":{"input":10,"output":5,"reasoning":0,"cache":{"read":0,"write":0}},"time":{"created":1733011201000}}"#,
)
.unwrap();
let mut litellm = HashMap::new();
litellm.insert(
"openai/gpt-4o".to_string(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let messages = parse_all_messages_with_pricing_with_env_strategy(
temp_dir.path().to_str().unwrap(),
&["opencode".to_string()],
Some(&pricing),
false,
&scanner::ScannerSettings::default(),
);
let embedded = messages
.iter()
.find(|message| message.dedup_key.as_deref() == Some("msg-reported"))
.expect("embedded-cost message should parse");
let missing = messages
.iter()
.find(|message| message.dedup_key.as_deref() == Some("msg-missing"))
.expect("missing-cost message should parse");
assert_eq!(
embedded.cost, 0.05,
"OpenCode computes cost at request time; the embedded value must not be overwritten by LiteLLM repricing"
);
assert_eq!(embedded.cost_source, crate::CostSource::ProviderReported);
assert_eq!(missing.cost, 0.2);
assert_eq!(missing.cost_source, crate::CostSource::Estimated);
}
#[test]
fn test_gjc_explicit_zero_cost_is_preserved_while_absent_cost_reprices() {
let temp_dir = tempfile::TempDir::new().unwrap();
let session_dir = temp_dir.path().join(".gjc/agent/sessions/project-1");
std::fs::create_dir_all(&session_dir).unwrap();
std::fs::write(
session_dir.join("session.jsonl"),
r#"{"type":"session","id":"gjc_ses_cost","cwd":"/work/project-1"}
{"type":"message","id":"msg_zero","message":{"role":"assistant","model":"gpt-4o","provider":"openai","timestamp":1733011200000,"usage":{"input":10,"output":5,"cost":{"total":0.0}}}}
{"type":"message","id":"msg_absent","message":{"role":"assistant","model":"gpt-4o","provider":"openai","timestamp":1733011201000,"usage":{"input":10,"output":5}}}"#,
)
.unwrap();
let mut litellm = HashMap::new();
litellm.insert(
"openai/gpt-4o".to_string(),
pricing::ModelPricing {
input_cost_per_token: Some(0.01),
output_cost_per_token: Some(0.02),
..Default::default()
},
);
let pricing = pricing::PricingService::new(litellm, HashMap::new());
let messages = parse_all_messages_with_pricing_with_env_strategy(
temp_dir.path().to_str().unwrap(),
&["gjc".to_string()],
Some(&pricing),
false,
&scanner::ScannerSettings::default(),
);
let explicit_zero = messages
.iter()
.find(|message| message.dedup_key.as_deref() == Some("gjc_ses_cost:msg_zero"))
.expect("explicit-zero message should parse");
let absent = messages
.iter()
.find(|message| message.dedup_key.as_deref() == Some("gjc_ses_cost:msg_absent"))
.expect("absent-cost message should parse");
assert_eq!(explicit_zero.cost, 0.0);
assert_eq!(absent.cost, 0.2);
}
#[test]
fn test_gjc_idless_replay_dedup_stable_across_ordinal_shift() {
let temp_dir = tempfile::TempDir::new().unwrap();
let session_dir = temp_dir.path().join(".gjc/agent/sessions/project-1");
let child_dir = session_dir.join("session");
std::fs::create_dir_all(&child_dir).unwrap();
let assistant_line = r#"{"type":"message","message":{"role":"assistant","model":"gpt-4o","provider":"openai","timestamp":1733011200000,"usage":{"input":10,"output":5,"cost":{"total":0.03}}}}"#;
std::fs::write(
session_dir.join("session.jsonl"),
format!(
"{}\n{}\n",
r#"{"type":"session","id":"gjc_ses_replay_idless","cwd":"/work/project-1"}"#,
assistant_line
),
)
.unwrap();
std::fs::write(
child_dir.join("1-replay.jsonl"),
format!(
"{}\n{}\n{}\n",
r#"{"type":"session","id":"gjc_ses_replay_idless","cwd":"/work/project-1"}"#,
r#"{"type":"service_tier_change","tier":"pro"}"#,
assistant_line
),
)
.unwrap();
let messages = parse_all_messages_with_pricing_with_env_strategy(
temp_dir.path().to_str().unwrap(),
&["gjc".to_string()],
None,
false,
&scanner::ScannerSettings::default(),
);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].cost, 0.03);
}
#[test]
fn test_parse_local_clients_honors_scanner_settings_opencode_db_paths() {
let temp_dir = tempfile::TempDir::new().unwrap();
let outside_dir = temp_dir.path().join("elsewhere");
std::fs::create_dir_all(&outside_dir).unwrap();
let external_db = outside_dir.join("opencode.db");
let conn = rusqlite::Connection::open(&external_db).unwrap();
conn.execute_batch(
"PRAGMA journal_mode=WAL;
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"ext-msg-1",
"ext-session",
r#"{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"tokens": { "input": 42, "output": 7, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
"time": { "created": 1700000000000.0 }
}"#
],
)
.unwrap();
drop(conn);
let parsed_default = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["opencode".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed_default.counts.get(ClientId::OpenCode), 0);
assert!(parsed_default.messages.is_empty());
let parsed_with_settings = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["opencode".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
opencode_db_paths: vec![external_db.clone()],
..Default::default()
},
})
.unwrap();
assert_eq!(
parsed_with_settings.counts.get(ClientId::OpenCode),
1,
"scanner.opencodeDbPaths must reach the parse_local_clients path"
);
assert_eq!(parsed_with_settings.messages.len(), 1);
assert_eq!(parsed_with_settings.messages[0].client, "opencode");
assert_eq!(parsed_with_settings.messages[0].model_id, "claude-sonnet-4");
}
#[test]
fn test_parse_local_clients_honors_devin_cli_extra_scan_paths() {
let temp_dir = tempfile::TempDir::new().unwrap();
let external_dir = temp_dir.path().join("imports/devin/profile");
std::fs::create_dir_all(&external_dir).unwrap();
let external_db = external_dir.join("sessions.db");
let conn = rusqlite::Connection::open(&external_db).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
working_directory TEXT NOT NULL,
backend_type TEXT NOT NULL,
model TEXT NOT NULL,
title TEXT,
agent_mode TEXT NOT NULL,
created_at INTEGER NOT NULL,
last_activity_at INTEGER NOT NULL
);
CREATE TABLE message_nodes (
row_id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
node_id INTEGER NOT NULL,
parent_node_id INTEGER,
chat_message TEXT NOT NULL,
created_at INTEGER NOT NULL,
metadata TEXT
);",
)
.unwrap();
conn.execute(
"INSERT INTO sessions (id, working_directory, backend_type, model, agent_mode, created_at, last_activity_at) VALUES ('external-session', '/tmp/project', 'windsurf', 'gpt-5', 'accept-edits', 1, 1)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO message_nodes (session_id, node_id, chat_message, created_at) VALUES ('external-session', 1, ?1, 1700000000)",
[r#"{"role":"assistant","metadata":{"metrics":{"input_tokens":42,"output_tokens":7}}}"#],
)
.unwrap();
drop(conn);
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("devin-cli".to_string(), vec![external_dir]);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["devin-cli".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::DevinCli), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "devin-cli");
assert_eq!(parsed.messages[0].session_id, "external-session");
}
#[test]
fn test_parse_local_clients_devin_zero_cli_usage_does_not_suppress_desktop() {
let temp_dir = tempfile::TempDir::new().unwrap();
let db_path = temp_dir.path().join(".local/share/devin/cli/sessions.db");
std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
let conn = rusqlite::Connection::open(&db_path).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
working_directory TEXT NOT NULL,
backend_type TEXT NOT NULL,
model TEXT NOT NULL,
title TEXT,
agent_mode TEXT NOT NULL,
created_at INTEGER NOT NULL,
last_activity_at INTEGER NOT NULL
);
CREATE TABLE message_nodes (
row_id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
node_id INTEGER NOT NULL,
parent_node_id INTEGER,
chat_message TEXT NOT NULL,
created_at INTEGER NOT NULL,
metadata TEXT
);",
)
.unwrap();
conn.execute(
"INSERT INTO sessions (id, working_directory, backend_type, model, title, agent_mode, created_at, last_activity_at) VALUES ('cli-session', '/tmp/project', 'windsurf', 'gpt-5', 'Desktop task', 'accept-edits', 1, 1)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO message_nodes (session_id, node_id, chat_message, created_at) VALUES ('cli-session', 1, ?1, 1700000000)",
[r#"{"role":"assistant","metadata":{"metrics":{"input_tokens":0,"output_tokens":0}}}"#],
)
.unwrap();
drop(conn);
let desktop_dir = temp_dir
.path()
.join("Library/Application Support/Devin/User/acp-events");
std::fs::create_dir_all(&desktop_dir).unwrap();
std::fs::write(
desktop_dir.join("desktop-file.ndjson"),
concat!(
r#"{"notification":{"sessionUpdate":"session_info_update","title":"Desktop task"}}"#,
"\n",
r#"{"notification":{"sessionUpdate":"usage_update","_meta":{"cognition.ai/inputTokens":100,"cognition.ai/outputTokens":20,"cognition.ai/cachedReadTokens":10}}}"#,
"\n"
),
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["devin-cli".to_string(), "devin-desktop".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::DevinCli), 0);
assert_eq!(parsed.counts.get(ClientId::DevinDesktop), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "devin-desktop");
assert_eq!(parsed.messages[0].session_id, "cli-session");
assert_eq!(parsed.messages[0].model_id, "gpt-5");
assert_eq!(parsed.messages[0].input, 90);
assert_eq!(parsed.messages[0].cache_read, 10);
}
#[test]
fn test_parse_local_clients_devin_nonzero_cli_usage_dedups_desktop_row_but_keeps_raw_count() {
let temp_dir = tempfile::TempDir::new().unwrap();
let db_path = temp_dir.path().join(".local/share/devin/cli/sessions.db");
std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
let conn = rusqlite::Connection::open(&db_path).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
working_directory TEXT NOT NULL,
backend_type TEXT NOT NULL,
model TEXT NOT NULL,
title TEXT,
agent_mode TEXT NOT NULL,
created_at INTEGER NOT NULL,
last_activity_at INTEGER NOT NULL
);
CREATE TABLE message_nodes (
row_id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
node_id INTEGER NOT NULL,
parent_node_id INTEGER,
chat_message TEXT NOT NULL,
created_at INTEGER NOT NULL,
metadata TEXT
);",
)
.unwrap();
conn.execute(
"INSERT INTO sessions (id, working_directory, backend_type, model, title, agent_mode, created_at, last_activity_at) VALUES ('cli-session', '/tmp/project', 'windsurf', 'gpt-5', 'Desktop task', 'accept-edits', 1, 1)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO message_nodes (session_id, node_id, chat_message, created_at) VALUES ('cli-session', 1, ?1, 1700000000)",
[r#"{"role":"assistant","metadata":{"metrics":{"input_tokens":50,"output_tokens":25}}}"#],
)
.unwrap();
drop(conn);
let desktop_dir = temp_dir
.path()
.join("Library/Application Support/Devin/User/acp-events");
std::fs::create_dir_all(&desktop_dir).unwrap();
std::fs::write(
desktop_dir.join("desktop-file.ndjson"),
concat!(
r#"{"notification":{"sessionUpdate":"session_info_update","title":"Desktop task"}}"#,
"\n",
r#"{"notification":{"sessionUpdate":"usage_update","_meta":{"cognition.ai/inputTokens":100,"cognition.ai/outputTokens":20,"cognition.ai/cachedReadTokens":10}}}"#,
"\n"
),
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["devin-cli".to_string(), "devin-desktop".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "devin-cli");
assert_eq!(parsed.messages[0].session_id, "cli-session");
assert_eq!(parsed.counts.get(ClientId::DevinCli), 1);
assert!(parsed.counts.get(ClientId::DevinDesktop) > 0);
}
#[test]
fn test_parse_local_clients_desktop_uses_configured_cli_lookup_without_cli_usage() {
let temp_dir = tempfile::TempDir::new().unwrap();
let external_dir = temp_dir.path().join("imports/devin/profile");
std::fs::create_dir_all(&external_dir).unwrap();
let external_db = external_dir.join("sessions.db");
let conn = rusqlite::Connection::open(&external_db).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
title TEXT,
model TEXT,
working_directory TEXT
);",
)
.unwrap();
conn.execute(
"INSERT INTO sessions (id, title, model, working_directory) VALUES ('external-session', 'External desktop task', 'claude-sonnet-4', '/tmp/external-project')",
[],
)
.unwrap();
drop(conn);
let desktop_dir = temp_dir
.path()
.join("Library/Application Support/Devin/User/acp-events");
std::fs::create_dir_all(&desktop_dir).unwrap();
std::fs::write(
desktop_dir.join("desktop-file.ndjson"),
concat!(
r#"{"notification":{"sessionUpdate":"session_info_update","title":"External desktop task"}}"#,
"\n",
r#"{"notification":{"sessionUpdate":"usage_update","_meta":{"cognition.ai/inputTokens":100,"cognition.ai/outputTokens":20}}}"#,
"\n"
),
)
.unwrap();
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("devin-cli".to_string(), vec![external_dir]);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["devin-desktop".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::DevinCli), 0);
assert_eq!(parsed.counts.get(ClientId::DevinDesktop), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "devin-desktop");
assert_eq!(parsed.messages[0].session_id, "external-session");
assert_eq!(parsed.messages[0].model_id, "claude-sonnet-4");
assert_eq!(
parsed.messages[0].workspace_key.as_deref(),
Some("/tmp/external-project")
);
}
#[test]
fn test_devin_desktop_lookup_cache_separates_database_snapshots() {
let temp_dir = tempfile::TempDir::new().unwrap();
let db_path = temp_dir.path().join("sessions.db");
let conn = rusqlite::Connection::open(&db_path).unwrap();
conn.execute_batch(
"CREATE TABLE sessions (
id TEXT PRIMARY KEY,
title TEXT,
model TEXT,
working_directory TEXT
);
INSERT INTO sessions (id, title, model, working_directory)
VALUES ('cli-session', 'Snapshot task', 'gpt-5', '/tmp/project');",
)
.unwrap();
drop(conn);
let desktop_path = temp_dir.path().join("desktop-file.ndjson");
std::fs::write(
&desktop_path,
concat!(
r#"{"notification":{"sessionUpdate":"session_info_update","title":"Snapshot task"}}"#,
"\n",
r#"{"notification":{"sessionUpdate":"usage_update","_meta":{"cognition.ai/inputTokens":100,"cognition.ai/outputTokens":20}}}"#,
"\n"
),
)
.unwrap();
let first_fingerprint =
match message_cache::SourceFingerprint::check_devin_desktop_path_samples_only(
&desktop_path,
std::slice::from_ref(&db_path),
None,
)
.unwrap()
{
message_cache::FingerprintStatus::Changed(fingerprint) => fingerprint,
message_cache::FingerprintStatus::Unchanged => {
panic!("an uncached Desktop source must build a fingerprint")
}
};
let lookup_cache = std::sync::Mutex::new(HashMap::new());
let first_cell = super::devin_desktop_lookup_cell_for_snapshot(
&lookup_cache,
std::slice::from_ref(&db_path),
&first_fingerprint,
);
let first_lookup = first_cell.get_or_init(|| {
crate::sessions::devin::load_devin_desktop_session_lookup(std::slice::from_ref(
&db_path,
))
});
let first_messages = crate::sessions::devin::parse_devin_desktop_ndjson_with_lookup(
&desktop_path,
first_lookup,
);
assert_eq!(first_messages[0].model_id, "gpt-5");
let conn = rusqlite::Connection::open(&db_path).unwrap();
conn.execute(
"UPDATE sessions SET model = 'claude-sonnet-4' WHERE id = 'cli-session'",
[],
)
.unwrap();
drop(conn);
let second_fingerprint =
match message_cache::SourceFingerprint::check_devin_desktop_path_samples_only(
&desktop_path,
std::slice::from_ref(&db_path),
None,
)
.unwrap()
{
message_cache::FingerprintStatus::Changed(fingerprint) => fingerprint,
message_cache::FingerprintStatus::Unchanged => {
panic!("an uncached Desktop source must build a fingerprint")
}
};
assert_ne!(
first_fingerprint.related_files,
second_fingerprint.related_files
);
let second_cell = super::devin_desktop_lookup_cell_for_snapshot(
&lookup_cache,
std::slice::from_ref(&db_path),
&second_fingerprint,
);
assert!(
!Arc::ptr_eq(&first_cell, &second_cell),
"different database snapshots must not share a lookup cell"
);
let second_lookup = second_cell.get_or_init(|| {
crate::sessions::devin::load_devin_desktop_session_lookup(std::slice::from_ref(
&db_path,
))
});
let second_messages = crate::sessions::devin::parse_devin_desktop_ndjson_with_lookup(
&desktop_path,
second_lookup,
);
assert_eq!(second_messages[0].model_id, "claude-sonnet-4");
assert_eq!(lookup_cache.lock().unwrap().len(), 2);
}
#[test]
fn test_parse_local_clients_honors_scanner_extra_scan_paths_for_hermes_profile_db() {
let temp_dir = tempfile::TempDir::new().unwrap();
let profile_dir = temp_dir.path().join("external-hermes/director_planning");
std::fs::create_dir_all(&profile_dir).unwrap();
let profile_db = profile_dir.join("state.db");
let conn = create_hermes_sqlite_db(&profile_db);
insert_hermes_session(
&conn,
"hermes-extra-session",
"claude-sonnet-4",
2,
100,
25,
0.07,
);
drop(conn);
let parsed_default = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["hermes".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed_default.counts.get(ClientId::Hermes), 0);
assert!(parsed_default.messages.is_empty());
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("hermes".to_string(), vec![profile_dir]);
let parsed_with_settings = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["hermes".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed_with_settings.counts.get(ClientId::Hermes), 2);
assert_eq!(parsed_with_settings.messages.len(), 1);
assert_eq!(parsed_with_settings.messages[0].client, "hermes");
assert_eq!(
parsed_with_settings.messages[0].agent.as_deref(),
Some("Hermes Agent")
);
assert_eq!(
parsed_with_settings.messages[0].session_id,
"hermes-extra-session"
);
assert_eq!(parsed_with_settings.messages[0].model_id, "claude-sonnet-4");
assert_eq!(parsed_with_settings.messages[0].input, 100);
assert_eq!(parsed_with_settings.messages[0].output, 25);
}
#[test]
fn test_parse_local_clients_honors_scanner_extra_scan_paths_for_zed_threads_db() {
let temp_dir = tempfile::TempDir::new().unwrap();
let extra_threads_dir = temp_dir.path().join("custom-zed/threads");
std::fs::create_dir_all(&extra_threads_dir).unwrap();
let threads_db = extra_threads_dir.join("threads.db");
let conn = create_zed_sqlite_db(&threads_db);
insert_zed_thread(&conn, "zed-extra-thread", "claude-sonnet-4-5");
drop(conn);
let parsed_default = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["zed".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed_default.counts.get(ClientId::Zed), 0);
assert!(parsed_default.messages.is_empty());
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("zed".to_string(), vec![extra_threads_dir]);
let parsed_with_settings = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["zed".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed_with_settings.counts.get(ClientId::Zed), 1);
assert_eq!(parsed_with_settings.messages.len(), 1);
assert_eq!(parsed_with_settings.messages[0].client, "zed");
assert_eq!(
parsed_with_settings.messages[0].session_id,
"zed-extra-thread"
);
assert_eq!(
parsed_with_settings.messages[0].model_id,
"claude-sonnet-4-5"
);
assert_eq!(parsed_with_settings.messages[0].input, 42);
assert_eq!(parsed_with_settings.messages[0].output, 7);
}
#[test]
fn test_submit_default_graph_includes_antigravity_cache_rows() {
let temp_dir = tempfile::TempDir::new().unwrap();
let sessions_dir = temp_dir
.path()
.join(".config/tokmesh/antigravity-cache/sessions");
std::fs::create_dir_all(&sessions_dir).unwrap();
std::fs::write(
sessions_dir.join("ag-submit.jsonl"),
r#"{"type":"usage","sessionId":"ag-submit","modelId":"model_placeholder_m84","timestamp":1711200000000,"input":12,"output":4,"cacheRead":2,"cacheWrite":0,"reasoning":1,"responseId":"resp-ag"}
"#,
)
.unwrap();
let mut clients: Vec<String> = ClientId::iter()
.filter(|client| client.submit_default())
.map(|client| client.as_str().to_string())
.collect();
clients.push("synthetic".to_string());
let rt = tokio::runtime::Runtime::new().unwrap();
let graph = rt
.block_on(generate_graph_with_loaded_pricing(
ReportOptions {
home_dir: Some(temp_dir.path().to_string_lossy().to_string()),
use_env_roots: false,
clients: Some(clients),
since: None,
until: None,
year: None,
group_by: GroupBy::default(),
scanner_settings: scanner::ScannerSettings::default(),
},
None,
))
.unwrap();
assert_eq!(graph.summary.clients, vec!["antigravity"]);
assert_eq!(graph.summary.models, vec!["gemini-3-flash-preview"]);
assert_eq!(graph.summary.total_tokens, 19);
assert_eq!(graph.contributions.len(), 1);
assert_eq!(graph.contributions[0].clients[0].client, "antigravity");
assert_eq!(
graph.contributions[0].clients[0].model_id,
"gemini-3-flash-preview"
);
}
#[test]
fn test_parse_local_clients_dedups_zed_threads_across_default_and_extra_dbs() {
let temp_dir = tempfile::TempDir::new().unwrap();
let default_threads_dir = temp_dir.path().join(".local/share/zed/threads");
std::fs::create_dir_all(&default_threads_dir).unwrap();
let default_db = default_threads_dir.join("threads.db");
let conn = create_zed_sqlite_db(&default_db);
insert_zed_thread(&conn, "shared-zed-thread", "claude-sonnet-4-5");
drop(conn);
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("zed".to_string(), vec![default_threads_dir.clone()]);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["zed".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Zed), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].session_id, "shared-zed-thread");
}
#[test]
fn test_parse_local_clients_zed_extra_scan_paths_nonexistent_dir_is_silent() {
let temp_dir = tempfile::TempDir::new().unwrap();
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert(
"zed".to_string(),
vec![temp_dir.path().join("does/not/exist")],
);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["zed".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Zed), 0);
assert!(parsed.messages.is_empty());
}
#[test]
fn test_parse_local_clients_dedups_hermes_sessions_across_default_and_extra_dbs() {
let temp_dir = tempfile::TempDir::new().unwrap();
let default_dir = temp_dir.path().join(".hermes");
std::fs::create_dir_all(&default_dir).unwrap();
let default_db = default_dir.join("state.db");
let default_conn = create_hermes_sqlite_db(&default_db);
insert_hermes_session(
&default_conn,
"shared-hermes-session",
"claude-sonnet-4",
2,
100,
25,
0.07,
);
drop(default_conn);
let profile_dir = temp_dir.path().join(".hermes/profiles/director_planning");
std::fs::create_dir_all(&profile_dir).unwrap();
let profile_db = profile_dir.join("state.db");
let profile_conn = create_hermes_sqlite_db(&profile_db);
insert_hermes_session(
&profile_conn,
"shared-hermes-session",
"claude-sonnet-4",
9,
999,
999,
9.99,
);
drop(profile_conn);
let mut extra_scan_paths = std::collections::BTreeMap::new();
extra_scan_paths.insert("hermes".to_string(), vec![profile_db]);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["hermes".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
extra_scan_paths,
..Default::default()
},
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Hermes), 2);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].session_id, "shared-hermes-session");
assert_eq!(parsed.messages[0].input, 100);
assert_eq!(parsed.messages[0].output, 25);
}
#[test]
fn test_parse_local_clients_claude_filter_ignores_scanner_settings_opencode_db_paths() {
let temp_dir = tempfile::TempDir::new().unwrap();
let claude_dir = temp_dir.path().join(".claude/projects/myproject");
std::fs::create_dir_all(&claude_dir).unwrap();
std::fs::write(
claude_dir.join("conversation.jsonl"),
r#"{"type":"assistant","timestamp":"2024-12-01T10:00:00.000Z","requestId":"req_001","message":{"id":"msg_001","model":"claude-3-5-sonnet","usage":{"input_tokens":100,"output_tokens":50}}}
"#,
)
.unwrap();
let outside_dir = temp_dir.path().join("elsewhere");
std::fs::create_dir_all(&outside_dir).unwrap();
let external_db = outside_dir.join("opencode.db");
let conn = rusqlite::Connection::open(&external_db).unwrap();
conn.execute_batch(
"PRAGMA journal_mode=WAL;
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.unwrap();
conn.execute(
"INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
rusqlite::params![
"leaked-opencode",
"should-not-show-up",
r#"{
"role": "assistant",
"modelID": "claude-sonnet-4",
"providerID": "anthropic",
"tokens": { "input": 9999, "output": 9999, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
"time": { "created": 1700000000000.0 }
}"#
],
)
.unwrap();
drop(conn);
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["claude".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings {
opencode_db_paths: vec![external_db.clone()],
..Default::default()
},
})
.unwrap();
assert_eq!(
parsed.counts.get(ClientId::OpenCode),
0,
"OpenCode count must stay zero under a Claude-only filter even \
when scanner.opencodeDbPaths is set"
);
assert_eq!(
parsed.counts.get(ClientId::Claude),
1,
"Claude message must still be counted"
);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "claude");
assert!(
parsed.messages.iter().all(|m| m.client != "opencode"),
"no OpenCode messages may leak into a Claude-only result, got {:?}",
parsed.messages
);
}
#[test]
fn test_parse_local_clients_claude_transcripts_count_only_usage_metadata() {
let temp_dir = tempfile::TempDir::new().unwrap();
let transcripts_dir = temp_dir.path().join(".claude/transcripts");
std::fs::create_dir_all(&transcripts_dir).unwrap();
std::fs::write(
transcripts_dir.join("ses_123456789012345678901234567.jsonl"),
r#"{"type":"user","timestamp":"2026-04-01T10:00:00.000Z","message":{"content":"Wrapped prompt"}}
{"type":"assistant","timestamp":"2026-04-01T10:00:01.000Z","requestId":"req_wrapper","message":{"id":"msg_wrapper","model":"claude-sonnet-4","usage":{"input_tokens":123,"output_tokens":45,"cache_read_input_tokens":67,"cache_creation_input_tokens":8}}}
"#,
)
.unwrap();
std::fs::write(
transcripts_dir.join("ses_765432109876543210987654321.jsonl"),
r#"{"type":"user","timestamp":"2026-04-01T10:00:00.000Z","message":{"content":"Wrapped prompt"}}
{"type":"tool_use","timestamp":"2026-04-01T10:00:01.000Z","message":{"content":"Run tool"}}
{"type":"tool_result","timestamp":"2026-04-01T10:00:02.000Z","message":{"content":"Tool result"}}
"#,
)
.unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["claude".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Claude), 1);
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].client, "claude");
assert_eq!(
parsed.messages[0].session_id,
"ses_123456789012345678901234567"
);
assert_eq!(parsed.messages[0].model_id, "claude-sonnet-4");
assert_eq!(parsed.messages[0].input, 123);
assert_eq!(parsed.messages[0].output, 45);
assert_eq!(parsed.messages[0].cache_read, 67);
assert_eq!(parsed.messages[0].cache_write, 8);
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_refreshes_cc_mirror_provider_when_variant_metadata_changes() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let variant_dir = source_home.path().join(".cc-mirror/kimi-code");
let config_dir = source_home.path().join("mirror-configs/kimi-code");
let project_dir = config_dir.join("projects/project-one");
std::fs::create_dir_all(&project_dir).unwrap();
std::fs::create_dir_all(&variant_dir).unwrap();
let variant_path = variant_dir.join("variant.json");
std::fs::write(
&variant_path,
format!(
r#"{{"name":"kimi-code","provider":"kimi","configDir":"{}"}}"#,
config_dir.display()
),
)
.unwrap();
let session_path = project_dir.join("session.jsonl");
std::fs::write(
&session_path,
r#"{"type":"assistant","timestamp":"2024-12-01T10:00:00.000Z","requestId":"req_001","message":{"id":"msg_001","model":"claude-3-5-sonnet","usage":{"input_tokens":100,"output_tokens":50}}}
"#,
)
.unwrap();
let first_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["claude".to_string()],
None,
);
assert_eq!(first_messages.len(), 1);
assert_eq!(first_messages[0].client, "cc-mirror/kimi-code");
assert_eq!(first_messages[0].provider_id, "kimi");
std::fs::write(
&variant_path,
format!(
r#"{{"name":"kimi-code","provider":"minimax","configDir":"{}"}}"#,
config_dir.display()
),
)
.unwrap();
let refreshed_messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["claude".to_string()],
None,
);
assert_eq!(refreshed_messages.len(), 1);
assert_eq!(refreshed_messages[0].client, "cc-mirror/kimi-code");
assert_eq!(refreshed_messages[0].provider_id, "minimax");
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
#[serial_test::serial]
fn test_parse_all_messages_keeps_normal_claude_when_cc_mirror_points_at_claude_config() {
let cache_home = tempfile::TempDir::new().unwrap();
let source_home = tempfile::TempDir::new().unwrap();
let original_home = std::env::var("HOME").ok();
std::env::set_var("HOME", cache_home.path());
{
let claude_dir = source_home.path().join(".claude");
let project_dir = claude_dir.join("projects/project-one");
std::fs::create_dir_all(&project_dir).unwrap();
let session_path = project_dir.join("session.jsonl");
std::fs::write(
&session_path,
r#"{"type":"assistant","timestamp":"2024-12-01T10:00:00.000Z","requestId":"req_001","message":{"id":"msg_001","model":"claude-3-5-sonnet","usage":{"input_tokens":100,"output_tokens":50}}}
"#,
)
.unwrap();
let variant_dir = source_home.path().join(".cc-mirror/plain-mirror");
std::fs::create_dir_all(&variant_dir).unwrap();
std::fs::write(
variant_dir.join("variant.json"),
format!(
r#"{{"name":"plain-mirror","provider":"mirror","configDir":"{}"}}"#,
claude_dir.display()
),
)
.unwrap();
let messages = parse_all_messages_with_pricing(
source_home.path().to_str().unwrap(),
&["claude".to_string()],
None,
);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].client, "claude");
}
match original_home {
Some(home) => std::env::set_var("HOME", home),
None => std::env::remove_var("HOME"),
}
}
#[test]
fn test_parse_local_clients_amp_partial_ledger_recovers_message_fallback_day() {
use chrono::TimeZone;
let temp_dir = tempfile::TempDir::new().unwrap();
let amp_dir = temp_dir.path().join(".local/share/amp/threads");
std::fs::create_dir_all(&_dir).unwrap();
let thread_created = chrono::DateTime::parse_from_rfc3339("2026-04-04T12:00:00Z")
.unwrap()
.timestamp_millis();
let ledger_timestamp = chrono::DateTime::parse_from_rfc3339("2026-04-08T12:00:00Z")
.unwrap()
.timestamp_millis();
let thread = format!(
r#"{{
"id": "thread-amp-gap",
"created": {thread_created},
"usageLedger": {{
"events": [
{{
"timestamp": "2026-04-08T12:00:00Z",
"model": "claude-sonnet-4-0",
"credits": 0.75,
"tokens": {{ "input": 100, "output": 20 }}
}}
]
}},
"messages": [
{{
"role": "assistant",
"messageId": 1,
"usage": {{
"model": "claude-sonnet-4-0",
"inputTokens": 100,
"outputTokens": 20,
"credits": 0.75
}}
}},
{{
"role": "assistant",
"messageId": 2,
"usage": {{
"model": "claude-sonnet-4-0",
"inputTokens": 50,
"outputTokens": 10,
"credits": 0.40
}}
}}
]
}}"#
);
std::fs::write(amp_dir.join("T-thread-amp-gap.json"), thread).unwrap();
let parsed = parse_local_clients(LocalParseOptions {
home_dir: Some(temp_dir.path().to_str().unwrap().to_string()),
use_env_roots: false,
clients: Some(vec!["amp".to_string()]),
since: None,
until: None,
year: None,
scanner_settings: scanner::ScannerSettings::default(),
})
.unwrap();
assert_eq!(parsed.counts.get(ClientId::Amp), 2);
assert_eq!(parsed.messages.len(), 2);
let dates: HashSet<String> = parsed.messages.iter().map(|msg| msg.date.clone()).collect();
let local_date = |timestamp_ms: i64| {
chrono::Local
.timestamp_millis_opt(timestamp_ms)
.single()
.unwrap()
.format("%Y-%m-%d")
.to_string()
};
assert!(dates.contains(&local_date(thread_created + 2000)));
assert!(dates.contains(&local_date(ledger_timestamp)));
}
#[test]
fn test_retain_for_requested_clients_gjc_superset_of_9router() {
let gjc_requested: HashSet<&str> = HashSet::from(["gjc"]);
assert!(retain_for_requested_clients(
"9router",
"deepseek-ai/deepseek-v4-flash",
"nvidia",
&gjc_requested
));
let ninerouter_requested: HashSet<&str> = HashSet::from(["9router"]);
assert!(retain_for_requested_clients(
"9router",
"deepseek-ai/deepseek-v4-flash",
"nvidia",
&ninerouter_requested
));
assert!(!retain_for_requested_clients(
"gjc",
"claude-sonnet-4",
"anthropic",
&ninerouter_requested
));
assert!(!retain_for_requested_clients(
"claude",
"gpt-4o",
"openai",
&gjc_requested
));
}
#[test]
fn test_filter_messages_preserves_pi_9router_when_no_duplicate() {
let messages = vec![
UnifiedMessage::new(
"pi",
"deepseek_v4_flash_free",
"9router",
"session-1",
1783412353188,
TokenBreakdown::default(),
0.0,
),
UnifiedMessage::new(
"9router",
"deepseek-ai/deepseek-v4-flash",
"nvidia",
"session-2",
1783412353188,
TokenBreakdown {
input: 100,
output: 50,
cache_read: 0,
cache_write: 0,
reasoning: 0,
},
0.05,
),
];
let filtered = filter_messages_for_report(messages, &ReportOptions::default());
assert_eq!(filtered.len(), 2);
}
}