fix(billing): 文本模型 - 修复流式 Token 用量结算

为 Token 计费的流式 Chat Completions 请求开启 usage 终态块，并解析 Chat Completions 与 Responses SSE 用量。

backend/internal/handler/auth.go

@@ -715,8 +715,10 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 	modelName := proxyRequestModelForPath(path, c.GetHeader("Content-Type"), body)
 	protocol := model.ChannelInterfaceType("")
 	capability := "text"
+	var channelModel *model.ChannelModel
 	if !(c.Request.Method == http.MethodGet && path == "/models") {
-		channelModel, modelErr := svc.SystemChannelModel(channel.ID, modelName)
+		var modelErr error
+		channelModel, modelErr = svc.SystemChannelModel(channel.ID, modelName)
 		if modelErr != nil || channelModel.Protocol == "" {
 			fail(c, http.StatusForbidden, errors.New("当前系统渠道未授权该模型或模型协议尚未配置"))
 			return
@@ -728,6 +730,13 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 		fail(c, http.StatusForbidden, err)
 		return
 	}
+	if channelModel != nil && channelModel.BillingMode == "token" && protocol == model.ChannelInterfaceChatCompletion {
+		body, err = service.EnsureChatCompletionStreamUsageRequest(body)
+		if err != nil {
+			fail(c, http.StatusBadRequest, err)
+			return
+		}
+	}
 	billingOrderID := ""
 	query := c.Request.URL.Query()
 	for _, key := range []string{"key", "api_key", "access_token", "token"} {

backend/internal/service/analytics.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/csv"
 	"encoding/json"
@@ -939,13 +940,16 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) {
 	if log.ProviderRequestID == "" {
 		log.ProviderRequestID = providerRequestIDFromPath(log.Path)
 	}
-	if len(responseBody) == 0 || !json.Valid(responseBody) {
+	payloads := providerResponsePayloads(responseBody)
+	if len(payloads) == 0 {
 		return
 	}
-	var payload map[string]any
-	if json.Unmarshal(responseBody, &payload) != nil {
-		return
+	for _, payload := range payloads {
+		s.enrichAPICallLogPayload(log, payload)
 	}
+}
+
+func (s *Service) enrichAPICallLogPayload(log *model.ApiCallLog, payload map[string]any) {
 	if data, ok := payload["data"].(map[string]any); ok {
 		for key, value := range data {
 			if _, exists := payload[key]; !exists {
@@ -953,6 +957,14 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) {
 			}
 		}
 	}
+	// Responses API 的终态 SSE 把实际响应（包括 usage）放在 response 字段中。
+	if response, ok := payload["response"].(map[string]any); ok {
+		for key, value := range response {
+			if _, exists := payload[key]; !exists {
+				payload[key] = value
+			}
+		}
+	}
 	if log.Status == model.ApiCallStatusFailed {
 		errorCode, errorMessage := providerFailureDetails(payload)
 		log.ErrorCode = errorCode
@@ -962,9 +974,17 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) {
 	}
 	usage, _ := payload["usage"].(map[string]any)
 	if usage != nil {
-		log.UsageAvailable = true
-		log.InputTokens = firstInt64(usage, "input_tokens", "prompt_tokens")
-		log.OutputTokens = firstInt64(usage, "output_tokens", "completion_tokens")
+		inputTokens, inputAvailable := firstInt64Value(usage, "input_tokens", "prompt_tokens")
+		outputTokens, outputAvailable := firstInt64Value(usage, "output_tokens", "completion_tokens")
+		if inputAvailable {
+			log.InputTokens = inputTokens
+		}
+		if outputAvailable {
+			log.OutputTokens = outputTokens
+		}
+		if inputAvailable || outputAvailable {
+			log.UsageAvailable = true
+		}
 		// 火山方舟视频的输入 Token 恒为 0；查询任务以 completion_tokens 为实际用量，
 		// 兼容只返回 total_tokens 的同协议中转实现。
 		if log.Capability == "video" && strings.Contains(log.Path, "/contents/generations/tasks") && log.OutputTokens == 0 {
@@ -981,9 +1001,17 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) {
 		}
 	}
 	if usageMetadata, ok := payload["usageMetadata"].(map[string]any); ok {
-		log.UsageAvailable = true
-		log.InputTokens = firstInt64(usageMetadata, "promptTokenCount")
-		log.OutputTokens = firstInt64(usageMetadata, "candidatesTokenCount")
+		inputTokens, inputAvailable := firstInt64Value(usageMetadata, "promptTokenCount")
+		outputTokens, outputAvailable := firstInt64Value(usageMetadata, "candidatesTokenCount")
+		if inputAvailable {
+			log.InputTokens = inputTokens
+		}
+		if outputAvailable {
+			log.OutputTokens = outputTokens
+		}
+		if inputAvailable || outputAvailable {
+			log.UsageAvailable = true
+		}
 		log.CachedTokens = firstInt64(usageMetadata, "cachedContentTokenCount")
 	}
 	log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id"), stringField(payload, "name"), log.ProviderRequestID)
@@ -1003,6 +1031,45 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) {
 	}
 }
 
+func providerResponsePayloads(responseBody []byte) []map[string]any {
+	if len(responseBody) == 0 {
+		return nil
+	}
+	var payload map[string]any
+	if json.Unmarshal(responseBody, &payload) == nil {
+		return []map[string]any{payload}
+	}
+
+	// 流式文本的用量只出现在最后一个 SSE data 事件中，不能把整段响应当作 JSON。
+	result := make([]map[string]any, 0)
+	scanner := bufio.NewScanner(bytes.NewReader(responseBody))
+	scanner.Buffer(make([]byte, 64<<10), max(len(responseBody)+1, 64<<10))
+	dataLines := make([]string, 0, 1)
+	flush := func() {
+		raw := strings.TrimSpace(strings.Join(dataLines, "\n"))
+		dataLines = dataLines[:0]
+		if raw == "" || raw == "[DONE]" {
+			return
+		}
+		var event map[string]any
+		if json.Unmarshal([]byte(raw), &event) == nil {
+			result = append(result, event)
+		}
+	}
+	for scanner.Scan() {
+		line := strings.TrimSuffix(scanner.Text(), "\r")
+		if line == "" {
+			flush()
+			continue
+		}
+		if strings.HasPrefix(line, "data:") {
+			dataLines = append(dataLines, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " "))
+		}
+	}
+	flush()
+	return result
+}
+
 func providerRequestIDFromPath(path string) string {
 	parts := strings.Split(strings.Trim(strings.TrimSpace(path), "/"), "/")
 	for index := len(parts) - 1; index >= 0; index-- {
@@ -1019,18 +1086,23 @@ func providerRequestIDFromPath(path string) string {
 }
 
 func firstInt64(values map[string]any, keys ...string) int64 {
+	value, _ := firstInt64Value(values, keys...)
+	return value
+}
+
+func firstInt64Value(values map[string]any, keys ...string) (int64, bool) {
 	for _, key := range keys {
 		switch value := values[key].(type) {
 		case float64:
-			return int64(value)
+			return int64(value), true
 		case int64:
-			return value
+			return value, true
 		case json.Number:
-			parsed, _ := value.Int64()
-			return parsed
+			parsed, err := value.Int64()
+			return parsed, err == nil
 		}
 	}
-	return 0
+	return 0, false
 }
 
 func (s *Service) recordActivity(userID string, event string, count int) {

backend/internal/service/finance_token_test.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"encoding/json"
 	"testing"
 
 	"infinite-canvas/backend/internal/model"
@@ -83,3 +84,49 @@ func TestEnrichAPICallLogReadsArkVideoUsage(t *testing.T) {
 		t.Fatalf("EnrichAPICallLog() accepted empty Ark usage: %#v", missing)
 	}
 }
+
+func TestEnrichAPICallLogReadsChatCompletionStreamUsage(t *testing.T) {
+	log := &model.ApiCallLog{Capability: "text", Path: "/v1/chat/completions"}
+	stream := []byte("data: {\"id\":\"chatcmpl-test\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n" +
+		"data: {\"id\":\"chatcmpl-test\",\"choices\":[],\"usage\":{\"prompt_tokens\":23,\"completion_tokens\":7,\"prompt_tokens_details\":{\"cached_tokens\":5}}}\n\n" +
+		"data: [DONE]\n\n")
+	(&Service{}).EnrichAPICallLog(log, stream)
+	if !log.UsageAvailable || log.InputTokens != 23 || log.OutputTokens != 7 || log.CachedTokens != 5 {
+		t.Fatalf("EnrichAPICallLog() = %#v", log)
+	}
+}
+
+func TestEnrichAPICallLogReadsResponsesStreamUsage(t *testing.T) {
+	log := &model.ApiCallLog{Capability: "text", Path: "/v1/responses"}
+	stream := []byte("event: response.output_text.delta\n" +
+		"data: {\"type\":\"response.output_text.delta\",\"delta\":\"ok\"}\n\n" +
+		"event: response.completed\n" +
+		"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-test\",\"status\":\"completed\",\"usage\":{\"input_tokens\":31,\"output_tokens\":11,\"input_tokens_details\":{\"cached_tokens\":9}}}}\n\n")
+	(&Service{}).EnrichAPICallLog(log, stream)
+	if !log.UsageAvailable || log.InputTokens != 31 || log.OutputTokens != 11 || log.CachedTokens != 9 || log.ProviderRequestID != "resp-test" {
+		t.Fatalf("EnrichAPICallLog() = %#v", log)
+	}
+}
+
+func TestEnrichAPICallLogRejectsEmptyTextUsage(t *testing.T) {
+	log := &model.ApiCallLog{Capability: "text", Path: "/v1/chat/completions"}
+	(&Service{}).EnrichAPICallLog(log, []byte(`{"usage":{}}`))
+	if log.UsageAvailable {
+		t.Fatalf("EnrichAPICallLog() accepted empty text usage: %#v", log)
+	}
+}
+
+func TestEnsureChatCompletionStreamUsageRequest(t *testing.T) {
+	data, err := EnsureChatCompletionStreamUsageRequest([]byte(`{"model":"gpt-test","stream":true,"stream_options":{"custom":true}}`))
+	if err != nil {
+		t.Fatalf("EnsureChatCompletionStreamUsageRequest() error = %v", err)
+	}
+	var payload map[string]any
+	if err := json.Unmarshal(data, &payload); err != nil {
+		t.Fatalf("json.Unmarshal() error = %v", err)
+	}
+	options, _ := payload["stream_options"].(map[string]any)
+	if options["include_usage"] != true || options["custom"] != true {
+		t.Fatalf("stream_options = %#v", options)
+	}
+}

backend/internal/service/provider.go

@@ -113,6 +113,7 @@ type providerAnalyticsContext struct {
 	UserID            string
 	TaskID            string
 	BillingOrderID    string
+	BillingMode       string
 	Capability        string
 	Operation         string
 	ChannelID         string
@@ -125,6 +126,12 @@ type providerAnalyticsContext struct {
 
 func withProviderAnalytics(ctx context.Context, service *Service, task model.Task) context.Context {
 	metadata := providerAnalyticsContext{Service: service, UserID: task.UserID, TaskID: task.ID, BillingOrderID: task.BillingOrderID, Capability: capabilityFromTaskType(task.Type), Operation: task.Operation, Model: task.Model, ProviderRequestID: task.ProviderRequestID}
+	// 账单模式随请求上下文传递，流式协议据此只为 Token 计费开启 usage 终态块。
+	if service != nil && task.BillingOrderID != "" {
+		if order, err := service.repo.BillingOrder(task.BillingOrderID); err == nil {
+			metadata.BillingMode = order.BillingMode
+		}
+	}
 	var input struct {
 		Mode   string         `json:"mode"`
 		Config providerConfig `json:"config"`
@@ -2417,6 +2424,12 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn
 
 func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool) (string, error) {
 	if stream {
+		metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
+		if protocol == "chat-completion" && metadata.BillingMode == "token" {
+			if err := ensureChatCompletionStreamUsage(body); err != nil {
+				return "", err
+			}
+		}
 		return postStreamingText(ctx, config, path, body, protocol)
 	}
 	var payload map[string]interface{}

backend/internal/service/text_stream_usage.go

@@ -0,0 +1,37 @@
+package service
+
+import (
+	"encoding/json"
+	"errors"
+)
+
+// EnsureChatCompletionStreamUsageRequest 为 Token 计费的流式 Chat Completions
+// 请求开启最终 usage 块；非流式请求保持原样，避免改变普通接口合同。
+func EnsureChatCompletionStreamUsageRequest(data []byte) ([]byte, error) {
+	var payload map[string]any
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return nil, err
+	}
+	stream, _ := payload["stream"].(bool)
+	if !stream {
+		return data, nil
+	}
+	if err := ensureChatCompletionStreamUsage(payload); err != nil {
+		return nil, err
+	}
+	return json.Marshal(payload)
+}
+
+func ensureChatCompletionStreamUsage(payload map[string]any) error {
+	options := map[string]any{}
+	if value, exists := payload["stream_options"]; exists {
+		var ok bool
+		options, ok = value.(map[string]any)
+		if !ok {
+			return errors.New("stream_options 必须是 JSON 对象")
+		}
+	}
+	options["include_usage"] = true
+	payload["stream_options"] = options
+	return nil
+}