admin/CC-Switch
1
0
Fork 0
mirror of https://github.com/farion1231/cc-switch.git synced 2026-07-24 12:44:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(usage): account for cache-write tokens across schema versions

Browse Source 
Parse cache_write_tokens from OpenAI usage details and preserve cache creation data across Chat, Responses, and Anthropic conversion paths.

Add explicit input-token semantics to request logs and rollups so legacy rows subtract cache reads only while new total-inclusive rows subtract both cache reads and writes. Migrate v12 databases, normalize rollups to fresh input, and cover historical backfill behavior with regression tests.
This commit is contained in:
Jason
2026-07-11 12:27:39 +08:00
parent 06039540ff
commit f991726ff0
13 changed files with 443 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/database/dao/usage_rollup.rs
+42 -4
View File
@@ -4,6 +4,7 @@
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use crate::services::sql_helpers::{fresh_input_sql, INPUT_TOKEN_SEMANTICS_FRESH};
use crate::services::usage_stats::effective_usage_log_filter;
use chrono::{Duration, Local, TimeZone};
@@ -115,6 +116,8 @@ impl Database {
fn do_rollup_and_prune(conn: &rusqlite::Connection, cutoff: i64) -> Result<u64, AppError> {
// Aggregate old logs, merging with any pre-existing rollup rows via LEFT JOIN.
let effective_filter = effective_usage_log_filter("l");
let fresh_detail_input = fresh_input_sql("l");
let fresh_old_input = fresh_input_sql("old");
// request_model 维度保留路由接管的「客户端别名 → 真实模型」映射，
// pricing_model 维度保留写入时的计价基准（request 计价模式下与 model 分叉）；
// 明细行的这两列可能为 NULL（历史/手工数据），归一为 ''。
@@ -124,15 +127,16 @@ impl Database {
request_count, success_count,
input_tokens, output_tokens,
cache_read_tokens, cache_creation_tokens,
total_cost_usd, avg_latency_ms)
input_token_semantics, total_cost_usd, avg_latency_ms)
SELECT
d, a, p, m, rm, pm,
COALESCE(old.request_count, 0) + new_req,
COALESCE(old.success_count, 0) + new_succ,
COALESCE(old.input_tokens, 0) + new_in,
COALESCE({fresh_old_input}, 0) + new_in,
COALESCE(old.output_tokens, 0) + new_out,
COALESCE(old.cache_read_tokens, 0) + new_cr,
COALESCE(old.cache_creation_tokens, 0) + new_cc,
{INPUT_TOKEN_SEMANTICS_FRESH},
CAST(COALESCE(CAST(old.total_cost_usd AS REAL), 0) + new_cost AS TEXT),
CASE WHEN COALESCE(old.request_count, 0) + new_req > 0
THEN (COALESCE(old.avg_latency_ms, 0) * COALESCE(old.request_count, 0)
@@ -147,7 +151,7 @@ impl Database {
COALESCE(l.pricing_model, '') as pm,
COUNT(*) as new_req,
SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END) as new_succ,
COALESCE(SUM(l.input_tokens), 0) as new_in,
COALESCE(SUM({fresh_detail_input}), 0) as new_in,
COALESCE(SUM(l.output_tokens), 0) as new_out,
COALESCE(SUM(l.cache_read_tokens), 0) as new_cr,
COALESCE(SUM(l.cache_creation_tokens), 0) as new_cc,
@@ -327,7 +331,7 @@ mod tests {
let (provider_id, request_count, input_tokens, output_tokens, cache_read_tokens) = &rows[0];
assert_eq!(provider_id, "openai");
assert_eq!(*request_count, 1);
assert_eq!(*input_tokens, 100);
assert_eq!(*input_tokens, 90, "rollup stores normalized fresh input");
assert_eq!(*output_tokens, 20);
assert_eq!(*cache_read_tokens, 10);
@@ -340,6 +344,40 @@ mod tests {
Ok(())
}
#[test]
fn test_rollup_normalizes_total_cache_semantics_to_fresh() -> Result<(), AppError> {
let db = Database::memory()?;
let old_ts = chrono::Utc::now().timestamp() - 40 * 86400;
{
let conn = crate::database::lock_conn!(db.conn);
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_token_semantics, total_cost_usd,
latency_ms, status_code, created_at
) VALUES ('total-semantics-rollup', 'p1', 'codex', 'gpt-5.5',
100, 5, 10, 20, 1, '0.10', 100, 200, ?1)",
[old_ts],
)?;
}
assert_eq!(db.rollup_and_prune(30)?, 1);
let conn = crate::database::lock_conn!(db.conn);
let row: (i64, i64, i64, i64) = conn.query_row(
"SELECT input_tokens, cache_read_tokens, cache_creation_tokens,
input_token_semantics
FROM usage_daily_rollups WHERE model = 'gpt-5.5'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)?;
assert_eq!(row, (70, 10, 20, 2));
Ok(())
}
#[test]
fn test_rollup_preserves_request_model_dimension() -> Result<(), AppError> {
let db = Database::memory()?;
src-tauri/src/database/mod.rs
+1 -1
View File
@@ -50,7 +50,7 @@ use std::sync::Mutex;
/// 当前 Schema 版本号
/// 每次修改表结构时递增，并在 schema.rs 中添加相应的迁移逻辑
pub(crate) const SCHEMA_VERSION: i32 = 12;
pub(crate) const SCHEMA_VERSION: i32 = 13;
/// 安全地序列化 JSON，避免 unwrap panic
pub(crate) fn to_json_string<T: Serialize>(value: &T) -> Result<String, AppError> {
src-tauri/src/database/schema.rs
+74
View File
@@ -189,6 +189,7 @@ impl Database {
pricing_model TEXT,
input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0, cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
input_token_semantics INTEGER NOT NULL DEFAULT 0,
input_cost_usd TEXT NOT NULL DEFAULT '0', output_cost_usd TEXT NOT NULL DEFAULT '0',
cache_read_cost_usd TEXT NOT NULL DEFAULT '0', cache_creation_cost_usd TEXT NOT NULL DEFAULT '0',
total_cost_usd TEXT NOT NULL DEFAULT '0', latency_ms INTEGER NOT NULL, first_token_ms INTEGER,
@@ -275,6 +276,7 @@ impl Database {
output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
input_token_semantics INTEGER NOT NULL DEFAULT 0,
total_cost_usd TEXT NOT NULL DEFAULT '0',
avg_latency_ms INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (date, app_type, provider_id, model, request_model, pricing_model)
@@ -478,6 +480,11 @@ impl Database {
Self::migrate_v11_to_v12(conn)?;
Self::set_user_version(conn, 12)?;
}
12 => {
log::info!("迁移数据库从 v12 到 v13（记录输入 token 缓存语义）");
Self::migrate_v12_to_v13(conn)?;
Self::set_user_version(conn, 13)?;
}
_ => {
return Err(AppError::Database(format!(
"未知的数据库版本 {version}，无法迁移到 {SCHEMA_VERSION}"
@@ -650,6 +657,7 @@ impl Database {
request_model TEXT,
input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0, cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
input_token_semantics INTEGER NOT NULL DEFAULT 0,
input_cost_usd TEXT NOT NULL DEFAULT '0', output_cost_usd TEXT NOT NULL DEFAULT '0',
cache_read_cost_usd TEXT NOT NULL DEFAULT '0', cache_creation_cost_usd TEXT NOT NULL DEFAULT '0',
total_cost_usd TEXT NOT NULL DEFAULT '0', latency_ms INTEGER NOT NULL, first_token_ms INTEGER,
@@ -1322,6 +1330,30 @@ impl Database {
Ok(())
}
/// v12 -> v13：记录 input_tokens 是否包含缓存写入。
///
/// 默认 0 表示旧版/未知语义；旧 Codex 行只包含 cache read，不包含
/// cache creation。新代理行会显式写入 1(total-inclusive) 或 2(fresh)。
fn migrate_v12_to_v13(conn: &Connection) -> Result<(), AppError> {
if Self::table_exists(conn, "proxy_request_logs")? {
Self::add_column_if_missing(
conn,
"proxy_request_logs",
"input_token_semantics",
"INTEGER NOT NULL DEFAULT 0",
)?;
}
if Self::table_exists(conn, "usage_daily_rollups")? {
Self::add_column_if_missing(
conn,
"usage_daily_rollups",
"input_token_semantics",
"INTEGER NOT NULL DEFAULT 0",
)?;
}
Ok(())
}
/// 插入默认模型定价数据
/// 格式: (model_id, display_name, input, output, cache_read, cache_creation)
/// 注意: model_id 使用短横线格式（如 claude-haiku-4-5），与 API 返回的模型名称标准化后一致
@@ -2661,3 +2693,45 @@ impl Database {
Ok(true)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn migrate_v12_to_v13_adds_input_token_semantics_columns() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE proxy_request_logs (request_id TEXT PRIMARY KEY)",
[],
)?;
conn.execute(
"CREATE TABLE usage_daily_rollups (date TEXT PRIMARY KEY)",
[],
)?;
Database::set_user_version(&conn, 12)?;
Database::apply_schema_migrations_on_conn(&conn)?;
assert_eq!(Database::get_user_version(&conn)?, 13);
assert!(Database::has_column(
&conn,
"proxy_request_logs",
"input_token_semantics"
)?);
assert!(Database::has_column(
&conn,
"usage_daily_rollups",
"input_token_semantics"
)?);
let log_default: i64 = conn.query_row(
"SELECT dflt_value = '0' FROM pragma_table_info('proxy_request_logs')
WHERE name = 'input_token_semantics'",
[],
|row| row.get(0),
)?;
assert_eq!(log_default, 1);
Ok(())
}
}
src-tauri/src/proxy/providers/streaming.rs
+17 -3
View File
@@ -80,6 +80,8 @@ struct Usage {
struct PromptTokensDetails {
#[serde(default)]
cached_tokens: u32,
#[serde(default)]
cache_write_tokens: u32,
}
#[derive(Debug, Clone)]
@@ -103,7 +105,7 @@ fn build_anthropic_usage_json(usage: &Usage) -> Value {
// OpenAI prompt_tokens 含缓存，Anthropic input_tokens 不含，需减去 cache_read 与 cache_creation
// （三桶互斥，恒等 input + cache_read + cache_creation == prompt_tokens）。
let cached = extract_cache_read_tokens(usage).unwrap_or(0);
let cache_creation = usage.cache_creation_input_tokens.unwrap_or(0);
let cache_creation = extract_cache_write_tokens(usage).unwrap_or(0);
let input_tokens = usage
.prompt_tokens
.saturating_sub(cached)
@@ -233,7 +235,7 @@ pub fn create_anthropic_sse_stream<E: std::error::Error + Send + 'static>(
if let Some(u) = &chunk.usage {
let cached = extract_cache_read_tokens(u).unwrap_or(0);
let cache_creation =
u.cache_creation_input_tokens.unwrap_or(0);
extract_cache_write_tokens(u).unwrap_or(0);
let input = u
.prompt_tokens
.saturating_sub(cached)
@@ -683,6 +685,18 @@ fn extract_cache_read_tokens(usage: &Usage) -> Option<u32> {
.filter(|&v| v > 0)
}
/// Extract cache-write tokens from direct compatibility fields or OpenAI details.
fn extract_cache_write_tokens(usage: &Usage) -> Option<u32> {
if let Some(value) = usage.cache_creation_input_tokens {
return Some(value);
}
usage
.prompt_tokens_details
.as_ref()
.map(|details| details.cache_write_tokens)
.filter(|value| *value > 0)
}
/// 映射停止原因
fn map_stop_reason(finish_reason: Option<&str>) -> Option<String> {
finish_reason.map(|r| {
@@ -1061,7 +1075,7 @@ mod tests {
let input = concat!(
"data: {\"id\":\"chatcmpl_cc\",\"model\":\"glm-5.1\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"tool-1\",\"type\":\"function\",\"function\":{\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"pwd\\\"}\"}}]}}]}\n\n",
"data: {\"id\":\"chatcmpl_cc\",\"model\":\"glm-5.1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n",
"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1000,\"completion_tokens\":50,\"prompt_tokens_details\":{\"cached_tokens\":600},\"cache_creation_input_tokens\":300}}\n\n",
"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1000,\"completion_tokens\":50,\"prompt_tokens_details\":{\"cached_tokens\":600,\"cache_write_tokens\":300}}}\n\n",
"data: [DONE]\n\n"
);
src-tauri/src/proxy/providers/transform.rs
+7 -1
View File
@@ -665,7 +665,7 @@ pub fn openai_to_anthropic(body: Value) -> Result<Value, ProxyError> {
// 不再扣减），若不减则缓存会被计入 input 与各 cache 桶两次。三桶互斥，恒等：
// input + cache_read + cache_creation == prompt_tokens（inclusive 上游）。
// 与流式 build_anthropic_usage_json (#2774) 及 transform_gemini 的 saturating_sub 对称。
// 最终 cache_read：直传字段优先于 nested；cache_creation 仅来自直传字段（OpenAI 无此概念）。
// 最终 cache_read/cache_creation：直传字段优先于 OpenAI nested details。
let cached = usage
.get("cache_read_input_tokens")
.and_then(|v| v.as_u64())
@@ -678,6 +678,12 @@ pub fn openai_to_anthropic(body: Value) -> Result<Value, ProxyError> {
let cache_creation = usage
.get("cache_creation_input_tokens")
.and_then(|v| v.as_u64())
.or_else(|| {
usage
.pointer("/prompt_tokens_details/cache_write_tokens")
.or_else(|| usage.pointer("/input_tokens_details/cache_write_tokens"))
.and_then(|v| v.as_u64())
})
.unwrap_or(0);
let input_tokens = usage
.get("prompt_tokens")
src-tauri/src/proxy/providers/transform_codex_anthropic.rs
+24 -23
View File
@@ -125,19 +125,14 @@ pub(crate) fn map_anthropic_stop_reason_to_status(
/// Builds Responses usage from Anthropic usage.
///
/// Anthropic's `input_tokens` is the "cache-excluded" fresh input; OpenAI/Responses'
/// `input_tokens` includes cache hits. To keep downstream metering correct, this
/// adds them (symmetric to the subtraction done for the Claude side in
/// `transform_responses`):
/// input_tokens = input + cache_read
/// Anthropic's `input_tokens` is the cache-excluded fresh input. OpenAI Responses
/// reports total input and exposes cache reads/writes as subsets:
/// input_tokens = fresh + cache_read + cache_creation
/// input_tokens_details.cached_tokens = cache_read
/// input_tokens_details.cache_write_tokens = cache_creation
///
/// Note: **do not** fold `cache_creation` into `input_tokens`. The Codex billing
/// calculator (usage/calculator.rs) only subtracts `cache_read` for codex
/// (`billable = input - cache_read`), and separately lists cache-creation cost via
/// `cache_creation_input_tokens`; if creation were also added into
/// `input_tokens`, it would be double-charged at both the input price and the
/// cache-creation price.
/// The internal billing parser subtracts both subsets before charging the normal
/// input rate, then prices cache reads and writes separately.
pub(crate) fn build_responses_usage_from_anthropic(usage: Option<&Value>) -> Value {
let u = match usage {
Some(v) if v.is_object() => v,
@@ -166,10 +161,10 @@ pub(crate) fn build_responses_usage_from_anthropic(usage: Option<&Value>) -> Val
.and_then(|v| v.as_u64())
.unwrap_or(0);
let input_tokens = fresh_input.saturating_add(cache_read);
let total_tokens = input_tokens
.saturating_add(cache_creation)
.saturating_add(output);
let input_tokens = fresh_input
.saturating_add(cache_read)
.saturating_add(cache_creation);
let total_tokens = input_tokens.saturating_add(output);
let mut result = json!({
"input_tokens": input_tokens,
@@ -177,10 +172,14 @@ pub(crate) fn build_responses_usage_from_anthropic(usage: Option<&Value>) -> Val
"total_tokens": total_tokens,
"output_tokens_details": { "reasoning_tokens": reasoning }
});
if cache_read > 0 {
result["input_tokens_details"] = json!({ "cached_tokens": cache_read });
if cache_read > 0 || cache_creation > 0 {
result["input_tokens_details"] = json!({
"cached_tokens": cache_read,
"cache_write_tokens": cache_creation
});
}
// Explicitly pass through cache_creation so the downstream usage parser (from_codex_response) attributes billing correctly.
// Keep the legacy top-level alias for one compatibility window. New code reads
// the official nested cache_write_tokens field first.
if cache_creation > 0 {
result["cache_creation_input_tokens"] = json!(cache_creation);
}
@@ -2078,18 +2077,20 @@ mod tests {
}
});
let result = anthropic_response_to_responses(input).unwrap();
// input_tokens = fresh + cache_read = 20 + 60 = 80 (excluding cache_creation).
// The Codex billing calculator only subtracts cache_read from input (→ billable=fresh=20),
// and separately lists cache-creation cost via cache_creation_input_tokens; folding creation into input would double-charge.
assert_eq!(result["usage"]["input_tokens"], 80);
// Responses input_tokens is the inclusive total: fresh + read + write.
assert_eq!(result["usage"]["input_tokens"], 100);
assert_eq!(result["usage"]["output_tokens"], 5);
assert_eq!(
result["usage"]["output_tokens_details"]["reasoning_tokens"],
3
);
// total still includes everything: 80 + cache_creation 20 + output 5 = 105
// total includes input total + output exactly once.
assert_eq!(result["usage"]["total_tokens"], 105);
assert_eq!(result["usage"]["input_tokens_details"]["cached_tokens"], 60);
assert_eq!(
result["usage"]["input_tokens_details"]["cache_write_tokens"],
20
);
// cache_creation is passed through explicitly for downstream billing attribution (counted only once)
assert_eq!(result["usage"]["cache_creation_input_tokens"], 20);
}
src-tauri/src/proxy/providers/transform_codex_chat.rs
+24 -6
View File
@@ -1629,12 +1629,26 @@ pub(crate) fn chat_usage_to_responses_usage(usage: Option<&Value>) -> Value {
"total_tokens": total_tokens
});
if let Some(cached) = usage
let cached = usage
.pointer("/prompt_tokens_details/cached_tokens")
.or_else(|| usage.pointer("/input_tokens_details/cached_tokens"))
.and_then(|v| v.as_u64())
{
result["input_tokens_details"] = json!({ "cached_tokens": cached });
.unwrap_or(0);
let cache_write = usage
.pointer("/prompt_tokens_details/cache_write_tokens")
.or_else(|| usage.pointer("/input_tokens_details/cache_write_tokens"))
.and_then(|v| v.as_u64())
.or_else(|| {
usage
.get("cache_creation_input_tokens")
.and_then(|v| v.as_u64())
})
.unwrap_or(0);
if cached > 0 || cache_write > 0 {
result["input_tokens_details"] = json!({
"cached_tokens": cached,
"cache_write_tokens": cache_write
});
}
if let Some(details) = usage
@@ -1653,8 +1667,8 @@ pub(crate) fn chat_usage_to_responses_usage(usage: Option<&Value>) -> Value {
if let Some(cache_read) = usage.get("cache_read_input_tokens") {
result["cache_read_input_tokens"] = cache_read.clone();
}
if let Some(cache_creation) = usage.get("cache_creation_input_tokens") {
result["cache_creation_input_tokens"] = cache_creation.clone();
if cache_write > 0 {
result["cache_creation_input_tokens"] = json!(cache_write);
}
result
@@ -2735,7 +2749,7 @@ mod tests {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15,
"prompt_tokens_details": {"cached_tokens": 3}
"prompt_tokens_details": {"cached_tokens": 3, "cache_write_tokens": 2}
}
});
@@ -2759,6 +2773,10 @@ mod tests {
assert_eq!(result["usage"]["input_tokens"], 10);
assert_eq!(result["usage"]["output_tokens"], 5);
assert_eq!(result["usage"]["input_tokens_details"]["cached_tokens"], 3);
assert_eq!(
result["usage"]["input_tokens_details"]["cache_write_tokens"],
2
);
}
#[test]
src-tauri/src/proxy/providers/transform_responses.rs
+31 -2
View File
@@ -259,10 +259,11 @@ pub(crate) fn map_responses_stop_reason(
/// 1. input_tokens: Anthropic `input_tokens` → OpenAI `prompt_tokens` → default 0
/// 2. output_tokens: Anthropic `output_tokens` → OpenAI `completion_tokens` → default 0
/// 3. cache_read_input_tokens: Direct field → nested input_tokens_details.cached_tokens → prompt_tokens_details.cached_tokens
/// 4. cache_creation_input_tokens: Direct field only
/// 4. cache_creation_input_tokens: Direct field → nested
/// input_tokens_details.cache_write_tokens → prompt_tokens_details.cache_write_tokens
///
/// **Cache Token Priority Order**:
/// 1. OpenAI nested details (`input_tokens_details.cached_tokens`, `prompt_tokens_details.cached_tokens`) as initial value
/// 1. OpenAI nested details (`cached_tokens`, `cache_write_tokens`) as initial values
/// 2. Direct Anthropic-style fields (`cache_read_input_tokens`, `cache_creation_input_tokens`) override if present
///
/// **Logging**:
@@ -345,6 +346,19 @@ pub(crate) fn build_anthropic_usage_from_responses(usage: Option<&Value>) -> Val
result["cache_read_input_tokens"] = json!(cached);
}
}
// GPT-5.6+ reports cache writes in the nested OpenAI token-details object.
// Treat writes as Anthropic cache creation so the downstream client and
// billing layer can distinguish them from fresh input.
let nested_cache_write = u
.pointer("/input_tokens_details/cache_write_tokens")
.and_then(|v| v.as_u64())
.or_else(|| {
u.pointer("/prompt_tokens_details/cache_write_tokens")
.and_then(|v| v.as_u64())
});
if let Some(cache_write) = nested_cache_write {
result["cache_creation_input_tokens"] = json!(cache_write);
}
// Step 2: Direct Anthropic-style fields override (authoritative if present)
// These preserve cache tokens even if input/output_tokens are missing
@@ -1710,6 +1724,21 @@ mod tests {
assert_eq!(result["cache_read_input_tokens"], json!(80));
}
#[test]
fn test_build_usage_cache_write_tokens_from_nested_details() {
let result = build_anthropic_usage_from_responses(Some(&json!({
"input_tokens": 100,
"output_tokens": 10,
"input_tokens_details": {
"cached_tokens": 30,
"cache_write_tokens": 20
}
})));
assert_eq!(result["input_tokens"], json!(50));
assert_eq!(result["cache_read_input_tokens"], json!(30));
assert_eq!(result["cache_creation_input_tokens"], json!(20));
}
#[test]
fn test_build_usage_cache_tokens_direct_override() {
let result = build_anthropic_usage_from_responses(Some(&json!({
src-tauri/src/proxy/usage/calculator.rs
+8 -5
View File
@@ -76,10 +76,13 @@ impl CostCalculator {
) -> CostBreakdown {
let million = Decimal::from(1_000_000);
// OpenAI/Gemini 风格的 input_tokens 包含缓存命中，需要扣除后再按输入价计费；
// OpenAI/Gemini 风格的 input_tokens 包含缓存读取和写入，需要扣除后再按输入价计费；
// Claude/Anthropic 风格的 input_tokens 已经是 fresh input，不能再次扣减。
let billable_input_tokens = if input_includes_cache_read {
usage.input_tokens.saturating_sub(usage.cache_read_tokens)
usage
.input_tokens
.saturating_sub(usage.cache_read_tokens)
.saturating_sub(usage.cache_creation_tokens)
} else {
usage.input_tokens
};
@@ -197,15 +200,15 @@ mod tests {
let cost = CostCalculator::calculate_for_app("codex", &usage, &pricing, multiplier);
// Codex/OpenAI 语义：input_tokens 包含 cached_tokens，需要扣除 cache_read_tokens
assert_eq!(cost.input_cost, Decimal::from_str("0.0024").unwrap());
// Codex/OpenAI 语义：input_tokens 包含 cache read/write，两桶都需扣除。
assert_eq!(cost.input_cost, Decimal::from_str("0.0021").unwrap());
assert_eq!(cost.output_cost, Decimal::from_str("0.0075").unwrap());
assert_eq!(cost.cache_read_cost, Decimal::from_str("0.00006").unwrap());
assert_eq!(
cost.cache_creation_cost,
Decimal::from_str("0.000375").unwrap()
);
assert_eq!(cost.total_cost, Decimal::from_str("0.010335").unwrap());
assert_eq!(cost.total_cost, Decimal::from_str("0.010035").unwrap());
}
#[test]
src-tauri/src/proxy/usage/logger.rs
+9 -1
View File
@@ -4,6 +4,7 @@ use super::calculator::{CostBreakdown, CostCalculator, ModelPricing};
use super::parser::TokenUsage;
use crate::database::{Database, PRICING_SOURCE_REQUEST, PRICING_SOURCE_RESPONSE};
use crate::error::AppError;
use crate::services::sql_helpers::{INPUT_TOKEN_SEMANTICS_FRESH, INPUT_TOKEN_SEMANTICS_TOTAL};
use crate::services::usage_stats::{find_model_pricing_row, is_placeholder_pricing_model};
use rust_decimal::Decimal;
use std::str::FromStr;
@@ -70,15 +71,21 @@ impl<'a> UsageLogger<'a> {
};
let created_at = chrono::Utc::now().timestamp();
let input_token_semantics = if matches!(log.app_type.as_str(), "codex" | "gemini") {
INPUT_TOKEN_SEMANTICS_TOTAL
} else {
INPUT_TOKEN_SEMANTICS_FRESH
};
conn.execute(
"INSERT OR REPLACE INTO proxy_request_logs (
request_id, provider_id, app_type, model, request_model, pricing_model,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_token_semantics,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd,
latency_ms, first_token_ms, status_code, error_message, session_id,
provider_type, is_streaming, cost_multiplier, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24)",
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)",
rusqlite::params![
log.request_id,
log.provider_id,
@@ -90,6 +97,7 @@ impl<'a> UsageLogger<'a> {
log.usage.output_tokens,
log.usage.cache_read_tokens,
log.usage.cache_creation_tokens,
input_token_semantics,
input_cost,
output_cost,
cache_read_cost,
src-tauri/src/proxy/usage/parser.rs
+55 -36
View File
@@ -9,6 +9,24 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
fn openai_cache_read_tokens(usage: &Value) -> u32 {
usage
.get("cache_read_input_tokens")
.or_else(|| usage.pointer("/input_tokens_details/cached_tokens"))
.or_else(|| usage.pointer("/prompt_tokens_details/cached_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0) as u32
}
fn openai_cache_write_tokens(usage: &Value) -> u32 {
usage
.get("cache_creation_input_tokens")
.or_else(|| usage.pointer("/input_tokens_details/cache_write_tokens"))
.or_else(|| usage.pointer("/prompt_tokens_details/cache_write_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0) as u32
}
/// Session 日志 request_id 前缀，与 `session_usage.rs` 中的格式保持一致
pub const SESSION_REQUEST_ID_PREFIX: &str = "session:";
@@ -250,25 +268,14 @@ impl TokenUsage {
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let cached_tokens = usage
.get("cache_read_input_tokens")
.and_then(|v| v.as_u64())
.or_else(|| {
usage
.get("input_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(|v| v.as_u64())
})
.unwrap_or(0) as u32;
let cached_tokens = openai_cache_read_tokens(usage);
let cache_write_tokens = openai_cache_write_tokens(usage);
Some(Self {
input_tokens: input_tokens? as u32,
output_tokens: output_tokens? as u32,
cache_read_tokens: cached_tokens,
cache_creation_tokens: usage
.get("cache_creation_input_tokens")
.and_then(|v| v.as_u64())
.unwrap_or(0) as u32,
cache_creation_tokens: cache_write_tokens,
model,
message_id: None,
})
@@ -285,19 +292,13 @@ impl TokenUsage {
let output_tokens = usage.get("output_tokens")?.as_u64()? as u32;
// 获取 cached_tokens (可能在 cache_read_input_tokens 或 input_tokens_details 中)
let cached_tokens = usage
.get("cache_read_input_tokens")
.and_then(|v| v.as_u64())
.or_else(|| {
usage
.get("input_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(|v| v.as_u64())
})
.unwrap_or(0) as u32;
let cached_tokens = openai_cache_read_tokens(usage);
let cache_write_tokens = openai_cache_write_tokens(usage);
// 调整 input_tokens: 减去 cached_tokens
let adjusted_input = input_tokens.saturating_sub(cached_tokens);
// 调整 input_tokens: OpenAI total input 同时包含 cache read/write 两桶。
let adjusted_input = input_tokens
.saturating_sub(cached_tokens)
.saturating_sub(cache_write_tokens);
// 提取响应中的模型名称
let model = body
@@ -309,10 +310,7 @@ impl TokenUsage {
input_tokens: adjusted_input,
output_tokens,
cache_read_tokens: cached_tokens,
cache_creation_tokens: usage
.get("cache_creation_input_tokens")
.and_then(|v| v.as_u64())
.unwrap_or(0) as u32,
cache_creation_tokens: cache_write_tokens,
model,
message_id: None,
})
@@ -391,11 +389,8 @@ impl TokenUsage {
let completion_tokens = usage.get("completion_tokens").and_then(|v| v.as_u64())?;
// 获取 cached_tokens (可能在 prompt_tokens_details 中)
let cached_tokens = usage
.get("prompt_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(|v| v.as_u64())
.unwrap_or(0) as u32;
let cached_tokens = openai_cache_read_tokens(usage);
let cache_write_tokens = openai_cache_write_tokens(usage);
// 提取响应中的模型名称
let model = body
@@ -407,7 +402,7 @@ impl TokenUsage {
input_tokens: prompt_tokens as u32,
output_tokens: completion_tokens as u32,
cache_read_tokens: cached_tokens,
cache_creation_tokens: 0,
cache_creation_tokens: cache_write_tokens,
model,
message_id: None,
})
@@ -797,6 +792,30 @@ mod tests {
assert_eq!(usage.cache_read_tokens, 300);
}
#[test]
fn test_codex_response_parsing_cache_write_tokens_in_details() {
let response = json!({
"usage": {
"input_tokens": 1000,
"output_tokens": 500,
"input_tokens_details": {
"cached_tokens": 300,
"cache_write_tokens": 200
}
}
});
let usage = TokenUsage::from_codex_response(&response).unwrap();
assert_eq!(usage.input_tokens, 1000);
assert_eq!(usage.cache_read_tokens, 300);
assert_eq!(usage.cache_creation_tokens, 200);
let adjusted = TokenUsage::from_codex_response_adjusted(&response).unwrap();
assert_eq!(adjusted.input_tokens, 500);
assert_eq!(adjusted.cache_read_tokens, 300);
assert_eq!(adjusted.cache_creation_tokens, 200);
}
#[test]
fn test_codex_response_adjusted() {
let response = json!({
src-tauri/src/services/sql_helpers.rs
+52 -6
View File
@@ -18,13 +18,16 @@
/// would happen with the opposite default.
const CACHE_INCLUSIVE_APP_TYPES: &[&str] = &["codex", "gemini"];
pub(crate) const INPUT_TOKEN_SEMANTICS_LEGACY: i64 = 0;
pub(crate) const INPUT_TOKEN_SEMANTICS_TOTAL: i64 = 1;
pub(crate) const INPUT_TOKEN_SEMANTICS_FRESH: i64 = 2;
/// Build an SQL expression that returns the cache-normalized `input_tokens`
/// for a single row in `proxy_request_logs` or `usage_daily_rollups`.
///
/// For rows whose `app_type` is in [`CACHE_INCLUSIVE_APP_TYPES`] and
/// `input_tokens >= cache_read_tokens`, returns
/// `input_tokens - cache_read_tokens`. For all other rows the original
/// `input_tokens` is returned unchanged.
/// Legacy rows subtract cache reads only. New total-inclusive rows subtract
/// both cache reads and writes. Rollups normalized to fresh input are returned
/// unchanged.
///
/// Pass an empty string to reference the columns directly (no alias),
/// or a table alias such as `"l"` to emit `l.input_tokens` style references.
@@ -40,7 +43,15 @@ pub fn fresh_input_sql(alias: &str) -> String {
.collect::<Vec<_>>()
.join(", ");
format!(
"CASE WHEN {prefix}app_type IN ({app_type_list}) AND {prefix}input_tokens >= {prefix}cache_read_tokens \
"CASE \
WHEN {prefix}input_token_semantics = {INPUT_TOKEN_SEMANTICS_FRESH} THEN {prefix}input_tokens \
WHEN {prefix}app_type IN ({app_type_list}) \
AND {prefix}input_token_semantics = {INPUT_TOKEN_SEMANTICS_TOTAL} \
AND {prefix}input_tokens >= ({prefix}cache_read_tokens + {prefix}cache_creation_tokens) \
THEN ({prefix}input_tokens - {prefix}cache_read_tokens - {prefix}cache_creation_tokens) \
WHEN {prefix}app_type IN ({app_type_list}) \
AND {prefix}input_token_semantics = {INPUT_TOKEN_SEMANTICS_LEGACY} \
AND {prefix}input_tokens >= {prefix}cache_read_tokens \
THEN ({prefix}input_tokens - {prefix}cache_read_tokens) \
ELSE {prefix}input_tokens END"
)
@@ -60,7 +71,8 @@ mod tests {
input_tokens INTEGER NOT NULL DEFAULT 0,
output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
cache_creation_tokens INTEGER NOT NULL DEFAULT 0
cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
input_token_semantics INTEGER NOT NULL DEFAULT 0
);",
)
.unwrap();
@@ -131,4 +143,38 @@ mod tests {
let value: i64 = conn.query_row(&sql, [], |r| r.get(0)).unwrap();
assert_eq!(value, 100);
}
#[test]
fn fresh_input_subtracts_cache_write_for_total_semantics() {
let conn = setup_conn();
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, app_type, input_tokens, cache_read_tokens,
cache_creation_tokens, input_token_semantics
) VALUES ('codex-total', 'codex', 1000, 300, 200, ?1)",
[INPUT_TOKEN_SEMANTICS_TOTAL],
)
.unwrap();
let expr = fresh_input_sql("l");
let sql = format!("SELECT {expr} FROM proxy_request_logs l");
let value: i64 = conn.query_row(&sql, [], |row| row.get(0)).unwrap();
assert_eq!(value, 500);
}
#[test]
fn fresh_input_keeps_normalized_rollup_value() {
let conn = setup_conn();
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, app_type, input_tokens, cache_read_tokens,
cache_creation_tokens, input_token_semantics
) VALUES ('codex-fresh', 'codex', 500, 300, 200, ?1)",
[INPUT_TOKEN_SEMANTICS_FRESH],
)
.unwrap();
let expr = fresh_input_sql("l");
let sql = format!("SELECT {expr} FROM proxy_request_logs l");
let value: i64 = conn.query_row(&sql, [], |row| row.get(0)).unwrap();
assert_eq!(value, 500);
}
}
src-tauri/src/services/usage_stats.rs
+99 -14
View File
@@ -5,7 +5,9 @@
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use crate::proxy::usage::calculator::ModelPricing;
use crate::services::sql_helpers::fresh_input_sql;
use crate::services::sql_helpers::{
fresh_input_sql, INPUT_TOKEN_SEMANTICS_FRESH, INPUT_TOKEN_SEMANTICS_TOTAL,
};
use chrono::{Local, NaiveDate, TimeZone, Timelike};
use rusqlite::{params, Connection, OptionalExtension};
use serde::{Deserialize, Serialize};
@@ -135,6 +137,9 @@ pub struct RequestLogDetail {
pub output_tokens: u32,
pub cache_read_tokens: u32,
pub cache_creation_tokens: u32,
/// Internal storage semantics; omitted from the UI/API payload.
#[serde(skip)]
pub input_token_semantics: i64,
pub input_cost_usd: String,
pub output_cost_usd: String,
pub cache_read_cost_usd: String,
@@ -154,15 +159,15 @@ pub struct RequestLogDetail {
pub pricing_model: Option<String>,
}
/// 把 25 列的查询结果映射为 `RequestLogDetail`。
/// 把 26 列的查询结果映射为 `RequestLogDetail`。
///
/// 调用方的 SELECT **必须**按以下顺序返回 25 列：
/// 调用方的 SELECT **必须**按以下顺序返回 26 列：
/// `request_id, provider_id, provider_name, app_type, model, request_model,
/// cost_multiplier, input_tokens, output_tokens, cache_read_tokens,
/// cache_creation_tokens, input_cost_usd, output_cost_usd, cache_read_cost_usd,
/// cache_creation_cost_usd, total_cost_usd, is_streaming, latency_ms,
/// first_token_ms, duration_ms, status_code, error_message, created_at,
/// data_source, pricing_model`
/// data_source, pricing_model, input_token_semantics`
///
/// 不需要 provider_name 时（如 backfill）SELECT `NULL AS provider_name` 占位即可。
fn row_to_request_log_detail(row: &rusqlite::Row<'_>) -> rusqlite::Result<RequestLogDetail> {
@@ -194,6 +199,7 @@ fn row_to_request_log_detail(row: &rusqlite::Row<'_>) -> rusqlite::Result<Reques
created_at: row.get(22)?,
data_source: row.get(23)?,
pricing_model: row.get(24)?,
input_token_semantics: row.get::<_, i64>(25)?,
})
}
@@ -1526,7 +1532,8 @@ impl Database {
l.input_tokens, l.output_tokens, l.cache_read_tokens, l.cache_creation_tokens,
l.input_cost_usd, l.output_cost_usd, l.cache_read_cost_usd, l.cache_creation_cost_usd, l.total_cost_usd,
l.is_streaming, l.latency_ms, l.first_token_ms, l.duration_ms,
l.status_code, l.error_message, l.created_at, l.data_source, l.pricing_model
l.status_code, l.error_message, l.created_at, l.data_source, l.pricing_model,
l.input_token_semantics
FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
{where_clause}
@@ -1569,7 +1576,8 @@ impl Database {
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd,
is_streaming, latency_ms, first_token_ms, duration_ms,
status_code, error_message, created_at, l.data_source, l.pricing_model
status_code, error_message, created_at, l.data_source, l.pricing_model,
l.input_token_semantics
FROM proxy_request_logs l
LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
WHERE l.request_id = ?"
@@ -1725,7 +1733,7 @@ impl Database {
input_cost_usd, output_cost_usd, cache_read_cost_usd,
cache_creation_cost_usd, total_cost_usd, is_streaming, latency_ms,
first_token_ms, duration_ms, status_code, error_message, created_at,
data_source, pricing_model
data_source, pricing_model, input_token_semantics
FROM proxy_request_logs
WHERE CAST(total_cost_usd AS REAL) <= 0
AND (input_tokens > 0 OR output_tokens > 0
@@ -1806,15 +1814,21 @@ impl Database {
let million = rust_decimal::Decimal::from(1_000_000u64);
// 与 CostCalculator::calculate_for_app 保持一致的计算逻辑：
// 1. Codex/Gemini 的 input_tokens 包含 cache_read_tokens，需要扣除后按输入价计费
// 1. 历史 Codex/Gemini 行只包含 cache read；新 total 行还包含 cache write。
// 2. Claude/Anthropic 的 input_tokens 已经是 fresh input，不能再次扣减
// 3. 各项成本是基础成本（不含倍率），倍率只作用于最终总价
let input_includes_cache_read = matches!(log.app_type.as_str(), "codex" | "gemini");
let billable_input_tokens = if input_includes_cache_read {
(log.input_tokens as u64).saturating_sub(log.cache_read_tokens as u64)
} else {
log.input_tokens as u64
};
let cache_inclusive_app = matches!(log.app_type.as_str(), "codex" | "gemini");
let billable_input_tokens =
if !cache_inclusive_app || log.input_token_semantics == INPUT_TOKEN_SEMANTICS_FRESH {
log.input_tokens as u64
} else if log.input_token_semantics == INPUT_TOKEN_SEMANTICS_TOTAL {
(log.input_tokens as u64)
.saturating_sub(log.cache_read_tokens as u64)
.saturating_sub(log.cache_creation_tokens as u64)
} else {
// v12 and earlier: input included cache reads but excluded cache writes.
(log.input_tokens as u64).saturating_sub(log.cache_read_tokens as u64)
};
let input_cost =
rust_decimal::Decimal::from(billable_input_tokens) * pricing.input / million;
let output_cost =
@@ -2492,6 +2506,77 @@ mod tests {
Ok(())
}
#[test]
fn test_backfill_distinguishes_legacy_and_total_cache_semantics() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = lock_conn!(db.conn);
// v12 mirror row: input = fresh + read; creation was reported separately.
insert_usage_log(
&conn,
"legacy-cache-semantics",
"codex",
"p1",
"gpt-5.5",
"proxy",
1000,
800_000,
0,
600_000,
200_000,
200,
"0",
)?;
// v13 proxy row: input = fresh + read + creation.
insert_usage_log(
&conn,
"total-cache-semantics",
"codex",
"p1",
"gpt-5.5",
"proxy",
1001,
1_000_000,
0,
600_000,
200_000,
200,
"0",
)?;
conn.execute(
"UPDATE proxy_request_logs
SET input_token_semantics = ?1
WHERE request_id = 'total-cache-semantics'",
[INPUT_TOKEN_SEMANTICS_TOTAL],
)?;
}
assert_eq!(db.backfill_missing_usage_costs()?, 2);
let conn = lock_conn!(db.conn);
let mut stmt = conn.prepare(
"SELECT request_id, input_cost_usd
FROM proxy_request_logs
WHERE request_id IN ('legacy-cache-semantics', 'total-cache-semantics')
ORDER BY request_id",
)?;
let rows = stmt
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?
.collect::<Result<Vec<_>, _>>()?;
assert_eq!(
rows,
vec![
("legacy-cache-semantics".to_string(), "1.000000".to_string()),
("total-cache-semantics".to_string(), "1.000000".to_string()),
]
);
Ok(())
}
#[test]
fn test_backfill_missing_usage_costs_uses_stored_multiplier() -> Result<(), AppError> {
let db = Database::memory()?;