Fix usage token cache accounting

Author: k0ngk0ng

AGENTS.md

@@ -0,0 +1,13 @@
+当前项目是基于 CLIProxyAPI(简称cpa) 做的 usage 统计程序 cpa-usage
+
+CLIProxyAPI 代码在 https://github.com/router-for-me/CLIProxyAPI
+已经同步到目录：/Users/jchen/Sources/github.com/router-for-me/CLIProxyAPI
+
+该项目目的是能更好的观测 API 的使用情况
+
+2026-06-18 usage token 语义核查：
+- 以本地 CLIProxyAPI `origin/main` / `v7.2.16` 为准，队列中的 `tokens.input_tokens` 是 CPA 原样转发的上游 input/prompt 计数，不应全局理解为 `new/uncached`。
+- OpenAI/Codex/OpenAI-compatible/Gemini 风格：`input_tokens`/`promptTokenCount` 通常已经是总输入，`cached_tokens`/`cachedContentTokenCount` 是其中的缓存命中部分；新输入应按 `input_tokens - cached_tokens` 估算，UI 不应再把二者相加为 Input。
+- Claude/Anthropic 风格：`input_tokens` 是最后 cache breakpoint 之后的未缓存输入，总输入应为 `input_tokens + cache_read_tokens + cache_creation_tokens`。
+- CPA 的 `cached_tokens` 不是可靠的“cache read”字段：Claude 解析在 read 为 0 时会 fallback 到 cache creation；有 `cache_read_tokens/cache_creation_tokens` 时优先使用这两个拆分字段。
+- 当前修复策略：ingest 入口把新数据规范化为 `input_tokens=NEW`、`cached_tokens/cache_read_tokens=CACHE READ`、`cache_creation_tokens=CACHE WRITE`；历史数据用 `scripts/fix_usage_token_history.sh` 做同口径修复，脚本会先备份命中行。

internal/ingest/decoder.go

@@ -30,6 +30,7 @@ func Decode(message string) (storage.UsageEvent, error) {
 	if ts.IsZero() {
 		ts = time.Now().UTC()
 	}
+	tokens := normalizeUsageTokens(rec)
 	return storage.UsageEvent{
 		EventKey:            requestID,
 		Timestamp:           ts.UTC(),
@@ -45,13 +46,13 @@ func Decode(message string) (storage.UsageEvent, error) {
 		RequestID:           requestID,
 		LatencyMs:           rec.LatencyMs,
 		TTFTMs:              rec.TTFTMs,
-		InputTokens:         rec.Tokens.InputTokens,
-		OutputTokens:        rec.Tokens.OutputTokens,
-		ReasoningTokens:     rec.Tokens.ReasoningTokens,
-		CachedTokens:        rec.Tokens.CachedTokens,
-		CacheReadTokens:     rec.Tokens.CacheReadTokens,
-		CacheCreationTokens: rec.Tokens.CacheCreationTokens,
-		TotalTokens:         rec.Tokens.TotalTokens,
+		InputTokens:         tokens.InputTokens,
+		OutputTokens:        tokens.OutputTokens,
+		ReasoningTokens:     tokens.ReasoningTokens,
+		CachedTokens:        tokens.CachedTokens,
+		CacheReadTokens:     tokens.CacheReadTokens,
+		CacheCreationTokens: tokens.CacheCreationTokens,
+		TotalTokens:         tokens.TotalTokens,
 		Failed:              rec.Failed,
 		FailStatusCode:      rec.Fail.StatusCode,
 		FailBody:            strings.TrimSpace(rec.Fail.Body),
@@ -94,6 +95,49 @@ func resolveAPIGroupKey(rec cpa.UsageRecord) string {
 	return "unknown"
 }
 
+func normalizeUsageTokens(rec cpa.UsageRecord) cpa.UsageTokens {
+	tokens := rec.Tokens
+	if isClaudeStyleUsage(rec) {
+		if tokens.CacheReadTokens != 0 || tokens.CacheCreationTokens != 0 {
+			tokens.CachedTokens = tokens.CacheReadTokens
+		}
+		return tokens
+	}
+
+	cacheRead := tokens.CacheReadTokens
+	if cacheRead == 0 {
+		cacheRead = tokens.CachedTokens
+	}
+	cacheWrite := tokens.CacheCreationTokens
+	cachedInput := cacheRead + cacheWrite
+	if cachedInput > 0 {
+		tokens.InputTokens = subtractFloor(tokens.InputTokens, cachedInput)
+		tokens.CachedTokens = cacheRead
+		tokens.CacheReadTokens = cacheRead
+	}
+	return tokens
+}
+
+func isClaudeStyleUsage(rec cpa.UsageRecord) bool {
+	for _, value := range []string{rec.Provider, rec.Model, rec.Endpoint} {
+		v := strings.ToLower(strings.TrimSpace(value))
+		if strings.Contains(v, "claude") || strings.Contains(v, "anthropic") {
+			return true
+		}
+	}
+	return false
+}
+
+func subtractFloor(value, delta int64) int64 {
+	if delta <= 0 {
+		return value
+	}
+	if value <= delta {
+		return 0
+	}
+	return value - delta
+}
+
 func compactRawJSON(raw json.RawMessage) string {
 	raw = bytes.TrimSpace(raw)
 	if len(raw) == 0 || bytes.Equal(raw, []byte("null")) || !json.Valid(raw) {

internal/ingest/decoder_test.go

@@ -21,9 +21,9 @@ func TestDecodeNewUsageQueueFields(t *testing.T) {
 		"failed":true,
 		"fail":{"status_code":429,"body":" rate limited "},
 		"response_headers":{"Retry-After":["30"],"X-Upstream-Request-Id":["upstream-req-1"]},
-		"provider":"openai",
-		"model":"gpt-5.4",
-		"alias":"client-gpt",
+		"provider":"claude",
+		"model":"claude-sonnet-4",
+		"alias":"client-claude",
 		"endpoint":"POST /v1/chat/completions",
 		"auth_type":"apikey",
 		"api_key":"test-key",
@@ -39,7 +39,7 @@ func TestDecodeNewUsageQueueFields(t *testing.T) {
 	if ev.EventKey != "ctx-request-id" || ev.RequestID != "ctx-request-id" {
 		t.Fatalf("request ids = event_key %q request_id %q", ev.EventKey, ev.RequestID)
 	}
-	if ev.Alias != "client-gpt" || ev.TTFTMs != 320 {
+	if ev.Alias != "client-claude" || ev.TTFTMs != 320 {
 		t.Fatalf("alias/ttft = %q/%d", ev.Alias, ev.TTFTMs)
 	}
 	if ev.CacheReadTokens != 4 || ev.CacheCreationTokens != 5 {
@@ -55,3 +55,63 @@ func TestDecodeNewUsageQueueFields(t *testing.T) {
 		t.Fatalf("reasoning/service tier = %q/%q", ev.ReasoningEffort, ev.ServiceTier)
 	}
 }
+
+func TestDecodeNormalizesTotalInputStyleCacheTokens(t *testing.T) {
+	raw := `{
+		"timestamp":"2026-04-25T00:00:00Z",
+		"provider":"openai",
+		"model":"gpt-5",
+		"request_id":"req-openai-cache",
+		"tokens":{
+			"input_tokens":1000,
+			"output_tokens":50,
+			"cached_tokens":900,
+			"total_tokens":1050
+		}
+	}`
+
+	ev, err := Decode(raw)
+	if err != nil {
+		t.Fatalf("Decode: %v", err)
+	}
+	if ev.InputTokens != 100 {
+		t.Fatalf("input/new tokens = %d, want 100", ev.InputTokens)
+	}
+	if ev.CachedTokens != 900 || ev.CacheReadTokens != 900 || ev.CacheCreationTokens != 0 {
+		t.Fatalf("cache split = cached %d read %d write %d, want 900/900/0", ev.CachedTokens, ev.CacheReadTokens, ev.CacheCreationTokens)
+	}
+	if ev.TotalTokens != 1050 {
+		t.Fatalf("total tokens = %d, want preserved 1050", ev.TotalTokens)
+	}
+}
+
+func TestDecodePreservesClaudeInputAndUsesExplicitCacheRead(t *testing.T) {
+	raw := `{
+		"timestamp":"2026-04-25T00:00:00Z",
+		"provider":"claude",
+		"model":"claude-sonnet-4",
+		"request_id":"req-claude-cache",
+		"tokens":{
+			"input_tokens":100,
+			"output_tokens":50,
+			"cached_tokens":900,
+			"cache_read_tokens":0,
+			"cache_creation_tokens":900,
+			"total_tokens":1050
+		}
+	}`
+
+	ev, err := Decode(raw)
+	if err != nil {
+		t.Fatalf("Decode: %v", err)
+	}
+	if ev.InputTokens != 100 {
+		t.Fatalf("input/new tokens = %d, want 100", ev.InputTokens)
+	}
+	if ev.CachedTokens != 0 || ev.CacheReadTokens != 0 || ev.CacheCreationTokens != 900 {
+		t.Fatalf("cache split = cached %d read %d write %d, want 0/0/900", ev.CachedTokens, ev.CacheReadTokens, ev.CacheCreationTokens)
+	}
+	if ev.TotalTokens != 1050 {
+		t.Fatalf("total tokens = %d, want preserved 1050", ev.TotalTokens)
+	}
+}