fix(gateway): normalize streamed Anthropic usage

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)

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 {

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 {

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()
 

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 {

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{}

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