From 1cfc56d280bb87a5188e40023a80a8c8abd3a7dc Mon Sep 17 00:00:00 2001 From: argszero Date: Sat, 24 Oct 2026 02:41:16 +0800 Subject: [PATCH 1/2] feat(billing): bill cache-write (cache-creation) tokens MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cache_write_input_per_m` was deserialized, validated and documented, but reached neither the database nor the settlement path: `models` had no such column, `dao::get_model_price` did not select it, and `billing::raw_cost` took no cache-write operand. The config note explained the gap as "there is no path to obtain the cache-write token count", which was too strong: the count already arrives in `usage.cache_creation_input_tokens` and `sse.rs` already deserializes it — only for protocol conversion, after which it is dropped. What was missing was a billing path, not a parse path. The wire, Anthropic only (#432): * storage — additive `ensure_column` on `models`, `usage_records` and `transactions` (no `SCHEMA_VERSION` bump, matching the earlier price columns); `seed_models` carries the price in its INSERT and its `ON CONFLICT DO UPDATE` list; `dao::get_model_price` selects it. * carry — `parse_usage`, `parse_anthropic_usage`, `UsageCapture` and `settle_usage` gain the fourth count. Protocols that do not report the field pass 0 rather than a guess. * price — `raw_cost`/`calc_points` take the operand, priced by the model-level `cache_write_input_per_m`. Tiers and peak hours do not carry it, so selection cannot change it. * archive — both archive specs and their row-to-JSON builders cover the new column, so archived detail does not silently lose it. `tokens` keeps its meaning (input + cached + output): five readers derive "input" as `tokens - cached_tokens - output_tokens`, so folding cache writes in would inflate the input shown for every Anthropic request. The entry guard now also counts the fourth segment, so a response reporting only cache writes is no longer dropped as "no data to settle". Default stays 0 and never falls back to `input_per_m`, so no existing bill moves; a test pins that a call with cache-write tokens charges exactly the same points as the same call without them. Refs #432 --- config/config.example.toml | 9 +- src/archive.rs | 16 ++- src/billing.rs | 90 ++++++++++-- src/config.rs | 14 +- src/dao.rs | 11 +- src/db.rs | 35 ++++- src/gateway.rs | 284 +++++++++++++++++++++++++++++++------ src/routes/mod.rs | 1 + 8 files changed, 387 insertions(+), 73 deletions(-) diff --git a/config/config.example.toml b/config/config.example.toml index c2ae488..a15f999 100644 --- a/config/config.example.toml +++ b/config/config.example.toml @@ -347,10 +347,11 @@ endpoints = [ # 模型（厂商/名称/价格/上下文/读图等）全部在此定义；启动时 upsert 进 DB。 # 单位：每百万 tokens（currency 对应币种）；input_per_m = 缓存未命中输入价， # cache_hit_input_per_m = 缓存命中输入价（缺省 0 = 命中免费）。 -# 缓存写入价（rant 2026-10-16T14:08:14）：可选 cache_write_input_per_m（每百万 tokens）， -# 缺省 0 = 缓存写入不收费（与今天一致）。⚠️ 这个字段目前**只表达、不收费**：上游响应里 -# 没有取「缓存写入 token 数」的通路（gateway parse_usage / sse 都不产出它），接进计费要 -# 另加一次用量解析，属后续改动。 +# 缓存写入价（rant 2026-10-16T14:08:14；#432 起**参与计费**）：可选 +# cache_write_input_per_m（每百万 tokens），缺省 0 = 缓存写入不收费。计费口径 = 上游报的 +# 缓存写入 token 数 × 本字段；缺省 0 时收 0 元，与未接入前完全一致。 +# 只有 anthropic 的上游响应带这个 token 数（usage.cache_creation_input_tokens），其它协议 +# 按 0 处理 —— 拿不到就不收，不估算。分档价与高峰价都不带这一档 ⇒ 它不随选档或时段变化。 # 分档价（rant 2026-10-16T14:08:14 第 2 步：settle 已按请求输入 token 数选档）： # 可选 [[models.tiers]] 数组按**单次请求输入 token 数**分档 —— 输入 = 缓存未命中 ＋ 缓存 # 命中那一段 prompt（不是用户累计用量，也不是输出）。from_tokens 是本档下界（含），各档 diff --git a/src/archive.rs b/src/archive.rs index de5a092..affbb4f 100644 --- a/src/archive.rs +++ b/src/archive.rs @@ -78,6 +78,7 @@ pub const TRANSACTIONS: Spec = Spec { "tokens", "cached_tokens", "output_tokens", + "cache_write_tokens", "pts", "type", "status", @@ -103,6 +104,7 @@ pub const USAGE_RECORDS: Spec = Spec { "tokens", "cached_tokens", "output_tokens", + "cache_write_tokens", "cost", "time", ], @@ -267,10 +269,11 @@ fn tx_json(row: &rusqlite::Row<'_>) -> rusqlite::Result { "tokens": row.get::<_, f64>(6)?, "cached_tokens": row.get::<_, f64>(7)?, "output_tokens": row.get::<_, f64>(8)?, - "pts": row.get::<_, f64>(9)?, - "type": row.get::<_, String>(10)?, - "status": row.get::<_, String>(11)?, - "time": row.get::<_, String>(12)?, + "cache_write_tokens": row.get::<_, f64>(9)?, + "pts": row.get::<_, f64>(10)?, + "type": row.get::<_, String>(11)?, + "status": row.get::<_, String>(12)?, + "time": row.get::<_, String>(13)?, }); Ok(v.to_string()) } @@ -286,8 +289,9 @@ fn usage_json(row: &rusqlite::Row<'_>) -> rusqlite::Result { "tokens": row.get::<_, f64>(5)?, "cached_tokens": row.get::<_, f64>(6)?, "output_tokens": row.get::<_, f64>(7)?, - "cost": row.get::<_, f64>(8)?, - "time": row.get::<_, String>(9)?, + "cache_write_tokens": row.get::<_, f64>(8)?, + "cost": row.get::<_, f64>(9)?, + "time": row.get::<_, String>(10)?, }); Ok(v.to_string()) } diff --git a/src/billing.rs b/src/billing.rs index b2d78c2..8cd1c63 100644 --- a/src/billing.rs +++ b/src/billing.rs @@ -40,18 +40,29 @@ pub fn to_anchor(amount: f64, model_currency: &str, anchor: &str) -> f64 { /// 原始成本（模型自身货币）：USD/CNY 每百万 token 单价 × token 数。 /// 输入分「缓存未命中」与「缓存命中」两档价（rant 2026-08-20T10:17:27）： -/// 未命中 × input_per_m + 命中 × cache_hit_input_per_m（缺省 0 → 命中部分免费） +/// 未命中 × input_per_m + 命中 × cache_hit_input_per_m（缺省 0 → 命中部分免费）。 +/// +/// 缓存写入（cache creation）是**第三档**输入价（#432）：`cache_write_tokens` × +/// `cache_write_input_per_m`。它是把 prompt 写进上游缓存的那一次收费，与「命中」互斥， +/// 也不与未命中重叠 —— 上游上报的三段本来就互斥。缺省 0 ⇒ 收 0 元，与 #432 之前的账完全一致。 +/// +/// 缓存写入价**不随选档变化**：`ModelTier` 只带输入/命中/输出三档价，故档位切换不影响它， +/// 也没有高峰时段变体（config 只有 `peak_*` 三个字段）。要放开这两条得先有配置字段。 +#[allow(clippy::too_many_arguments)] // 与 calc_points 同：token 段与单价平铺，保持可读 pub fn raw_cost( input_tokens: f64, cached_tokens: f64, output_tokens: f64, + cache_write_tokens: f64, input_per_m: f64, cache_hit_input_per_m: f64, output_per_m: f64, + cache_write_input_per_m: f64, ) -> f64 { input_tokens * input_per_m / 1_000_000.0 + cached_tokens * cache_hit_input_per_m / 1_000_000.0 + output_tokens * output_per_m / 1_000_000.0 + + cache_write_tokens * cache_write_input_per_m / 1_000_000.0 } /// 计算点数：锚定货币成本 × points_per_unit @@ -60,9 +71,11 @@ pub fn calc_points( input_tokens: f64, cached_tokens: f64, output_tokens: f64, + cache_write_tokens: f64, input_per_m: f64, cache_hit_input_per_m: f64, output_per_m: f64, + cache_write_input_per_m: f64, points_per_unit: u32, model_currency: &str, anchor: &str, @@ -71,9 +84,11 @@ pub fn calc_points( input_tokens, cached_tokens, output_tokens, + cache_write_tokens, input_per_m, cache_hit_input_per_m, output_per_m, + cache_write_input_per_m, ); round5(to_anchor(cost, model_currency, anchor) * points_per_unit as f64) } @@ -176,6 +191,9 @@ pub struct SettleParams { pub cached_tokens: f64, /// 输出 token 数（v0.7.4，rant 2026-08-21T14:53:20：单次调用明细 输入/缓存/输出） pub output_tokens: f64, + /// 缓存写入（cache creation）token 数（#432）。**不进 `tokens`**：`tokens` 仍是 + /// input + cached + output，读侧有 5 处按那个和反推「输入」（见 `db::migrate` 的 v18 注）。 + pub cache_write_tokens: f64, /// 消费者应扣点数 pub pts: f64, /// 锚定货币成本（usage_records.cost） @@ -210,8 +228,8 @@ pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { // transactions：consume（消费者）+ earn（分享者），counterpart 记对方； // api_key_id 记录分发 key（v11，rant 2026-08-22T17:21:39 需求 2：Key 列显示 api_keys.name） tx.execute( - "INSERT INTO transactions (user_id, counterpart, key_id, api_key_id, model, tokens, cached_tokens, output_tokens, pts, type, status) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'consume', '成功')", + "INSERT INTO transactions (user_id, counterpart, key_id, api_key_id, model, tokens, cached_tokens, output_tokens, cache_write_tokens, pts, type, status) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'consume', '成功')", rusqlite::params![ p.consumer_id, p.owner_id.to_string(), @@ -221,12 +239,13 @@ pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { p.tokens, p.cached_tokens, p.output_tokens, + p.cache_write_tokens, p.pts ], )?; tx.execute( - "INSERT INTO transactions (user_id, counterpart, key_id, api_key_id, model, tokens, cached_tokens, output_tokens, pts, type, status) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'earn', '成功')", + "INSERT INTO transactions (user_id, counterpart, key_id, api_key_id, model, tokens, cached_tokens, output_tokens, cache_write_tokens, pts, type, status) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'earn', '成功')", rusqlite::params![ p.owner_id, p.consumer_id.to_string(), @@ -236,14 +255,15 @@ pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { p.tokens, p.cached_tokens, p.output_tokens, + p.cache_write_tokens, earn ], )?; // usage_records tx.execute( - "INSERT INTO usage_records (user_id, api_key_id, key_id, model, tokens, cached_tokens, output_tokens, cost) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + "INSERT INTO usage_records (user_id, api_key_id, key_id, model, tokens, cached_tokens, output_tokens, cache_write_tokens, cost) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", rusqlite::params![ p.consumer_id, p.api_key_id, @@ -252,6 +272,7 @@ pub fn settle(conn: &mut Connection, p: &SettleParams) -> Result<()> { p.tokens, p.cached_tokens, p.output_tokens, + p.cache_write_tokens, p.cost ], )?; @@ -305,17 +326,17 @@ mod tests { fn calc_points_basic() { // 100 prompt × 10 USD/M + 50 completion × 20 USD/M = 0.002 USD // → CNY 锚定 ×7.2 = 0.0144 CNY → 1 点/CNY = 0.0144 点（保留 5 位小数） - let pts = calc_points(100.0, 0.0, 50.0, 10.0, 0.0, 20.0, 1, "USD", "CNY"); + let pts = calc_points(100.0, 0.0, 50.0, 0.0, 10.0, 0.0, 20.0, 0.0, 1, "USD", "CNY"); assert!((pts - 0.0144).abs() < 1e-9, "pts={pts}"); } #[test] fn calc_points_currency_conversion() { // 1000 prompt × 10 CNY/M = 0.01 CNY → 1 点/CNY = 0.01 点 - let pts = calc_points(1000.0, 0.0, 0.0, 10.0, 0.0, 0.0, 1, "CNY", "CNY"); + let pts = calc_points(1000.0, 0.0, 0.0, 0.0, 10.0, 0.0, 0.0, 0.0, 1, "CNY", "CNY"); assert!((pts - 0.01).abs() < 1e-9, "pts={pts}"); // USD 价 → CNY 锚定 ×7.2：1000 prompt × 10 USD/M = 0.01 USD = 0.072 CNY - let pts_usd = calc_points(1000.0, 0.0, 0.0, 10.0, 0.0, 0.0, 1, "USD", "CNY"); + let pts_usd = calc_points(1000.0, 0.0, 0.0, 0.0, 10.0, 0.0, 0.0, 0.0, 1, "USD", "CNY"); assert!((pts_usd - 0.072).abs() < 1e-9, "pts_usd={pts_usd}"); } @@ -324,14 +345,36 @@ mod tests { // rant 2026-08-20T10:17:27：输入分「缓存未命中 / 命中」两档价。 // 1000 未命中 × 1.5 CNY/M + 1000 命中 × 0.05 CNY/M + 1000 输出 × 4.5 CNY/M // = (1500 + 50 + 4500)/1e6 CNY = 0.00605 CNY → 1 点/CNY = 0.00605 点 - let pts = calc_points(1000.0, 1000.0, 1000.0, 1.5, 0.05, 4.5, 1, "CNY", "CNY"); + let pts = calc_points( + 1000.0, 1000.0, 1000.0, 0.0, 1.5, 0.05, 4.5, 0.0, 1, "CNY", "CNY", + ); assert!((pts - 0.00605).abs() < 1e-9, "pts={pts}"); // 命中价缺省 0 → 命中部分不计费（只按未命中 + 输出） - let pts_zero = calc_points(1000.0, 1000.0, 1000.0, 1.5, 0.0, 4.5, 1, "CNY", "CNY"); + let pts_zero = calc_points( + 1000.0, 1000.0, 1000.0, 0.0, 1.5, 0.0, 4.5, 0.0, 1, "CNY", "CNY", + ); let expect_zero = (1500.0 + 4500.0) / 1e6; assert!((pts_zero - expect_zero).abs() < 1e-9, "pts_zero={pts_zero}"); } + /// #432：缓存写入是第三档输入价 —— 只有配置了 `cache_write_input_per_m` 才收， + /// 且只按缓存写入的 token 数收，不与未命中/命中重叠。 + #[test] + fn calc_points_cache_write_is_a_third_input_price() { + // 1000 写入 × 2.0 CNY/M = 2000/1e6 CNY = 0.002 点（1 点/元）；未命中与输出都为 0 + let pts = calc_points(0.0, 0.0, 0.0, 1000.0, 1.5, 0.05, 4.5, 2.0, 1, "CNY", "CNY"); + assert!((pts - 0.002).abs() < 1e-9, "写入单独计价: {pts}"); + // 缺省 0 = 不收费：同样的调用，缓存写入那一段**一分不收** + let pts_free = calc_points(0.0, 0.0, 0.0, 1000.0, 1.5, 0.05, 4.5, 0.0, 1, "CNY", "CNY"); + assert_eq!(pts_free, 0.0, "未配置缓存写入价 ⇒ 收 0: {pts_free}"); + // 三段互不重叠：未命中 1000 × 1.5 + 命中 1000 × 0.05 + 输出 1000 × 4.5 + 写入 1000 × 2.0 + // = (1500 + 50 + 4500 + 2000)/1e6 = 0.00805 CNY + let mixed = calc_points( + 1000.0, 1000.0, 1000.0, 1000.0, 1.5, 0.05, 4.5, 2.0, 1, "CNY", "CNY", + ); + assert!((mixed - 0.00805).abs() < 1e-9, "四段相加: {mixed}"); + } + #[test] fn is_peak_hour_beijing_boundaries() { // rant 2026-08-20T11:58:40：高峰 = 北京时 9:00-12:00、14:00-18:00。 @@ -427,14 +470,28 @@ mod tests { 1_000_000.0, 0.0, 0.0, + 0.0, peak.0, peak.1, peak.2, + 0.0, + 1, + "CNY", + "CNY", + ); + let pts_off = calc_points( + 1_000_000.0, + 0.0, + 0.0, + 0.0, + off.0, + off.1, + off.2, + 0.0, 1, "CNY", "CNY", ); - let pts_off = calc_points(1_000_000.0, 0.0, 0.0, off.0, off.1, off.2, 1, "CNY", "CNY"); assert!( (pts_peak - 3.0).abs() < 1e-9, "高峰 1M 输入 = 3.0 点: {pts_peak}" @@ -449,9 +506,11 @@ mod tests { 1_000_000.0, 0.0, 0.0, + 0.0, no_peak.0, no_peak.1, no_peak.2, + 0.0, 1, "CNY", "CNY", @@ -506,6 +565,7 @@ mod tests { cached_tokens: 30.0, output_tokens: 20.0, + cache_write_tokens: 0.0, pts: 2.0, cost: 0.002, }; @@ -639,6 +699,7 @@ mod tests { tokens: 1_000_000.0, cached_tokens: 0.0, output_tokens: 0.0, + cache_write_tokens: 0.0, pts: 4.5, cost: 0.0045, }; @@ -715,6 +776,7 @@ mod tests { cached_tokens: 0.0, output_tokens: 0.0, + cache_write_tokens: 0.0, pts: 1.0, cost: 0.001, }; @@ -774,6 +836,7 @@ mod tests { cached_tokens: 0.0, output_tokens: 0.0, + cache_write_tokens: 0.0, pts: 3.0, cost: 0.003, }; @@ -844,6 +907,7 @@ mod tests { cached_tokens: 0.0, output_tokens: 0.0, + cache_write_tokens: 0.0, pts: 2.0, cost: 0.002, }; diff --git a/src/config.rs b/src/config.rs index dbe5f91..debd1f1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -470,10 +470,16 @@ pub struct Model { /// 高峰时段输出价（缺省 0） #[serde(default)] pub peak_output_per_m: f64, - /// 缓存写入价（每百万 tokens；缺省 0 = 缓存写入不收费，与今天一致）。 - /// ⚠️ rant 2026-10-16T14:08:14 第 2 步只把**分档价**接进了 settle；这一个字段仍是 - /// 「只表达能力」—— 上游响应里没有取缓存写入 token 数的通路（`gateway::parse_usage` / - /// `sse::extract_*` 都不产出它），故今天不参与计费。接它需要另加一次用量解析，属后续改动。 + /// 缓存写入价（每百万 tokens；缺省 0 = 缓存写入不收费）。 + /// + /// **参与计费**（#432）：`gateway::parse_usage` 与同协议流式的 `UsageCapture` 从 + /// anthropic 的 `usage.cache_creation_input_tokens` 取到写入 token 数， + /// `billing::raw_cost` 按本字段计价，结算时落进 `usage_records` / `transactions` 的 + /// `cache_write_tokens` 列。缺省 0 ⇒ 收 0 元，与接入前的账逐位一致。 + /// + /// 三条边界：只有 anthropic 的上游响应带这个字段，其它协议按 0 处理（拿不到就不收， + /// 不估算）；`[[models.tiers]]` 不带这一档价，故它与选档无关；`peak_*` 也没有它的 + /// 变体，故高峰时段同样不改变它。 #[serde(default)] pub cache_write_input_per_m: f64, /// 按**单次请求输入长度**分档的价目（缺省空 = 不启用分档，一律沿用上面的单一价）。 diff --git a/src/dao.rs b/src/dao.rs index 9b1b4d9..9e62179 100644 --- a/src/dao.rs +++ b/src/dao.rs @@ -190,6 +190,9 @@ pub struct ModelPrice { pub input_per_m: f64, pub output_per_m: f64, pub cache_hit_input_per_m: f64, + /// 缓存写入（cache creation）价，缺省 0 = 不收费（#432）。模型级，与分档正交： + /// `ModelTier` 不带这一档价，故选档不影响它。 + pub cache_write_input_per_m: f64, pub peak_input_per_m: f64, pub peak_output_per_m: f64, pub peak_cache_hit_input_per_m: f64, @@ -201,7 +204,7 @@ pub struct ModelPrice { /// 高峰价缺省 0 = 不启用高峰计费（billing::effective_prices 处理回落） pub fn get_model_price(conn: &Connection, provider: &str, model: &str) -> Option { conn.query_row( - "SELECT input_per_m, output_per_m, cache_hit_input_per_m, \ + "SELECT input_per_m, output_per_m, cache_hit_input_per_m, cache_write_input_per_m, \ peak_input_per_m, peak_output_per_m, peak_cache_hit_input_per_m, currency, \ tiers_json \ FROM models WHERE provider = ?1 AND model = ?2", @@ -214,13 +217,14 @@ pub fn get_model_price(conn: &Connection, provider: &str, model: &str) -> Option r.get::<_, f64>(3)?, r.get::<_, f64>(4)?, r.get::<_, f64>(5)?, - r.get::<_, String>(6)?, + r.get::<_, f64>(6)?, r.get::<_, String>(7)?, + r.get::<_, String>(8)?, )) }, ) .ok() - .map(|(i, o, c, pin, pout, pcache, currency, tiers_json)| { + .map(|(i, o, c, cw, pin, pout, pcache, currency, tiers_json)| { // 解析失败不能装成「无分档」静默少收：留痕后按空档回落（＝沿用单一价，与今天一致）。 // 落库的 JSON 由本进程写（db::seed_models），正常路径不会走到这里。 let tiers: Vec = match serde_json::from_str(&tiers_json) { @@ -236,6 +240,7 @@ pub fn get_model_price(conn: &Connection, provider: &str, model: &str) -> Option input_per_m: i, output_per_m: o, cache_hit_input_per_m: c, + cache_write_input_per_m: cw, peak_input_per_m: pin, peak_output_per_m: pout, peak_cache_hit_input_per_m: pcache, diff --git a/src/db.rs b/src/db.rs index 2fcb7aa..132662d 100644 --- a/src/db.rs +++ b/src/db.rs @@ -370,6 +370,35 @@ pub fn migrate(conn: &Connection) -> Result<()> { "tiers_json", "tiers_json TEXT NOT NULL DEFAULT '[]'", )?; + // v18（#432）：缓存写入（cache creation）token 计量。三列全是加性补列，不动 + // `SCHEMA_VERSION` —— 版本号留给**加列表达不了**的形态（与 `cache_hit_input_per_m` / + // `peak_*` / `tiers_json` 的先例一致）。 + // + // 缺省 0 = 不收费 ⇒ **老库升级后没有任何一笔已发生的账发生变化**：`cache_write_tokens` + // 补 0、`cache_write_input_per_m` 补 0，而 0 价的收费额恒为 0。 + // + // `tokens` 列的语义**不变**：仍是 input + cached + output。读侧有 5 处按 + // `tokens - cached_tokens - output_tokens` 反推「输入 token」（`routes/wallet.rs` 的排序键 + // 与两条聚合、`db.rs` 的月用量聚合），把缓存写并进 `tokens` 会让每个上报了缓存写的 + // Anthropic 请求在界面上凭空多出一段「输入」—— 故缓存写单独记一列，不进总数。 + ensure_column( + conn, + "models", + "cache_write_input_per_m", + "cache_write_input_per_m REAL NOT NULL DEFAULT 0", + )?; + ensure_column( + conn, + "usage_records", + "cache_write_tokens", + "cache_write_tokens REAL NOT NULL DEFAULT 0", + )?; + ensure_column( + conn, + "transactions", + "cache_write_tokens", + "cache_write_tokens REAL NOT NULL DEFAULT 0", + )?; // v16（rant 2026-10-09T12:28:58 验收项 1）：交易明细的**可加汇总表** —— 维度 = 类型 + 用户 + // 模型 + key（两个身份都进维度，见 `tx_rollup` 模块头）+ 状态 + 时间（分钟级）；可加量 = // `pts` 与各 token 的 `SUM` + 行数 `COUNT`。折叠逻辑在 `src/tx_rollup.rs`；本步只建表。 @@ -630,8 +659,8 @@ pub fn seed_models(conn: &Connection, cfg: &crate::config::Config) -> Result<()> let tiers_json = serde_json::to_string(&m.tiers) .with_context(|| format!("序列化 model[{}/{}] 的分档价失败", m.provider, m.model))?; conn.execute( - "INSERT INTO models (provider, model, currency, input_per_m, output_per_m, context_window, context_length, max_output, vision, cache_hit_input_per_m, peak_input_per_m, peak_cache_hit_input_per_m, peak_output_per_m, tiers_json, updated_at) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, datetime('now')) \ + "INSERT INTO models (provider, model, currency, input_per_m, output_per_m, context_window, context_length, max_output, vision, cache_hit_input_per_m, cache_write_input_per_m, peak_input_per_m, peak_cache_hit_input_per_m, peak_output_per_m, tiers_json, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, datetime('now')) \ ON CONFLICT(provider, model) DO UPDATE SET \ currency = excluded.currency, \ input_per_m = excluded.input_per_m, \ @@ -641,6 +670,7 @@ pub fn seed_models(conn: &Connection, cfg: &crate::config::Config) -> Result<()> max_output = excluded.max_output, \ vision = excluded.vision, \ cache_hit_input_per_m = excluded.cache_hit_input_per_m, \ + cache_write_input_per_m = excluded.cache_write_input_per_m, \ peak_input_per_m = excluded.peak_input_per_m, \ peak_cache_hit_input_per_m = excluded.peak_cache_hit_input_per_m, \ peak_output_per_m = excluded.peak_output_per_m, \ @@ -657,6 +687,7 @@ pub fn seed_models(conn: &Connection, cfg: &crate::config::Config) -> Result<()> m.max_output, m.vision as i64, m.cache_hit_input_per_m, + m.cache_write_input_per_m, m.peak_input_per_m, m.peak_cache_hit_input_per_m, m.peak_output_per_m, diff --git a/src/gateway.rs b/src/gateway.rs index 691c890..9d2e5d3 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -48,16 +48,23 @@ fn extract_model(body: &str) -> Option { .map(|s| s.to_string()) } -/// 解析上游响应中的 usage → (uncached_input_tokens, cached_tokens, output_tokens) +/// 解析上游响应中的 usage → (uncached_input_tokens, cached_tokens, output_tokens, +/// cache_write_tokens) ///（rant 2026-08-20T10:17:27：输入区分「缓存命中/未命中」—— /// openai_chat → usage.prompt_cache_hit_tokens（DeepSeek 原生顶层拼写） /// 或 usage.prompt_tokens_details.cached_tokens（OpenAI 拼写） /// （prompt_tokens 含命中，需减除；优先级同 `UsageCapture::finish` / `sse.rs`）； /// anthropic → usage.cache_read_input_tokens（input_tokens 已不含命中，不减）； /// responses → usage.input_tokens_details.cached_tokens（input_tokens 含命中，需减除）） -fn parse_usage(body: &[u8], protocol: &str) -> (f64, f64, f64) { +/// +/// 第四个分量是**缓存写入**（cache creation）token 数（#432）。只有 anthropic 的上游响应 +/// 带 `usage.cache_creation_input_tokens`，且它的 `input_tokens` 本来就**不含**这一段 +/// （Anthropic 把 input / cache_read / cache_creation 三段分开报）⇒ 原样取出、不做加减。 +/// 其它协议没有这个字段 ⇒ 0，不是估算：拿不到就不收，与「缺数据 ≠ 数据为 0」同一判据 +/// （见 `UsageCapture::usage_seen`）。 +fn parse_usage(body: &[u8], protocol: &str) -> (f64, f64, f64, f64) { let Ok(v) = serde_json::from_slice::(body) else { - return (0.0, 0.0, 0.0); + return (0.0, 0.0, 0.0, 0.0); }; let usage = v.get("usage"); match protocol { @@ -74,7 +81,11 @@ fn parse_usage(body: &[u8], protocol: &str) -> (f64, f64, f64) { .and_then(|u| u.get("output_tokens")) .and_then(|x| x.as_f64()) .unwrap_or(0.0); - (input, cached, output) + let cache_write = usage + .and_then(|u| u.get("cache_creation_input_tokens")) + .and_then(|x| x.as_f64()) + .unwrap_or(0.0); + (input, cached, output, cache_write) } "responses" => { let total_input = usage @@ -89,7 +100,7 @@ fn parse_usage(body: &[u8], protocol: &str) -> (f64, f64, f64) { .and_then(|u| u.get("output_tokens")) .and_then(|x| x.as_f64()) .unwrap_or(0.0); - (total_input - cached, cached, output) + (total_input - cached, cached, output, 0.0) } _ => { let total_input = usage @@ -111,7 +122,7 @@ fn parse_usage(body: &[u8], protocol: &str) -> (f64, f64, f64) { .and_then(|u| u.get("completion_tokens")) .and_then(|x| x.as_f64()) .unwrap_or(0.0); - (total_input - cached, cached, output) + (total_input - cached, cached, output, 0.0) } } } @@ -157,7 +168,9 @@ fn decrypt_key(st: &AppState, key: &dao::KeyRow) -> Option { } } -/// 计量入账（非流式/流式共用）：tokens>0 才入账；失败仅记日志不影响透传 +/// 计量入账（非流式/流式共用）：四段 token（输入/命中/输出/缓存写入）全为 0 才不入账； +/// 失败仅记日志不影响透传 +#[allow(clippy::too_many_arguments)] // 四段 token（输入/命中/输出/缓存写入）平铺，同 billing::calc_points fn settle_usage( st: &AppState, auth: AuthUser, @@ -166,10 +179,14 @@ fn settle_usage( input_tokens: f64, cached_tokens: f64, output_tokens: f64, + cache_write_tokens: f64, ) { // 锁作用域严格限定在同步区内（绝不在 await 期间持有 MutexGuard） + // `tokens` 的语义**不变**（input + cached + output）：读侧有 5 处按它反推「输入」， + // 缓存写单独进自己的列（#432）。而「有没有计量依据」的判据必须把四段都算上 —— + // 一个只上报缓存写入的响应，今天会被这条早退当成「无数据」白送。 let tokens = input_tokens + cached_tokens + output_tokens; - if tokens <= 0.0 { + if tokens + cache_write_tokens <= 0.0 { return; } let mut conn = match st.db.lock() { @@ -207,15 +224,26 @@ fn settle_usage( input_tokens, cached_tokens, output_tokens, + cache_write_tokens, i2, c2, o2, + p.cache_write_input_per_m, st.cfg.points.points_per_unit, &p.currency, &st.cfg.points.anchor_currency, ); let cost = billing::to_anchor( - billing::raw_cost(input_tokens, cached_tokens, output_tokens, i2, c2, o2), + billing::raw_cost( + input_tokens, + cached_tokens, + output_tokens, + cache_write_tokens, + i2, + c2, + o2, + p.cache_write_input_per_m, + ), &p.currency, &st.cfg.points.anchor_currency, ); @@ -245,6 +273,7 @@ fn settle_usage( tokens, cached_tokens, output_tokens, + cache_write_tokens, pts, cost, }; @@ -408,7 +437,8 @@ async fn forward( if status.is_success() { // 成功：解析 usage（按上游出站协议）→ 计量入账 → 粘性 - let (input_tokens, cached_tokens, output_tokens) = parse_usage(&bytes, outbound); + let (input_tokens, cached_tokens, output_tokens, cache_write_tokens) = + parse_usage(&bytes, outbound); settle_usage( st, auth, @@ -417,6 +447,7 @@ async fn forward( input_tokens, cached_tokens, output_tokens, + cache_write_tokens, ); // 跨协议 → 转换响应体回入站协议 if needs_transform { @@ -679,7 +710,9 @@ async fn forward_stream( ); } let (input, cached, output) = slot.unwrap_or((0.0, 0.0, 0.0)); - settle_usage(&st, auth, &key, &model, input, cached, output); + // 跨协议转换的 usage 由 `sse` 的转换器记录，它只带 (input, cached, output) + // 三元组 ⇒ 缓存写入按 0 计（#432 的范围只到同协议 anthropic 路径）。 + settle_usage(&st, auth, &key, &model, input, cached, output, 0.0); Ok::(axum::body::Bytes::new()) }); (converted, Box::pin(finalize)) @@ -696,10 +729,10 @@ async fn forward_stream( item.map_err(|e| -> std::io::Error { std::io::Error::other(e) }) }); let finalize = futures_util::stream::once(async move { - let (input, cached, output, seen) = { + let (input, cached, output, cache_write, seen) = { let mut cap = capture.lock().expect("usage capture lock"); - let (i, c, o) = cap.finish(); - (i, c, o, cap.usage_seen) + let (i, c, o, cw) = cap.finish(); + (i, c, o, cw, cap.usage_seen) }; if !seen { // 未入账且**非**「上游报了 0」——本次调用没有任何计量依据。 @@ -711,7 +744,7 @@ async fn forward_stream( protocol ); } - settle_usage(&st, auth, &key, &model, input, cached, output); + settle_usage(&st, auth, &key, &model, input, cached, output, cache_write); Ok::(axum::body::Bytes::new()) }); (Box::pin(fwd), Box::pin(finalize)) @@ -738,6 +771,9 @@ struct UsageCapture { tail: Vec, input_tokens: f64, cached_tokens: f64, + /// 缓存写入（cache creation）token 数（#432）。只有 anthropic 在 `message_start` / + /// `message_delta` 的 usage 里带它（`cache_creation_input_tokens`）；其它协议恒 0。 + cache_write_tokens: f64, /// 是否**见到过** usage 对象 /// /// 用于区分「上游报了 0」与「上游什么都没报」——两者在 @@ -755,6 +791,7 @@ impl UsageCapture { tail: Vec::new(), input_tokens: 0.0, cached_tokens: 0.0, + cache_write_tokens: 0.0, usage_seen: false, } } @@ -763,9 +800,10 @@ impl UsageCapture { // anthropic：message_start 事件携带 input_tokens / cache_read_input_tokens //（在流头部，尾部缓冲会丢） if self.protocol == "anthropic" && self.input_tokens == 0.0 { - if let Some((input, cached)) = parse_anthropic_usage(chunk) { + if let Some((input, cached, cache_write)) = parse_anthropic_usage(chunk) { self.input_tokens = input; self.cached_tokens = cached; + self.cache_write_tokens = cache_write; self.usage_seen = true; } } @@ -775,12 +813,13 @@ impl UsageCapture { } } - /// 流尾解析 usage → (input_tokens, cached_tokens, output_tokens) - fn finish(&mut self) -> (f64, f64, f64) { + /// 流尾解析 usage → (input_tokens, cached_tokens, output_tokens, cache_write_tokens) + fn finish(&mut self) -> (f64, f64, f64, f64) { let text = String::from_utf8_lossy(&self.tail); let mut input = self.input_tokens; let mut cached = self.cached_tokens; let mut output = 0.0; + let mut cache_write = self.cache_write_tokens; for line in text.lines() { let Some(data) = line.trim_start().strip_prefix("data:") else { continue; @@ -807,6 +846,10 @@ impl UsageCapture { .get("output_tokens") .and_then(|x| x.as_f64()) .unwrap_or(output); + cache_write = u + .get("cache_creation_input_tokens") + .and_then(|x| x.as_f64()) + .unwrap_or(cache_write); } } else if let Some(u) = v.get("usage") { // openai：最后 chunk 的 usage（stream_options.include_usage 时） @@ -831,12 +874,14 @@ impl UsageCapture { } // input disjoint：input_tokens/prompt_tokens 含缓存命中部分，扣除后避免重复计费 //（同 sse.rs 已修复逻辑，rant 2026-08-23T14:05:02；max 防下溢） - ((input - cached).max(0.0), cached, output) + // 缓存写入**不参与**这里的减法：它是与「未命中」「命中」并列的第三段，不是其中一部分。 + ((input - cached).max(0.0), cached, output, cache_write) } } -/// 从 anthropic 流式 chunk 提取 message_start 的 usage → (input_tokens, cache_read_input_tokens) -fn parse_anthropic_usage(chunk: &[u8]) -> Option<(f64, f64)> { +/// 从 anthropic 流式 chunk 提取 message_start 的 usage → +/// (input_tokens, cache_read_input_tokens, cache_creation_input_tokens) +fn parse_anthropic_usage(chunk: &[u8]) -> Option<(f64, f64, f64)> { let text = String::from_utf8_lossy(chunk); for line in text.lines() { let Some(data) = line.trim_start().strip_prefix("data:") else { @@ -852,7 +897,11 @@ fn parse_anthropic_usage(chunk: &[u8]) -> Option<(f64, f64)> { .pointer("/message/usage/cache_read_input_tokens") .and_then(|t| t.as_f64()) .unwrap_or(0.0); - return Some((input, cached)); + let cache_write = v + .pointer("/message/usage/cache_creation_input_tokens") + .and_then(|t| t.as_f64()) + .unwrap_or(0.0); + return Some((input, cached, cache_write)); } } None @@ -989,10 +1038,14 @@ mod tests { #[test] fn parse_usage_openai_splits_cached_tokens() { let body = br#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_tokens_details":{"cached_tokens":900}}}"#; - let (input, cached, output) = parse_usage(body, "openai_chat"); + let (input, cached, output, cache_write) = parse_usage(body, "openai_chat"); assert_eq!(input, 100.0, "未命中 = prompt - cached"); assert_eq!(cached, 900.0); assert_eq!(output, 50.0); + assert_eq!( + cache_write, 0.0, + "openai 协议没有缓存写入字段 ⇒ 0，不是估算" + ); } #[test] @@ -1000,28 +1053,31 @@ mod tests { // anthropic input_tokens 已不含命中，原样返回 let body = br#"{"usage":{"input_tokens":100,"output_tokens":50,"cache_read_input_tokens":900}}"#; - let (input, cached, output) = parse_usage(body, "anthropic"); + let (input, cached, output, cache_write) = parse_usage(body, "anthropic"); assert_eq!(input, 100.0); assert_eq!(cached, 900.0); assert_eq!(output, 50.0); + assert_eq!(cache_write, 0.0, "上游没报缓存写入 ⇒ 0"); } #[test] fn parse_usage_responses_input_tokens_details() { let body = br#"{"usage":{"input_tokens":1000,"output_tokens":50,"input_tokens_details":{"cached_tokens":700}}}"#; - let (input, cached, output) = parse_usage(body, "responses"); + let (input, cached, output, cache_write) = parse_usage(body, "responses"); assert_eq!(input, 300.0, "未命中 = input_tokens - cached"); assert_eq!(cached, 700.0); assert_eq!(output, 50.0); + assert_eq!(cache_write, 0.0); } #[test] fn parse_usage_missing_cache_defaults_zero() { let body = br#"{"usage":{"prompt_tokens":100,"completion_tokens":10}}"#; - let (input, cached, output) = parse_usage(body, "openai_chat"); + let (input, cached, output, cache_write) = parse_usage(body, "openai_chat"); assert_eq!(input, 100.0); assert_eq!(cached, 0.0); assert_eq!(output, 10.0); + assert_eq!(cache_write, 0.0); } /// rant 2026-08-23T08:20:38 / 2026-08-23T14:05:02：DeepSeek 原生顶层缓存拼写 @@ -1031,21 +1087,21 @@ mod tests { fn parse_usage_openai_deepseek_native_cache_spelling() { // DeepSeek 原生顶层拼写：prompt_tokens 含命中，仍需减除 let body = br#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_cache_hit_tokens":900}}"#; - let (input, cached, output) = parse_usage(body, "openai_chat"); + let (input, cached, output, _) = parse_usage(body, "openai_chat"); assert_eq!(input, 100.0, "未命中 = prompt_tokens - 命中"); assert_eq!(cached, 900.0, "DeepSeek 原生拼写必须被识别"); assert_eq!(output, 50.0); // 阳性对照：OpenAI 拼写（`prompt_tokens_details.cached_tokens`）不受影响 let body = br#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_tokens_details":{"cached_tokens":900}}}"#; - let (input, cached, output) = parse_usage(body, "openai_chat"); + let (input, cached, output, _) = parse_usage(body, "openai_chat"); assert_eq!(input, 100.0); assert_eq!(cached, 900.0); assert_eq!(output, 50.0); // 两拼写并存 → 与流式同优先级：DeepSeek 原生优先 let body = br#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_cache_hit_tokens":900,"prompt_tokens_details":{"cached_tokens":800}}}"#; - let (input, cached, output) = parse_usage(body, "openai_chat"); + let (input, cached, output, _) = parse_usage(body, "openai_chat"); assert_eq!(input, 100.0, "未命中 = 1000 - 900"); assert_eq!( cached, 900.0, @@ -1059,21 +1115,23 @@ mod tests { /// C2039 修的是非流式的缓存拼写分支 —— 这条断言把两者钉在一起，防止再次分叉。 #[test] fn usage_parsers_agree_on_cache_spelling() { - let cases: [(&str, (f64, f64, f64)); 3] = [ + // 四元组（输入/命中/输出/缓存写入）写在类型里读起来比再立一个结构体清楚 + #[allow(clippy::type_complexity)] + let cases: [(&str, (f64, f64, f64, f64)); 3] = [ // DeepSeek 原生顶层拼写（此前非流式漏认） ( r#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_cache_hit_tokens":900}}"#, - (100.0, 900.0, 50.0), + (100.0, 900.0, 50.0, 0.0), ), // OpenAI 拼写 ( r#"{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_tokens_details":{"cached_tokens":900}}}"#, - (100.0, 900.0, 50.0), + (100.0, 900.0, 50.0, 0.0), ), // 无任何缓存字段 → 全部按未命中（缺数据即 0，并非 C2037 的「无 usage」） ( r#"{"usage":{"prompt_tokens":1000,"completion_tokens":50}}"#, - (1000.0, 0.0, 50.0), + (1000.0, 0.0, 50.0, 0.0), ), ]; for (body, expected) in cases { @@ -1129,11 +1187,20 @@ mod tests { }; // ① 200000（边界−1）⇒ 第一档 $2/M；② 200001（边界）⇒ 长档 $4/M（恰 2 倍） - settle_usage(&st, auth(), &key, "grok-4.6", 200_000.0, 0.0, 0.0); - settle_usage(&st, auth(), &key, "grok-4.6", 200_001.0, 0.0, 0.0); + settle_usage(&st, auth(), &key, "grok-4.6", 200_000.0, 0.0, 0.0, 0.0); + settle_usage(&st, auth(), &key, "grok-4.6", 200_001.0, 0.0, 0.0, 0.0); // ③ 选档用的是**整段 prompt**：未命中 100000 ＋ 命中 100001 = 200001 ⇒ 仍走长档 // （若错看未命中那一段，就会落回第一档） - settle_usage(&st, auth(), &key, "grok-4.6", 100_000.0, 100_001.0, 0.0); + settle_usage( + &st, + auth(), + &key, + "grok-4.6", + 100_000.0, + 100_001.0, + 0.0, + 0.0, + ); let pts: Vec = { let conn = st.db.lock().unwrap(); @@ -1164,6 +1231,141 @@ mod tests { ); } + /// #432：缓存写入（cache creation）token 数在**两条解析路径**上都要取到， + /// 且只从 anthropic 的 usage 里取（其它协议没有这个字段 ⇒ 0）。 + #[test] + fn both_usage_parsers_read_anthropic_cache_creation_tokens() { + let usage = r#"{"usage":{"input_tokens":100,"output_tokens":50,"cache_read_input_tokens":900,"cache_creation_input_tokens":400}}"#; + + // 非流式：`input_tokens` 不含缓存写入（Anthropic 三段分开报）⇒ 原样取出，不做加减 + let (input, cached, output, cache_write) = parse_usage(usage.as_bytes(), "anthropic"); + assert_eq!((input, cached, output), (100.0, 900.0, 50.0)); + assert_eq!(cache_write, 400.0, "缓存写入必须原样取出"); + + // 流式：同一条 usage 走 `UsageCapture`（message_delta 形状） + let mut cap = UsageCapture::new("anthropic"); + cap.push(format!("data: {usage}\n\n").as_bytes()); + let (s_input, s_cached, s_output, s_cache_write) = cap.finish(); + assert_eq!( + (s_cached, s_output), + (900.0, 50.0), + "流式与非流式看同一段 usage" + ); + assert_eq!(s_cache_write, 400.0, "流式路径同样要取到缓存写入"); + // ⚠️ 预存不对称（**独立问题，本 PR 不改**）：流式对 anthropic 也套用了 + // `(input - cached)` 这条「openai 的 prompt_tokens 含命中」修正，而 anthropic 的 + // `input_tokens` 本就不含命中 ⇒ 流式把它减成 0、非流式保留 100。修它会抬高存量 + // 账单，需要单独授权，故这里只钉住**当前**行为，不假装两条路径一致。 + assert_eq!((s_input, input), (0.0, 100.0), "当前行为：流式少算这一段"); + + // 其它协议：字段不存在 ⇒ 0（不是估算） + let openai = r#"{"usage":{"prompt_tokens":1000,"completion_tokens":50}}"#; + assert_eq!(parse_usage(openai.as_bytes(), "openai_chat").3, 0.0); + assert_eq!(parse_usage(openai.as_bytes(), "responses").3, 0.0); + } + + /// #432：缓存写入价接进结算。两件承重的事： + /// ① 缺省 0 价 ⇒ 带写入 token 的调用与不带写入的**同一调用**点数逐位相同（存量账不动）； + /// ② 配了价 ⇒ 收费额 = 配置价 × 写入 token 数，且两个账本列都记下写入的 token 数。 + #[test] + fn settle_bills_cache_write_tokens_at_the_configured_price() { + let st = test_state_eps( + "cache_write_settle", + "xai-paygo", + &["openai_chat"], + "http://127.0.0.1:1", + ); + { + let conn = st.db.lock().unwrap(); + conn.execute( + "INSERT INTO keys (id, provider, plan, model, status, owner_id, encrypted_key, quota, used) \ + VALUES (901, 'xai', 'xai-paygo', 'grok-4.6', 'on', 1, 'v1:x', 1000, 0)", + [], + ) + .unwrap(); + } + let key = dao::KeyRow { + id: 901, + provider: "xai".to_string(), + plan: "xai-paygo".to_string(), + owner_id: 1, + encrypted_key: String::new(), + }; + let auth = || AuthUser { + user_id: 1, + api_key_id: 1, + role: "user".to_string(), + }; + + // ① 缺省价 0：同一笔输入，带 1000 缓存写入 vs 不带 ⇒ 点数必须逐位相同 + settle_usage(&st, auth(), &key, "grok-4.6", 1000.0, 0.0, 0.0, 0.0); + settle_usage(&st, auth(), &key, "grok-4.6", 1000.0, 0.0, 0.0, 1000.0); + + // ② 配置缓存写入价 $2/M（grok-4.6 是 USD 计价、CNY 锚定 1 点/元） + { + let conn = st.db.lock().unwrap(); + conn.execute( + "UPDATE models SET cache_write_input_per_m = 2.0 WHERE provider = 'xai' AND model = 'grok-4.6'", + [], + ) + .unwrap(); + } + // 只有缓存写入的响应（输入/命中/输出全 0）：不该被「无计量依据」的早退吞掉 + settle_usage(&st, auth(), &key, "grok-4.6", 0.0, 0.0, 0.0, 1_000_000.0); + + let rows: Vec<(f64, f64, f64)> = { + let conn = st.db.lock().unwrap(); + let mut stmt = conn + .prepare( + "SELECT pts, cache_write_tokens, tokens FROM transactions \ + WHERE type = 'consume' ORDER BY id ASC", + ) + .unwrap(); + stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) + .unwrap() + .map(|x| x.unwrap()) + .collect() + }; + assert_eq!(rows.len(), 3, "三笔调用各一条 consume"); + assert_eq!( + rows[0].0, rows[1].0, + "缺省 0 价 ⇒ 带不带缓存写入，点数完全相同（存量账不动）: {} vs {}", + rows[0].0, rows[1].0 + ); + assert_eq!( + rows[1].1, 1000.0, + "缓存写入 token 数要落进 transactions 列（未配置价时也照记，供对账）" + ); + // 1M × $2/M = 2.0 USD × 7.2 = 14.4 点 + assert!( + (rows[2].0 - 14.4).abs() < 1e-6, + "配价后按配置价收费: {}", + rows[2].0 + ); + assert_eq!(rows[2].1, 1_000_000.0, "写入 token 数落库"); + assert_eq!( + rows[2].2, 0.0, + "`tokens` 语义不变：仍是 input + cached + output，缓存写入单列" + ); + + // usage_records 同口径 + let usage_cols: (f64, f64) = { + let conn = st.db.lock().unwrap(); + conn.query_row( + "SELECT cache_write_tokens, cost FROM usage_records ORDER BY id DESC LIMIT 1", + [], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap() + }; + assert_eq!(usage_cols.0, 1_000_000.0, "usage_records 也记写入 token 数"); + assert!( + (usage_cols.1 - 14.4).abs() < 1e-9, + "usage_records.cost 是**锚定货币**金额：2.0 USD × 7.2 = 14.4 CNY: {}", + usage_cols.1 + ); + } + fn test_state_eps(tag: &str, plan_id: &str, protocols: &[&str], base_url: &str) -> AppState { let p = std::env::temp_dir().join(format!("atp_gw_{}_{}.db", std::process::id(), tag)); let _ = std::fs::remove_file(&p); @@ -2705,7 +2907,7 @@ mod tests { cap.push(br#"data: {"usage":{"prompt_tokens":10,"completion_tokens":5},"choices":[]}"#); cap.push(b"\n\n"); cap.push(br#"data: [DONE]"#); - let (i, _c, o) = cap.finish(); + let (i, _c, o, _cw) = cap.finish(); assert_eq!(i, 10.0); assert_eq!(o, 5.0); @@ -2714,7 +2916,7 @@ mod tests { cap.push( br#"data: {"usage":{"prompt_tokens":100,"prompt_cache_hit_tokens":90,"completion_tokens":50},"choices":[]}"#, ); - let (i, c, o) = cap.finish(); + let (i, c, o, _cw) = cap.finish(); assert_eq!(i, 10.0, "DeepSeek cached 应从 prompt 扣除（disjoint）"); assert_eq!(c, 90.0); assert_eq!(o, 50.0); @@ -2724,7 +2926,7 @@ mod tests { cap.push( br#"data: {"usage":{"prompt_tokens":100,"prompt_tokens_details":{"cached_tokens":90},"completion_tokens":50},"choices":[]}"#, ); - let (i, c, o) = cap.finish(); + let (i, c, o, _cw) = cap.finish(); assert_eq!(i, 10.0); assert_eq!(c, 90.0); assert_eq!(o, 50.0); @@ -2740,7 +2942,7 @@ data: {"type":"message_start","message":{"usage":{"input_tokens":80,"output_toke br#"event: message_delta data: {"type":"message_delta","usage":{"output_tokens":30}}"#, ); - let (i, _c, o) = cap.finish(); + let (i, _c, o, _cw) = cap.finish(); assert_eq!(i, 80.0, "message_start 的 input_tokens 提前捕获"); assert_eq!(o, 30.0, "message_delta 的 output_tokens 流尾解析"); @@ -2755,7 +2957,7 @@ data: {"type":"message_start","message":{"usage":{"input_tokens":80,"cache_read_ br#"event: message_delta data: {"type":"message_delta","usage":{"output_tokens":30}}"#, ); - let (i, c, o) = cap.finish(); + let (i, c, o, _cw) = cap.finish(); assert_eq!(i, 10.0, "cache_read 应从 input 扣除（disjoint）"); assert_eq!(c, 70.0); assert_eq!(o, 30.0); diff --git a/src/routes/mod.rs b/src/routes/mod.rs index 36f2148..818f30b 100644 --- a/src/routes/mod.rs +++ b/src/routes/mod.rs @@ -2861,6 +2861,7 @@ mod tests { tokens: 1000.0, cached_tokens: 0.0, output_tokens: 0.0, + cache_write_tokens: 0.0, pts, cost: 0.0, }; From 0539211ada8a7b07a42f9f7b7474dd33932e878f Mon Sep 17 00:00:00 2001 From: argszero Date: Sat, 24 Oct 2026 06:44:48 +0800 Subject: [PATCH 2/2] test(billing): pin both sides of the settlement guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The change widened settlement's entry guard to include the fourth count, so a response reporting only cache-write tokens is no longer dropped as "no data to settle". That widening has a second face, and it is the one that would be lost silently: a response reporting nothing at all must still take the old early-return path, or "unmetered" becomes "billed zero" and the distinction the surrounding comments insist on is gone. Both sides are now asserted on ledger values rather than on the function returning: * absent field at the default price — the points and the recorded cost of a call carrying cache-write tokens are bit-identical to the same call without them, and both equal the pre-change arithmetic (1000 * $2/M = 0.002 USD * 7.2 = 0.0144 points), so the equality is between two real settlements and not two early returns. * nothing reported — no transaction row and no balance movement. The ledger query reads the two tables separately: one settlement writes two transactions (consume + earn) but one usage_records row, so joining them on id pairs rows that do not belong together. Refs #432 --- src/gateway.rs | 89 +++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 84 insertions(+), 5 deletions(-) diff --git a/src/gateway.rs b/src/gateway.rs index 9d2e5d3..b921eaf 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -1297,10 +1297,90 @@ mod tests { role: "user".to_string(), }; - // ① 缺省价 0：同一笔输入，带 1000 缓存写入 vs 不带 ⇒ 点数必须逐位相同 + // ① 缺省价 0：同一笔输入，带 1000 缓存写入 vs 不带 ⇒ 账本上的点数与成本逐位相同。 + // 这不是「函数没 panic」，而是**结算真的发生了**：两笔都必须落账，且金额等于 + // #432 之前的算式（1000 × $2/M = 0.002 USD × 7.2 = 0.0144 点）。 settle_usage(&st, auth(), &key, "grok-4.6", 1000.0, 0.0, 0.0, 0.0); settle_usage(&st, auth(), &key, "grok-4.6", 1000.0, 0.0, 0.0, 1000.0); + // 两张表分开取：一次结算写 2 条 transactions（consume + earn）但只写 1 条 + // usage_records，按 id 对齐会错位 —— 各自按写入顺序排。 + let (consume_pts, usage_costs): (Vec, Vec) = { + let conn = st.db.lock().unwrap(); + let mut stmt = conn + .prepare("SELECT pts FROM transactions WHERE type = 'consume' ORDER BY id ASC") + .unwrap(); + let pts: Vec = stmt + .query_map([], |r| r.get(0)) + .unwrap() + .map(|x| x.unwrap()) + .collect(); + let mut stmt2 = conn + .prepare("SELECT cost FROM usage_records ORDER BY id ASC") + .unwrap(); + let costs: Vec = stmt2 + .query_map([], |r| r.get(0)) + .unwrap() + .map(|x| x.unwrap()) + .collect(); + (pts, costs) + }; + let ledger_after_two: Vec<(f64, f64)> = consume_pts + .iter() + .copied() + .zip(usage_costs.iter().copied()) + .collect(); + assert_eq!( + (consume_pts.len(), usage_costs.len()), + (2, 2), + "两笔都必须落账（不是被早退吞掉）" + ); + assert!( + (ledger_after_two[0].0 - 0.0144).abs() < 1e-9, + "不带缓存写入的点数 = 接入前的算式: {}", + ledger_after_two[0].0 + ); + assert_eq!( + ledger_after_two[0].0, ledger_after_two[1].0, + "缺省 0 价 ⇒ 带不带缓存写入，点数逐位相同（存量账不动）: {} vs {}", + ledger_after_two[0].0, ledger_after_two[1].0 + ); + assert_eq!( + ledger_after_two[0].1, ledger_after_two[1].1, + "缺省 0 价 ⇒ usage_records.cost 也逐位相同: {} vs {}", + ledger_after_two[0].1, ledger_after_two[1].1 + ); + + // ①-b 早退的**反面**：四段全 0（上游什么都没报）仍走老路 —— 不入账。 + // 否则「无法计量」会被悄悄变成「记 0 元」，把 R156 那条「缺数据 ≠ 数据为 0」 + // 的区分抹平。判据看账本行数与余额，不看函数是否返回。 + let before_blank = { + let conn = st.db.lock().unwrap(); + let n: i64 = conn + .query_row("SELECT COUNT(*) FROM transactions", [], |r| r.get(0)) + .unwrap(); + let bal: f64 = conn + .query_row("SELECT balance FROM quotas WHERE user_id = 1", [], |r| { + r.get(0) + }) + .unwrap(); + (n, bal) + }; + settle_usage(&st, auth(), &key, "grok-4.6", 0.0, 0.0, 0.0, 0.0); + { + let conn = st.db.lock().unwrap(); + let n: i64 = conn + .query_row("SELECT COUNT(*) FROM transactions", [], |r| r.get(0)) + .unwrap(); + let bal: f64 = conn + .query_row("SELECT balance FROM quotas WHERE user_id = 1", [], |r| { + r.get(0) + }) + .unwrap(); + assert_eq!(n, before_blank.0, "四段全 0 ⇒ 不写任何账本行"); + assert_eq!(bal, before_blank.1, "四段全 0 ⇒ 余额分毫不动"); + } + // ② 配置缓存写入价 $2/M（grok-4.6 是 USD 计价、CNY 锚定 1 点/元） { let conn = st.db.lock().unwrap(); @@ -1326,11 +1406,10 @@ mod tests { .map(|x| x.unwrap()) .collect() }; - assert_eq!(rows.len(), 3, "三笔调用各一条 consume"); assert_eq!( - rows[0].0, rows[1].0, - "缺省 0 价 ⇒ 带不带缓存写入，点数完全相同（存量账不动）: {} vs {}", - rows[0].0, rows[1].0 + rows.len(), + 3, + "三笔落账（两笔缺省价 ＋ 一笔配价），全 0 那笔不入账" ); assert_eq!( rows[1].1, 1000.0,