Merge pull request #6137 from yan9651688/codex/fix-cn-anthropic-usage-billing

fix(billing): normalize CN Anthropic usage tokens

backend/internal/pkg/apicompat/types.go

@@ -155,12 +155,26 @@ func AnthropicStopReasonString(p *string) string {
 	return *p
 }
 
+// AnthropicPromptTokensDetails holds OpenAI-compatible prompt token details
+// occasionally included by Anthropic-compatible providers.
+type AnthropicPromptTokensDetails struct {
+	CachedTokens int `json:"cached_tokens,omitempty"`
+}
+
 // AnthropicUsage holds token counts in Anthropic format.
 type AnthropicUsage struct {
 	InputTokens              int `json:"input_tokens"`
 	OutputTokens             int `json:"output_tokens"`
 	CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
 	CacheReadInputTokens     int `json:"cache_read_input_tokens"`
+	// Anthropic-compatible providers can also expose their native OpenAI-style
+	// total/cache fields. Preserve them so callers can normalize provider totals
+	// into Anthropic's mutually-exclusive billing buckets.
+	PromptTokens          int                           `json:"prompt_tokens,omitempty"`
+	CachedTokens          int                           `json:"cached_tokens,omitempty"`
+	PromptTokensDetails   *AnthropicPromptTokensDetails `json:"prompt_tokens_details,omitempty"`
+	PromptCacheHitTokens  *int                          `json:"prompt_cache_hit_tokens,omitempty"`
+	PromptCacheMissTokens *int                          `json:"prompt_cache_miss_tokens,omitempty"`
 }
 
 // ---------------------------------------------------------------------------

backend/internal/service/gateway_anthropic_passthrough.go

@@ -698,6 +698,72 @@ func parseSSEUsagePassthrough(data string, usage *ClaudeUsage) {
 			usage.CacheCreationInputTokens = int(total)
 		}
 	}
+
+	// Kimi's Anthropic-compatible stream uses input_tokens with two meanings:
+	// message_start reports total prompt input, while message_delta reports only
+	// uncached input. prompt_tokens remains the total in both events. Normalize
+	// to ClaudeUsage's mutually-exclusive buckets so downstream billing does not
+	// subtract cache tokens from an already-uncached value.
+	usageNode := parsed.Get("usage")
+	if parsed.Get("type").String() == "message_start" {
+		usageNode = parsed.Get("message.usage")
+	}
+	normalizeAnthropicCompatiblePromptUsage(usageNode, usage)
+}
+
+// normalizeAnthropicCompatiblePromptUsage converts provider-native OpenAI-style
+// prompt/cache fields into Claude's mutually-exclusive usage buckets. Native
+// Anthropic responses do not expose these aliases and are left alone.
+func normalizeAnthropicCompatiblePromptUsage(usageNode gjson.Result, usage *ClaudeUsage) bool {
+	if usage == nil || !usageNode.Exists() {
+		return false
+	}
+	promptTokens := usageNode.Get("prompt_tokens")
+	promptCacheHitTokens := usageNode.Get("prompt_cache_hit_tokens")
+	promptCacheMissTokens := usageNode.Get("prompt_cache_miss_tokens")
+	if (!promptTokens.Exists() || promptTokens.Int() <= 0) &&
+		!promptCacheHitTokens.Exists() && !promptCacheMissTokens.Exists() {
+		return false
+	}
+
+	cacheReadTokens := usage.CacheReadInputTokens
+	if v := usageNode.Get("cache_read_input_tokens"); v.Exists() {
+		cacheReadTokens = int(v.Int())
+	}
+	if cacheReadTokens == 0 {
+		if v := usageNode.Get("cached_tokens"); v.Exists() {
+			cacheReadTokens = int(v.Int())
+		}
+	}
+	if cacheReadTokens == 0 {
+		if v := usageNode.Get("prompt_tokens_details.cached_tokens"); v.Exists() {
+			cacheReadTokens = int(v.Int())
+		}
+	}
+	if cacheReadTokens == 0 && promptCacheHitTokens.Exists() {
+		cacheReadTokens = max(int(promptCacheHitTokens.Int()), 0)
+	}
+
+	cacheCreationTokens := usage.CacheCreationInputTokens
+	if v := usageNode.Get("cache_creation_input_tokens"); v.Exists() {
+		cacheCreationTokens = int(v.Int())
+	}
+	if cacheCreationTokens == 0 {
+		cc5m := usageNode.Get("cache_creation.ephemeral_5m_input_tokens").Int()
+		cc1h := usageNode.Get("cache_creation.ephemeral_1h_input_tokens").Int()
+		if cc5m > 0 || cc1h > 0 {
+			cacheCreationTokens = int(cc5m + cc1h)
+		}
+	}
+
+	if promptCacheMissTokens.Exists() {
+		usage.InputTokens = max(int(promptCacheMissTokens.Int()), 0)
+	} else {
+		usage.InputTokens = max(int(promptTokens.Int())-cacheReadTokens-cacheCreationTokens, 0)
+	}
+	usage.CacheReadInputTokens = cacheReadTokens
+	usage.CacheCreationInputTokens = cacheCreationTokens
+	return true
 }
 
 func parseClaudeUsageFromResponseBody(body []byte) *ClaudeUsage {
@@ -731,6 +797,7 @@ func parseClaudeUsageFromResponseBody(body []byte) *ClaudeUsage {
 			usage.CacheReadInputTokens = int(cached)
 		}
 	}
+	normalizeAnthropicCompatiblePromptUsage(usageNode, usage)
 	return usage
 }
 

backend/internal/service/gateway_forward_as_responses.go

@@ -285,18 +285,46 @@ func mergeAnthropicUsage(dst *ClaudeUsage, src apicompat.AnthropicUsage) {
 	if dst == nil {
 		return
 	}
-	if src.InputTokens > 0 {
-		dst.InputTokens = src.InputTokens
+
+	// Some Anthropic-compatible providers retain OpenAI-style prompt/cache
+	// fields. Prefer those authoritative totals or hit/miss buckets over the
+	// overloaded input_tokens field. This covers Kimi's changing stream
+	// semantics as well as GLM/DeepSeek cache aliases.
+	if src.PromptTokens > 0 || src.PromptCacheHitTokens != nil || src.PromptCacheMissTokens != nil {
+		cacheReadTokens := src.CacheReadInputTokens
+		if cacheReadTokens == 0 && src.CachedTokens > 0 {
+			cacheReadTokens = src.CachedTokens
+		}
+		if cacheReadTokens == 0 && src.PromptTokensDetails != nil && src.PromptTokensDetails.CachedTokens > 0 {
+			cacheReadTokens = src.PromptTokensDetails.CachedTokens
+		}
+		if cacheReadTokens == 0 && src.PromptCacheHitTokens != nil {
+			cacheReadTokens = max(*src.PromptCacheHitTokens, 0)
+		}
+
+		if src.PromptCacheMissTokens != nil {
+			dst.InputTokens = max(*src.PromptCacheMissTokens, 0)
+		} else {
+			dst.InputTokens = max(src.PromptTokens-cacheReadTokens-src.CacheCreationInputTokens, 0)
+		}
+		dst.CacheReadInputTokens = cacheReadTokens
+		dst.CacheCreationInputTokens = src.CacheCreationInputTokens
+	} else {
+		if src.InputTokens > 0 {
+			dst.InputTokens = src.InputTokens
+		}
+		if src.CacheReadInputTokens > 0 {
+			dst.CacheReadInputTokens = src.CacheReadInputTokens
+		} else if src.CachedTokens > 0 {
+			dst.CacheReadInputTokens = src.CachedTokens
+		}
+		if src.CacheCreationInputTokens > 0 {
+			dst.CacheCreationInputTokens = src.CacheCreationInputTokens
+		}
 	}
 	if src.OutputTokens > 0 {
 		dst.OutputTokens = src.OutputTokens
 	}
-	if src.CacheReadInputTokens > 0 {
-		dst.CacheReadInputTokens = src.CacheReadInputTokens
-	}
-	if src.CacheCreationInputTokens > 0 {
-		dst.CacheCreationInputTokens = src.CacheCreationInputTokens
-	}
 }
 
 // parseAnthropicSSEField parses an SSE field line in the form "field:value" or "field: value".

backend/internal/service/kimi_anthropic_usage_test.go

@@ -0,0 +1,248 @@
+package service
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseSSEUsagePassthroughNormalizesKimiPromptUsage(t *testing.T) {
+	usage := &ClaudeUsage{}
+
+	parseSSEUsagePassthrough(`{"type":"message_start","message":{"usage":{"input_tokens":173306,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":0,"prompt_tokens":173306,"cached_tokens":0}}}`, usage)
+	require.Equal(t, 173306, usage.InputTokens)
+	require.Zero(t, usage.CacheReadInputTokens)
+
+	parseSSEUsagePassthrough(`{"type":"message_delta","usage":{"input_tokens":250,"cache_creation_input_tokens":0,"cache_read_input_tokens":173056,"output_tokens":166,"prompt_tokens":173306,"cached_tokens":173056}}`, usage)
+	require.Equal(t, 250, usage.InputTokens, "Kimi message_delta input_tokens is already the uncached bucket")
+	require.Equal(t, 173056, usage.CacheReadInputTokens)
+	require.Equal(t, 166, usage.OutputTokens)
+}
+
+func TestParseSSEUsagePassthroughKimiFullyCachedInputReplacesStartTotal(t *testing.T) {
+	usage := &ClaudeUsage{}
+
+	parseSSEUsagePassthrough(`{"type":"message_start","message":{"usage":{"input_tokens":173306,"prompt_tokens":173306}}}`, usage)
+	parseSSEUsagePassthrough(`{"type":"message_delta","usage":{"input_tokens":0,"cache_read_input_tokens":173306,"output_tokens":8,"prompt_tokens":173306,"cached_tokens":173306}}`, usage)
+
+	require.Zero(t, usage.InputTokens, "an explicit zero uncached bucket must not retain message_start's total")
+	require.Equal(t, 173306, usage.CacheReadInputTokens)
+}
+
+func TestParseClaudeUsageFromResponseBodyNormalizesCNProviderAliases(t *testing.T) {
+	tests := []struct {
+		name          string
+		body          string
+		wantInput     int
+		wantCacheRead int
+		wantOutput    int
+	}{
+		{
+			name:          "Kimi top-level cached_tokens",
+			body:          `{"usage":{"input_tokens":173306,"output_tokens":166,"cache_read_input_tokens":173056,"prompt_tokens":173306,"cached_tokens":173056}}`,
+			wantInput:     250,
+			wantCacheRead: 173056,
+			wantOutput:    166,
+		},
+		{
+			name:          "GLM nested prompt cache details",
+			body:          `{"usage":{"input_tokens":1200,"output_tokens":300,"prompt_tokens":1200,"prompt_tokens_details":{"cached_tokens":800}}}`,
+			wantInput:     400,
+			wantCacheRead: 800,
+			wantOutput:    300,
+		},
+		{
+			name:          "DeepSeek prompt cache hit and miss buckets",
+			body:          `{"usage":{"input_tokens":1200,"output_tokens":300,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400}}`,
+			wantInput:     400,
+			wantCacheRead: 800,
+			wantOutput:    300,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			usage := parseClaudeUsageFromResponseBody([]byte(tt.body))
+			require.Equal(t, tt.wantInput, usage.InputTokens)
+			require.Equal(t, tt.wantCacheRead, usage.CacheReadInputTokens)
+			require.Equal(t, tt.wantOutput, usage.OutputTokens)
+		})
+	}
+}
+
+func TestParseSSEUsagePassthroughNormalizesGLMAndDeepSeekAliases(t *testing.T) {
+	tests := []struct {
+		name          string
+		data          string
+		wantInput     int
+		wantCacheRead int
+	}{
+		{
+			name:          "GLM",
+			data:          `{"type":"message_delta","usage":{"input_tokens":1200,"output_tokens":30,"prompt_tokens":1200,"prompt_tokens_details":{"cached_tokens":800}}}`,
+			wantInput:     400,
+			wantCacheRead: 800,
+		},
+		{
+			name:          "DeepSeek",
+			data:          `{"type":"message_delta","usage":{"input_tokens":1200,"output_tokens":30,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400}}`,
+			wantInput:     400,
+			wantCacheRead: 800,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			usage := &ClaudeUsage{}
+			parseSSEUsagePassthrough(tt.data, usage)
+			require.Equal(t, tt.wantInput, usage.InputTokens)
+			require.Equal(t, tt.wantCacheRead, usage.CacheReadInputTokens)
+			require.Equal(t, 30, usage.OutputTokens)
+		})
+	}
+}
+
+func TestMergeAnthropicUsageNormalizesKimiStreamForOpenAIBilling(t *testing.T) {
+	var start apicompat.AnthropicStreamEvent
+	require.NoError(t, json.Unmarshal([]byte(`{"type":"message_start","message":{"usage":{"input_tokens":173306,"prompt_tokens":173306,"cached_tokens":0}}}`), &start))
+	var delta apicompat.AnthropicStreamEvent
+	require.NoError(t, json.Unmarshal([]byte(`{"type":"message_delta","usage":{"input_tokens":250,"cache_read_input_tokens":173056,"output_tokens":166,"prompt_tokens":173306,"cached_tokens":173056}}`), &delta))
+
+	usage := &ClaudeUsage{}
+	mergeAnthropicUsage(usage, start.Message.Usage)
+	mergeAnthropicUsage(usage, *delta.Usage)
+	require.Equal(t, 250, usage.InputTokens)
+	require.Equal(t, 173056, usage.CacheReadInputTokens)
+
+	openAIUsage := claudeUsageToOpenAIUsage(usage)
+	require.Equal(t, 173306, openAIUsage.InputTokens, "OpenAI gateway expects an inclusive input total")
+	require.Equal(t, 250, openAIUsage.InputTokens-openAIUsage.CacheReadInputTokens-openAIUsage.CacheCreationInputTokens)
+	require.Equal(t, 166, openAIUsage.OutputTokens)
+}
+
+func TestMergeAnthropicUsageNormalizesGLMAndDeepSeekAliases(t *testing.T) {
+	tests := []struct {
+		name string
+		raw  string
+	}{
+		{
+			name: "GLM",
+			raw:  `{"input_tokens":1200,"output_tokens":30,"prompt_tokens":1200,"prompt_tokens_details":{"cached_tokens":800}}`,
+		},
+		{
+			name: "DeepSeek",
+			raw:  `{"input_tokens":1200,"output_tokens":30,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400}`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var src apicompat.AnthropicUsage
+			require.NoError(t, json.Unmarshal([]byte(tt.raw), &src))
+
+			usage := &ClaudeUsage{}
+			mergeAnthropicUsage(usage, src)
+			require.Equal(t, 400, usage.InputTokens)
+			require.Equal(t, 800, usage.CacheReadInputTokens)
+
+			openAIUsage := claudeUsageToOpenAIUsage(usage)
+			require.Equal(t, 1200, openAIUsage.InputTokens)
+			require.Equal(t, 400, openAIUsage.InputTokens-openAIUsage.CacheReadInputTokens-openAIUsage.CacheCreationInputTokens)
+		})
+	}
+}
+
+func TestClaudeUsageToOpenAIUsagePreservesCNProviderNativeAnthropicBuckets(t *testing.T) {
+	tests := []struct {
+		name         string
+		usage        ClaudeUsage
+		wantTotal    int
+		wantUncached int
+	}{
+		{
+			name: "GLM",
+			usage: ClaudeUsage{
+				InputTokens:              2,
+				OutputTokens:             302,
+				CacheCreationInputTokens: 733,
+				CacheReadInputTokens:     376156,
+			},
+			wantTotal:    376891,
+			wantUncached: 2,
+		},
+		{
+			name: "DeepSeek",
+			usage: ClaudeUsage{
+				InputTokens:          400,
+				OutputTokens:         30,
+				CacheReadInputTokens: 800,
+			},
+			wantTotal:    1200,
+			wantUncached: 400,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			openAIUsage := claudeUsageToOpenAIUsage(&tt.usage)
+			require.Equal(t, tt.wantTotal, openAIUsage.InputTokens)
+			require.Equal(t, tt.wantUncached, openAIUsage.InputTokens-openAIUsage.CacheReadInputTokens-openAIUsage.CacheCreationInputTokens)
+			require.Equal(t, tt.usage.CacheReadInputTokens, openAIUsage.CacheReadInputTokens)
+			require.Equal(t, tt.usage.CacheCreationInputTokens, openAIUsage.CacheCreationInputTokens)
+		})
+	}
+}
+
+func TestCNProviderAnthropicUsageBillsUncachedInput(t *testing.T) {
+	tests := []struct {
+		name      string
+		model     string
+		body      string
+		wantInput int
+	}{
+		{
+			name:      "Kimi",
+			model:     "k3",
+			body:      `{"usage":{"input_tokens":173306,"output_tokens":166,"prompt_tokens":173306,"cached_tokens":173056}}`,
+			wantInput: 250,
+		},
+		{
+			name:      "GLM",
+			model:     "glm-5.2",
+			body:      `{"usage":{"input_tokens":1200,"output_tokens":30,"prompt_tokens":1200,"prompt_tokens_details":{"cached_tokens":800}}}`,
+			wantInput: 400,
+		},
+		{
+			name:      "DeepSeek",
+			model:     "deepseek-v4-flash",
+			body:      `{"usage":{"input_tokens":1200,"output_tokens":30,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400}}`,
+			wantInput: 400,
+		},
+	}
+
+	billing := NewBillingService(&config.Config{}, nil)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			claudeUsage := parseClaudeUsageFromResponseBody([]byte(tt.body))
+			openAIUsage := claudeUsageToOpenAIUsage(claudeUsage)
+			uncachedInput := max(openAIUsage.InputTokens-openAIUsage.CacheReadInputTokens-openAIUsage.CacheCreationInputTokens, 0)
+			require.Equal(t, tt.wantInput, uncachedInput)
+
+			cost, err := billing.CalculateCost(tt.model, UsageTokens{
+				InputTokens:         uncachedInput,
+				OutputTokens:        openAIUsage.OutputTokens,
+				CacheCreationTokens: openAIUsage.CacheCreationInputTokens,
+				CacheReadTokens:     openAIUsage.CacheReadInputTokens,
+			}, 1)
+			require.NoError(t, err)
+			require.Positive(t, cost.InputCost, "uncached input must contribute to the final charge")
+
+			pricing, err := billing.GetModelPricing(tt.model)
+			require.NoError(t, err)
+			require.InDelta(t, float64(tt.wantInput)*pricing.InputPricePerToken, cost.InputCost, 1e-12)
+		})
+	}
+}

backend/internal/service/openai_gateway_messages_anthropic_native.go

@@ -530,13 +530,15 @@ func (s *OpenAIGatewayService) nativeAnthropicStreamResult(
 }
 
 // claudeUsageToOpenAIUsage 把 Anthropic 格式 usage 映射到 OpenAI 网关统一的
-// 用量结构（字段一一对应）。
+// 用量结构。Anthropic 的 input_tokens 不含缓存读写，而 OpenAI 网关内部
+// 约定 InputTokens 是包含缓存明细的总输入；这里必须先合并，RecordUsage
+// 才能准确拆回互斥的计费桶。
 func claudeUsageToOpenAIUsage(u *ClaudeUsage) OpenAIUsage {
 	if u == nil {
 		return OpenAIUsage{}
 	}
 	return OpenAIUsage{
-		InputTokens:              u.InputTokens,
+		InputTokens:              u.InputTokens + u.CacheCreationInputTokens + u.CacheReadInputTokens,
 		OutputTokens:             u.OutputTokens,
 		CacheCreationInputTokens: u.CacheCreationInputTokens,
 		CacheReadInputTokens:     u.CacheReadInputTokens,