diff --git a/backend/internal/pkg/apicompat/types.go b/backend/internal/pkg/apicompat/types.go index b96d16b911..ab6df351ef 100644 --- a/backend/internal/pkg/apicompat/types.go +++ b/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"` } // --------------------------------------------------------------------------- diff --git a/backend/internal/service/gateway_anthropic_passthrough.go b/backend/internal/service/gateway_anthropic_passthrough.go index 320d800cec..ededd5b830 100644 --- a/backend/internal/service/gateway_anthropic_passthrough.go +++ b/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 } diff --git a/backend/internal/service/gateway_forward_as_responses.go b/backend/internal/service/gateway_forward_as_responses.go index cdd0197eab..6fcb1b2ed0 100644 --- a/backend/internal/service/gateway_forward_as_responses.go +++ b/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". diff --git a/backend/internal/service/kimi_anthropic_usage_test.go b/backend/internal/service/kimi_anthropic_usage_test.go new file mode 100644 index 0000000000..458ba9fb39 --- /dev/null +++ b/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) + }) + } +} diff --git a/backend/internal/service/openai_gateway_messages_anthropic_native.go b/backend/internal/service/openai_gateway_messages_anthropic_native.go index 8f107300b4..a4bc7b9f3a 100644 --- a/backend/internal/service/openai_gateway_messages_anthropic_native.go +++ b/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,