diff --git a/backend/internal/service/anthropic_chat_stream_usage_test.go b/backend/internal/service/anthropic_chat_stream_usage_test.go index 99ce4d2789..ff1558b073 100644 --- a/backend/internal/service/anthropic_chat_stream_usage_test.go +++ b/backend/internal/service/anthropic_chat_stream_usage_test.go @@ -22,19 +22,28 @@ func TestAnthropicChatStreamAuthoritativeUsage(t *testing.T) { for _, adapter := range []string{"anthropic", "native"} { for _, options := range []string{"", `,"stream_options":{"include_usage":true}`, `,"stream_options":{"include_usage":false}`} { for _, tc := range []struct { - name, start, delta string - want bool - prompt, output, cached, created int + name, start, delta, repeatDelta string + want bool + prompt, billableInput, totalInput, output, cached, created int }{ - {"cache", `,"usage":{"input_tokens":308,"cache_read_input_tokens":241,"cache_creation_input_tokens":17}`, `,"usage":{"output_tokens":49}`, true, 566, 49, 241, 17}, - {"zero_start", `,"usage":{"input_tokens":0,"output_tokens":0}`, "", true, 0, 0, 0, 0}, - {"zero_delta", "", `,"usage":{"input_tokens":0,"output_tokens":0}`, true, 0, 0, 0, 0}, - {"absent", "", "", false, 0, 0, 0, 0}, - {"null", `,"usage":null`, `,"usage":null`, false, 0, 0, 0, 0}, + {"cache", `,"usage":{"input_tokens":308,"cache_read_input_tokens":241,"cache_creation_input_tokens":17}`, `,"usage":{"output_tokens":49}`, "", true, 566, 308, 566, 49, 241, 17}, + {"deepseek_partial_cache", "", `,"usage":{"input_tokens":1200,"output_tokens":30,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400}`, "", true, 1200, 400, 1200, 30, 800, 0}, + {"openai_partial_cache", "", `,"usage":{"prompt_tokens":1200,"output_tokens":30,"prompt_tokens_details":{"cached_tokens":800}}`, "", true, 1200, 400, 1200, 30, 800, 0}, + {"repeated_cache_bucket", `,"usage":{"input_tokens":1200}`, `,"usage":{"output_tokens":30,"cache_read_input_tokens":800}`, `,"usage":{"output_tokens":30,"cache_read_input_tokens":800}`, true, 1200, 400, 1200, 30, 800, 0}, + {"kimi_full_cache", `,"usage":{"input_tokens":173306}`, `,"usage":{"output_tokens":49,"cache_read_input_tokens":173306}`, "", true, 173306, 0, 173306, 49, 173306, 0}, + {"kimi_full_cache_cached_tokens", `,"usage":{"input_tokens":173306}`, `,"usage":{"output_tokens":49,"cached_tokens":173306}`, "", true, 173306, 0, 173306, 49, 173306, 0}, + {"kimi_full_cache_prompt_details", `,"usage":{"input_tokens":173306}`, `,"usage":{"output_tokens":49,"prompt_tokens_details":{"cached_tokens":173306}}`, "", true, 173306, 0, 173306, 49, 173306, 0}, + {"zero_start", `,"usage":{"input_tokens":0,"output_tokens":0}`, "", "", true, 0, 0, 0, 0, 0, 0}, + {"zero_delta", "", `,"usage":{"input_tokens":0,"output_tokens":0}`, "", true, 0, 0, 0, 0, 0, 0}, + {"absent", "", "", "", false, 0, 0, 0, 0, 0, 0}, + {"null", `,"usage":null`, `,"usage":null`, "", false, 0, 0, 0, 0, 0, 0}, } { for _, terminal := range []string{"stop", "eof"} { t.Run(adapter+"/"+options+"/"+tc.name+"/"+terminal, func(t *testing.T) { sse := fmt.Sprintf("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"k3\",\"content\":[]%s}}\n\nevent: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}%s}\n\n", tc.start, tc.delta) + if tc.repeatDelta != "" { + sse += fmt.Sprintf("event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}%s}\n\n", tc.repeatDelta) + } if terminal == "stop" { // Repeated terminal events and EOF must not duplicate usage. sse += "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n" @@ -52,6 +61,7 @@ func TestAnthropicChatStreamAuthoritativeUsage(t *testing.T) { svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream} result, err := svc.forwardChatCompletionsViaNativeAnthropic(context.Background(), c, nativeAnthropicTestAccount(), body, "") require.NoError(t, err) + require.Equal(t, tc.totalInput, result.Usage.InputTokens) require.Equal(t, tc.output, result.Usage.OutputTokens) require.Equal(t, tc.cached, result.Usage.CacheReadInputTokens) require.Equal(t, tc.created, result.Usage.CacheCreationInputTokens) @@ -60,6 +70,7 @@ func TestAnthropicChatStreamAuthoritativeUsage(t *testing.T) { account := &Account{ID: 1, Platform: PlatformAnthropic, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "sk-test"}} result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, nil) require.NoError(t, err) + require.Equal(t, tc.billableInput, result.Usage.InputTokens) require.Equal(t, tc.output, result.Usage.OutputTokens) require.Equal(t, tc.cached, result.Usage.CacheReadInputTokens) require.Equal(t, tc.created, result.Usage.CacheCreationInputTokens) diff --git a/backend/internal/service/gateway_forward_as_chat_completions.go b/backend/internal/service/gateway_forward_as_chat_completions.go index 687e31cc6f..0936af2b46 100644 --- a/backend/internal/service/gateway_forward_as_chat_completions.go +++ b/backend/internal/service/gateway_forward_as_chat_completions.go @@ -430,6 +430,11 @@ func (s *GatewayService) handleCCStreamingFromAnthropic( mergeAnthropicUsage(&usage, event.Message.Usage) } + // Keep the outward Responses/Chat usage on the same normalized buckets used + // for billing, including converter handlers that consume event usage. + syncAnthropicResponsesUsage(anthState, usage) + normalizeAnthropicEventUsageForResponses(event, usage) + // Chain: Anthropic event → Responses events → CC chunks responsesEvents := apicompat.AnthropicEventToResponsesEvents(event, anthState) for _, resEvt := range responsesEvents { diff --git a/backend/internal/service/gateway_forward_as_responses.go b/backend/internal/service/gateway_forward_as_responses.go index 1b97a73961..6903e659ad 100644 --- a/backend/internal/service/gateway_forward_as_responses.go +++ b/backend/internal/service/gateway_forward_as_responses.go @@ -279,22 +279,22 @@ func mergeAnthropicUsage(dst *ClaudeUsage, src apicompat.AnthropicUsage) { return } + 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) + } + // 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 { @@ -303,13 +303,21 @@ func mergeAnthropicUsage(dst *ClaudeUsage, src apicompat.AnthropicUsage) { dst.CacheReadInputTokens = cacheReadTokens dst.CacheCreationInputTokens = src.CacheCreationInputTokens } else { + previousCacheReadTokens := dst.CacheReadInputTokens + previousCacheCreationTokens := dst.CacheCreationInputTokens 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.InputTokens == 0 && dst.InputTokens > 0 && (cacheReadTokens > 0 || src.CacheCreationInputTokens > 0) { + // Some compatible streams put the total prompt count in message_start, + // then provide only cumulative cache buckets in message_delta. Subtract + // only newly observed buckets so repeated deltas are idempotent. + newCacheReadTokens := max(cacheReadTokens-previousCacheReadTokens, 0) + newCacheCreationTokens := max(src.CacheCreationInputTokens-previousCacheCreationTokens, 0) + dst.InputTokens = max(dst.InputTokens-newCacheReadTokens-newCacheCreationTokens, 0) + } + if cacheReadTokens > 0 { + dst.CacheReadInputTokens = cacheReadTokens } if src.CacheCreationInputTokens > 0 { dst.CacheCreationInputTokens = src.CacheCreationInputTokens @@ -320,6 +328,29 @@ func mergeAnthropicUsage(dst *ClaudeUsage, src apicompat.AnthropicUsage) { } } +func syncAnthropicResponsesUsage(state *apicompat.AnthropicEventToResponsesState, usage ClaudeUsage) { + state.InputTokens = usage.InputTokens + state.OutputTokens = usage.OutputTokens + state.CacheReadInputTokens = usage.CacheReadInputTokens + state.CacheCreationInputTokens = usage.CacheCreationInputTokens +} + +func normalizeAnthropicEventUsageForResponses(event *apicompat.AnthropicStreamEvent, usage ClaudeUsage) { + normalize := func(dst *apicompat.AnthropicUsage) { + if dst == nil { + return + } + dst.InputTokens = usage.InputTokens + dst.OutputTokens = usage.OutputTokens + dst.CacheReadInputTokens = usage.CacheReadInputTokens + dst.CacheCreationInputTokens = usage.CacheCreationInputTokens + } + normalize(event.Usage) + if event.Message != nil { + normalize(&event.Message.Usage) + } +} + // parseAnthropicSSEField parses an SSE field line in the form "field:value" or "field: value". // According to the SSE spec (https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation), // the space after the colon is optional. This function handles both formats. @@ -545,6 +576,12 @@ func (s *GatewayService) handleResponsesStreamingResponse( mergeAnthropicUsage(&usage, event.Message.Usage) } + // Keep the terminal Responses usage aligned with the normalized billing + // buckets. Normalize the converter input too, so message handlers cannot + // restore the provider's overlapping raw input total. + syncAnthropicResponsesUsage(state, usage) + normalizeAnthropicEventUsageForResponses(event, usage) + // Convert to Responses events events := apicompat.AnthropicEventToResponsesEvents(event, state) for _, evt := range events { diff --git a/backend/internal/service/gateway_forward_as_responses_test.go b/backend/internal/service/gateway_forward_as_responses_test.go index 972c6bc1a6..353eeb8846 100644 --- a/backend/internal/service/gateway_forward_as_responses_test.go +++ b/backend/internal/service/gateway_forward_as_responses_test.go @@ -307,6 +307,116 @@ func TestHandleResponsesStreamingResponse_PreservesMessageStartCacheUsage(t *tes require.Contains(t, rec.Body.String(), `response.completed`) } +func TestHandleResponsesStreamingResponse_NormalizesTerminalUsage(t *testing.T) { + gin.SetMode(gin.TestMode) + + tests := []struct { + name string + startUsage string + deltaUsage string + wantInput int + wantOutput int + wantCached int + wantCacheCreation int + }{ + { + name: "cache_read_input_tokens full cache", + startUsage: `"input_tokens":173306`, + deltaUsage: `"output_tokens":8,"cache_read_input_tokens":173306`, + wantOutput: 8, + wantCached: 173306, + }, + { + name: "cached_tokens full cache", + startUsage: `"input_tokens":173306`, + deltaUsage: `"output_tokens":8,"cached_tokens":173306`, + wantOutput: 8, + wantCached: 173306, + }, + { + name: "prompt_tokens_details full cache", + startUsage: `"input_tokens":173306`, + deltaUsage: `"output_tokens":8,"prompt_tokens_details":{"cached_tokens":173306}`, + wantOutput: 8, + wantCached: 173306, + }, + { + name: "DeepSeek input total with hit and miss buckets", + startUsage: `"input_tokens":0`, + deltaUsage: `"input_tokens":1200,"output_tokens":30,"prompt_cache_hit_tokens":800,"prompt_cache_miss_tokens":400`, + wantInput: 400, + wantOutput: 30, + wantCached: 800, + }, + { + name: "OpenAI prompt total with cached details", + startUsage: `"input_tokens":0`, + deltaUsage: `"prompt_tokens":1200,"output_tokens":30,"prompt_tokens_details":{"cached_tokens":800}`, + wantInput: 400, + wantOutput: 30, + wantCached: 800, + }, + { + name: "cache creation only without prompt total or miss bucket", + startUsage: `"input_tokens":1200`, + deltaUsage: `"output_tokens":30,"cache_creation_input_tokens":800`, + wantInput: 400, + wantOutput: 30, + wantCacheCreation: 800, + }, + } + + for _, tt := range tests { + for _, terminal := range []string{"message_stop", "eof"} { + t.Run(tt.name+"/"+terminal, func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + lines := []string{ + `event: message_start`, + `data: {"type":"message_start","message":{"id":"msg_usage","type":"message","role":"assistant","content":[],"model":"k3","stop_reason":"","usage":{` + tt.startUsage + `}}}`, + ``, + `event: message_delta`, + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{` + tt.deltaUsage + `}}`, + ``, + } + if terminal == "message_stop" { + lines = append(lines, `event: message_stop`, `data: {"type":"message_stop"}`, ``) + } + resp := &http.Response{Body: io.NopCloser(strings.NewReader(strings.Join(lines, "\n")))} + + result, err := (&GatewayService{}).handleResponsesStreamingResponse(resp, c, "k3", "k3", nil, time.Now(), apicompat.ResponsesClientToolMapping{}) + require.NoError(t, err) + require.Equal(t, tt.wantInput, result.Usage.InputTokens) + require.Equal(t, tt.wantCached, result.Usage.CacheReadInputTokens) + require.Equal(t, tt.wantCacheCreation, result.Usage.CacheCreationInputTokens) + + var completed apicompat.ResponsesStreamEvent + for _, line := range strings.Split(rec.Body.String(), "\n") { + if !strings.HasPrefix(line, "data: ") { + continue + } + var event apicompat.ResponsesStreamEvent + require.NoError(t, json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &event)) + if event.Type == "response.completed" { + completed = event + } + } + require.NotNil(t, completed.Response) + require.NotNil(t, completed.Response.Usage) + require.Equal(t, tt.wantInput+tt.wantCached+tt.wantCacheCreation, completed.Response.Usage.InputTokens) + require.Equal(t, tt.wantOutput, completed.Response.Usage.OutputTokens) + require.Equal(t, completed.Response.Usage.InputTokens+tt.wantOutput, completed.Response.Usage.TotalTokens) + require.Equal(t, tt.wantCacheCreation, completed.Response.Usage.CacheCreationInputTokens) + if tt.wantCached == 0 { + require.Nil(t, completed.Response.Usage.InputTokensDetails) + } else { + require.Equal(t, tt.wantCached, completed.Response.Usage.InputTokensDetails.CachedTokens) + } + }) + } + } +} + func TestParseAnthropicSSEField(t *testing.T) { t.Parallel() diff --git a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go index 40b00c81e6..03c8b8e729 100644 --- a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go +++ b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go @@ -407,6 +407,11 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic( mergeAnthropicUsage(&usage, event.Message.Usage) } + // Keep the outward Responses/Chat usage on the same normalized buckets used + // for billing, including converter handlers that consume event usage. + syncAnthropicResponsesUsage(anthState, usage) + normalizeAnthropicEventUsageForResponses(event, usage) + // 客户端已断开：跳过转换与写出，继续读上游直到流结束（usage 完整、 // 连接及时归还），不再提前 return。 if clientDisconnected { diff --git a/backend/internal/service/openai_gateway_cn_fixes_test.go b/backend/internal/service/openai_gateway_cn_fixes_test.go index 0dfb7c35b6..470c789c74 100644 --- a/backend/internal/service/openai_gateway_cn_fixes_test.go +++ b/backend/internal/service/openai_gateway_cn_fixes_test.go @@ -11,9 +11,12 @@ package service import ( "context" + "encoding/json" "errors" + "io" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -109,6 +112,83 @@ func TestResponsesStreamingFromNativeAnthropic_ClientDisconnectDrainsUsage(t *te "output_tokens 必须来自排水读到的末尾 message_delta（断开即弃时会是 1）") } +func TestResponsesStreamingFromNativeAnthropic_NormalizesTerminalUsage(t *testing.T) { + gin.SetMode(gin.TestMode) + + tests := []struct { + name string + startUsage string + deltaUsage string + repeatDelta bool + wantInput int + wantOutput int + wantCached int + wantCacheCreation int + }{ + {name: "full cache", startUsage: `"input_tokens":1200`, deltaUsage: `"output_tokens":30,"cache_read_input_tokens":1200`, wantOutput: 30, wantCached: 1200}, + {name: "partial cache", startUsage: `"input_tokens":1200`, deltaUsage: `"output_tokens":30,"cache_read_input_tokens":800`, wantInput: 400, wantOutput: 30, wantCached: 800}, + {name: "cache creation", startUsage: `"input_tokens":1200`, deltaUsage: `"output_tokens":30,"cache_creation_input_tokens":800`, wantInput: 400, wantOutput: 30, wantCacheCreation: 800}, + {name: "repeated cumulative cache bucket", startUsage: `"input_tokens":1200`, deltaUsage: `"output_tokens":30,"cache_read_input_tokens":800`, repeatDelta: true, wantInput: 400, wantOutput: 30, wantCached: 800}, + } + + for _, tt := range tests { + for _, terminal := range []string{"message_stop", "eof"} { + t.Run(tt.name+"/"+terminal, func(t *testing.T) { + lines := []string{ + `event: message_start`, + `data: {"type":"message_start","message":{"id":"msg_usage","type":"message","role":"assistant","content":[],"model":"k3","stop_reason":"","usage":{` + tt.startUsage + `}}}`, + ``, + `event: message_delta`, + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{` + tt.deltaUsage + `}}`, + ``, + } + if tt.repeatDelta { + lines = append(lines, `event: message_delta`, `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{`+tt.deltaUsage+`}}`, ``) + } + if terminal == "message_stop" { + lines = append(lines, `event: message_stop`, `data: {"type":"message_stop"}`, ``) + } + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(strings.Join(lines, "\n")))} + + result, err := (&OpenAIGatewayService{}).handleResponsesStreamingFromNativeAnthropic( + resp, c, "k3", "k3", "k3", nil, time.Now(), apicompat.ResponsesClientToolMapping{}) + require.NoError(t, err) + require.Equal(t, tt.wantInput+tt.wantCached+tt.wantCacheCreation, result.Usage.InputTokens) + require.Equal(t, tt.wantOutput, result.Usage.OutputTokens) + require.Equal(t, tt.wantCached, result.Usage.CacheReadInputTokens) + require.Equal(t, tt.wantCacheCreation, result.Usage.CacheCreationInputTokens) + + var completed apicompat.ResponsesStreamEvent + for _, line := range strings.Split(rec.Body.String(), "\n") { + if !strings.HasPrefix(line, "data: ") { + continue + } + var event apicompat.ResponsesStreamEvent + require.NoError(t, json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &event)) + if event.Type == "response.completed" { + completed = event + } + } + require.NotNil(t, completed.Response) + require.NotNil(t, completed.Response.Usage) + require.Equal(t, tt.wantInput+tt.wantCached+tt.wantCacheCreation, completed.Response.Usage.InputTokens) + require.Equal(t, tt.wantOutput, completed.Response.Usage.OutputTokens) + require.Equal(t, completed.Response.Usage.InputTokens+tt.wantOutput, completed.Response.Usage.TotalTokens) + require.Equal(t, tt.wantCacheCreation, completed.Response.Usage.CacheCreationInputTokens) + if tt.wantCached == 0 { + require.Nil(t, completed.Response.Usage.InputTokensDetails) + } else { + require.Equal(t, tt.wantCached, completed.Response.Usage.InputTokensDetails.CachedTokens) + } + }) + } + } +} + func TestHandle403_CNProviderHTMLBodySkipsAccountPenalty(t *testing.T) { for _, platform := range []string{PlatformKimi, PlatformZhipu, PlatformDeepseek, PlatformMiniMax} { repo := &rateLimitAccountRepoStub{} diff --git a/backend/internal/service/openai_gateway_responses_anthropic_native.go b/backend/internal/service/openai_gateway_responses_anthropic_native.go index fa2e190b97..c5f8ef6887 100644 --- a/backend/internal/service/openai_gateway_responses_anthropic_native.go +++ b/backend/internal/service/openai_gateway_responses_anthropic_native.go @@ -397,6 +397,12 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic( mergeAnthropicUsage(&usage, event.Message.Usage) } + // Keep terminal Responses usage aligned with the normalized billing + // buckets. Normalize converter input too so raw overlapping totals cannot + // overwrite the state when message_start/message_delta handlers run. + syncAnthropicResponsesUsage(state, usage) + normalizeAnthropicEventUsageForResponses(event, usage) + events := apicompat.AnthropicEventToResponsesEvents(event, state) if clientDisconnected { return