diff --git a/backend/internal/service/gateway_anthropic_passthrough.go b/backend/internal/service/gateway_anthropic_passthrough.go index 533e780017..320d800cec 100644 --- a/backend/internal/service/gateway_anthropic_passthrough.go +++ b/backend/internal/service/gateway_anthropic_passthrough.go @@ -296,6 +296,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput( UpstreamModel: input.RequestModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: input.RequestStream, Duration: time.Since(input.StartTime), FirstTokenMs: firstTokenMs, diff --git a/backend/internal/service/gateway_forward.go b/backend/internal/service/gateway_forward.go index d8a105c7d8..17f9228d5d 100644 --- a/backend/internal/service/gateway_forward.go +++ b/backend/internal/service/gateway_forward.go @@ -888,6 +888,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A UpstreamModel: mappedModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: reqStream, Duration: time.Since(startTime), FirstTokenMs: firstTokenMs, @@ -913,8 +914,9 @@ func anthropicSpeedModel(parsed *ParsedRequest, result *ForwardResult) string { // 承载（Opus 4.7 的 fast mode 已被移除，传 speed=fast 会直接报错）。这里按模型和 // 平台收紧，避免上游根本没跑 fast 时仍然按 2x 计费——宁可漏收也不能多收。 // -// 注：判据是请求参数而非响应里的 usage.speed。等 usage 解析链路统一暴露该字段后， -// 应改为以响应为准。 +// 这里只决定请求侧的档位；上游响应里的 usage.speed 由 UpstreamResponseServiceTier +// 带回，用量记录时经 ResolveBillingServiceTier 只降不升（usage.speed=standard 则按 +// 标准价计费）。 func anthropicSpeedServiceTier(account *Account, speed, model string) *string { if account == nil || account.Platform != PlatformAnthropic || speed != "fast" { return nil diff --git a/backend/internal/service/gateway_record_usage_test.go b/backend/internal/service/gateway_record_usage_test.go index 5361644745..0be0477029 100644 --- a/backend/internal/service/gateway_record_usage_test.go +++ b/backend/internal/service/gateway_record_usage_test.go @@ -694,3 +694,74 @@ func TestGatewayServiceRecordUsage_ReasoningEffortNil(t *testing.T) { require.NotNil(t, usageRepo.lastLog) require.Nil(t, usageRepo.lastLog.ReasoningEffort) } + +// newGatewayRecordUsageServiceWithResolverForTest mirrors production wiring for +// token billing: a pricing resolver plus a grouped API key select the unified +// billing path, which is the only one that honours the service tier. +func newGatewayRecordUsageServiceWithResolverForTest(usageRepo UsageLogRepository) (*GatewayService, *APIKey) { + svc := newGatewayRecordUsageServiceForTest(usageRepo, &openAIRecordUsageUserRepoStub{}, &openAIRecordUsageSubRepoStub{}) + svc.resolver = NewModelPricingResolver(nil, svc.billingService) + groupID := int64(7) + return svc, &APIKey{ID: 1, GroupID: &groupID, Group: &Group{ID: groupID, RateMultiplier: 1.0}} +} + +func TestGatewayServiceRecordUsage_FastSpeedDowngradedByUpstreamResponse(t *testing.T) { + usageRepo := &openAIRecordUsageBestEffortLogRepoStub{} + svc, apiKey := newGatewayRecordUsageServiceWithResolverForTest(usageRepo) + + tier := "fast" + err := svc.RecordUsage(context.Background(), &RecordUsageInput{ + Result: &ForwardResult{ + RequestID: "fast_downgraded_test", + Usage: ClaudeUsage{InputTokens: 100, OutputTokens: 50}, + Model: "claude-opus-5", + Duration: time.Second, + ServiceTier: &tier, + UpstreamResponseServiceTier: "standard", + }, + APIKey: apiKey, + User: &User{ID: 1}, + Account: &Account{ID: 1, Platform: PlatformAnthropic}, + }) + + require.NoError(t, err) + require.NotNil(t, usageRepo.lastLog) + require.NotNil(t, usageRepo.lastLog.ServiceTier) + require.Equal(t, "standard", *usageRepo.lastLog.ServiceTier) + + tokens := UsageTokens{InputTokens: 100, OutputTokens: 50} + standardCost, err := svc.billingService.CalculateCost("claude-opus-5", tokens, 1.0) + require.NoError(t, err) + fastCost, err := svc.billingService.CalculateCostWithServiceTier("claude-opus-5", tokens, 1.0, "fast") + require.NoError(t, err) + require.Greater(t, fastCost.TotalCost, standardCost.TotalCost, "fast mode must carry a premium for the test to be meaningful") + require.InDelta(t, standardCost.TotalCost, usageRepo.lastLog.TotalCost, 1e-10) +} + +func TestGatewayServiceRecordUsage_FastSpeedHonouredKeepsPremium(t *testing.T) { + usageRepo := &openAIRecordUsageBestEffortLogRepoStub{} + svc, apiKey := newGatewayRecordUsageServiceWithResolverForTest(usageRepo) + + tier := "fast" + err := svc.RecordUsage(context.Background(), &RecordUsageInput{ + Result: &ForwardResult{ + RequestID: "fast_honoured_test", + Usage: ClaudeUsage{InputTokens: 100, OutputTokens: 50}, + Model: "claude-opus-5", + Duration: time.Second, + ServiceTier: &tier, + UpstreamResponseServiceTier: "fast", + }, + APIKey: apiKey, + User: &User{ID: 1}, + Account: &Account{ID: 1, Platform: PlatformAnthropic}, + }) + + require.NoError(t, err) + require.NotNil(t, usageRepo.lastLog) + require.Equal(t, "fast", *usageRepo.lastLog.ServiceTier) + + fastCost, err := svc.billingService.CalculateCostWithServiceTier("claude-opus-5", UsageTokens{InputTokens: 100, OutputTokens: 50}, 1.0, "fast") + require.NoError(t, err) + require.InDelta(t, fastCost.TotalCost, usageRepo.lastLog.TotalCost, 1e-10) +} diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index fd39ce2bc3..0c0140b338 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -614,13 +614,18 @@ type ForwardResult struct { // response before any client-facing rewrite or protocol conversion. UpstreamResponseModel string UpstreamResponseModelConflict bool - Stream bool - Duration time.Duration - FirstTokenMs *int // 首字时间（流式请求） - ClientDisconnect bool // 客户端是否在流式传输过程中断开 - ReasoningEffort *string - // ServiceTier records the billable request tier. OpenAI uses service_tier; - // Anthropic speed=fast is normalized to "fast". + // UpstreamResponseServiceTier is the tier the upstream reports having used + // (Anthropic usage.speed: "fast" / "standard"); "" when not declared. + UpstreamResponseServiceTier string + Stream bool + Duration time.Duration + FirstTokenMs *int // 首字时间（流式请求） + ClientDisconnect bool // 客户端是否在流式传输过程中断开 + ReasoningEffort *string + // ServiceTier records the tier requested by the client. OpenAI uses + // service_tier; Anthropic speed=fast is normalized to "fast". Usage recording + // lowers it to UpstreamResponseServiceTier when the upstream reports a + // cheaper tier (see ResolveBillingServiceTier). ServiceTier *string // 图片生成计费字段（图片生成模型使用） diff --git a/backend/internal/service/gateway_upstream_response.go b/backend/internal/service/gateway_upstream_response.go index 20af32314f..4476c60977 100644 --- a/backend/internal/service/gateway_upstream_response.go +++ b/backend/internal/service/gateway_upstream_response.go @@ -680,6 +680,7 @@ func partialStreamUsageResult(c *gin.Context, resp *http.Response, streamResult UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: true, Duration: time.Since(startTime), FirstTokenMs: streamResult.firstTokenMs, diff --git a/backend/internal/service/gateway_usage_billing.go b/backend/internal/service/gateway_usage_billing.go index e1fac3d6cf..0a770bc7a3 100644 --- a/backend/internal/service/gateway_usage_billing.go +++ b/backend/internal/service/gateway_usage_billing.go @@ -776,6 +776,7 @@ func (s *GatewayService) recordUsageCore(ctx context.Context, input *recordUsage account := input.Account subscription := input.Subscription ApplyForwardImageBillingResolution(result) + logServiceTierBillingDowngrade("service.gateway", account, result.RequestID, ApplyForwardServiceTierBillingResolution(result)) // 强制缓存计费：将 input_tokens 转为 cache_read_input_tokens // 用于粘性会话切换时的特殊计费处理 diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index aceb186fbc..d5639fbe31 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -548,6 +548,7 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: false, Duration: time.Since(startTime), } @@ -668,6 +669,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: true, Duration: time.Since(startTime), FirstTokenMs: firstTokenMs, diff --git a/backend/internal/service/openai_gateway_chat_completions_raw.go b/backend/internal/service/openai_gateway_chat_completions_raw.go index 70e623d2bf..a3eeecda2e 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw.go +++ b/backend/internal/service/openai_gateway_chat_completions_raw.go @@ -387,6 +387,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), ReasoningEffort: reasoningEffort, ServiceTier: serviceTier, Stream: true, @@ -488,6 +489,7 @@ func (s *OpenAIGatewayService) bufferRawChatCompletions( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), ReasoningEffort: reasoningEffort, ServiceTier: serviceTier, Stream: false, diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 3327a2f6a0..b4d717cec8 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1174,6 +1174,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), ServiceTier: serviceTier, ReasoningEffort: reasoningEffort, Stream: reqStream, diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 5c799777c7..c9113050e1 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -636,6 +636,7 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: false, Duration: time.Since(startTime), } @@ -939,6 +940,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( UpstreamModel: upstreamModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), Stream: true, Duration: time.Since(startTime), FirstTokenMs: firstTokenMs, diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 88f5717094..c8af9a70a1 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -514,6 +514,7 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( UpstreamModel: upstreamPassthroughModel, UpstreamResponseModel: observedUpstreamResponseModel(c), UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), ServiceTier: serviceTier, ReasoningEffort: reasoningEffort, Stream: reqStream, diff --git a/backend/internal/service/openai_gateway_record_usage_test.go b/backend/internal/service/openai_gateway_record_usage_test.go index 9cca810400..9f9c7b4c6a 100644 --- a/backend/internal/service/openai_gateway_record_usage_test.go +++ b/backend/internal/service/openai_gateway_record_usage_test.go @@ -2894,3 +2894,64 @@ func TestGatewayServiceCalculateRecordUsageCost_ChannelImageBillingNormalizesMis require.InDelta(t, 0.44, cost.TotalCost, 1e-12) require.InDelta(t, 0.44, cost.ActualCost, 1e-12) } + +func TestOpenAIGatewayServiceRecordUsage_ServiceTierDowngradedByUpstreamResponse(t *testing.T) { + usageRepo := &openAIRecordUsageLogRepoStub{inserted: true} + userRepo := &openAIRecordUsageUserRepoStub{} + subRepo := &openAIRecordUsageSubRepoStub{} + svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, subRepo, nil) + serviceTier := "priority" + usage := OpenAIUsage{InputTokens: 100, OutputTokens: 50} + + err := svc.RecordUsage(context.Background(), &OpenAIRecordUsageInput{ + Result: &OpenAIForwardResult{ + RequestID: "resp_service_tier_downgraded", + ServiceTier: &serviceTier, + UpstreamResponseServiceTier: "default", + Usage: usage, + Model: "gpt-5.4", + Duration: time.Second, + }, + APIKey: &APIKey{ID: 1017}, + User: &User{ID: 2017}, + Account: &Account{ID: 3017}, + }) + + require.NoError(t, err) + require.NotNil(t, usageRepo.lastLog) + require.NotNil(t, usageRepo.lastLog.ServiceTier) + require.Equal(t, "default", *usageRepo.lastLog.ServiceTier, "usage log must record the tier actually billed") + + baseCost, calcErr := svc.billingService.CalculateCost("gpt-5.4", UsageTokens{InputTokens: 100, OutputTokens: 50}, 1.0) + require.NoError(t, calcErr) + require.InDelta(t, baseCost.TotalCost, usageRepo.lastLog.TotalCost, 1e-10, "a request served at default must not pay the priority price") +} + +func TestOpenAIGatewayServiceRecordUsage_ServiceTierNeverRaisedByUpstreamResponse(t *testing.T) { + usageRepo := &openAIRecordUsageLogRepoStub{inserted: true} + userRepo := &openAIRecordUsageUserRepoStub{} + subRepo := &openAIRecordUsageSubRepoStub{} + svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, subRepo, nil) + usage := OpenAIUsage{InputTokens: 100, OutputTokens: 50} + + err := svc.RecordUsage(context.Background(), &OpenAIRecordUsageInput{ + Result: &OpenAIForwardResult{ + RequestID: "resp_service_tier_not_raised", + UpstreamResponseServiceTier: "priority", + Usage: usage, + Model: "gpt-5.4", + Duration: time.Second, + }, + APIKey: &APIKey{ID: 1018}, + User: &User{ID: 2018}, + Account: &Account{ID: 3018}, + }) + + require.NoError(t, err) + require.NotNil(t, usageRepo.lastLog) + require.Nil(t, usageRepo.lastLog.ServiceTier) + + baseCost, calcErr := svc.billingService.CalculateCost("gpt-5.4", UsageTokens{InputTokens: 100, OutputTokens: 50}, 1.0) + require.NoError(t, calcErr) + require.InDelta(t, baseCost.TotalCost, usageRepo.lastLog.TotalCost, 1e-10) +} diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 71e585f09e..27d94ea54c 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -247,11 +247,16 @@ type OpenAIForwardResult struct { // response before any client-facing rewrite or protocol conversion. UpstreamResponseModel string UpstreamResponseModelConflict bool + // UpstreamResponseServiceTier is the tier the upstream reports having used + // (response service_tier: "priority" / "default" / "flex" / ...); "" when not declared. + UpstreamResponseServiceTier string // UpstreamEndpoint is the actual upstream API path used for this request. // It avoids guessing when one downstream protocol can use multiple upstream endpoints. UpstreamEndpoint string - // ServiceTier records the OpenAI Responses API service tier, e.g. "priority" / "flex". - // Nil means the request did not specify a recognized tier. + // ServiceTier records the OpenAI Responses API service tier requested by the + // client, e.g. "priority" / "flex". Nil means the request did not specify a + // recognized tier. Usage recording lowers it to UpstreamResponseServiceTier + // when the upstream reports a cheaper tier (see ResolveBillingServiceTier). ServiceTier *string // ReasoningEffort is extracted from request body (reasoning.effort) or derived from model suffix. // Stored for usage records display; nil means not provided / not applicable. diff --git a/backend/internal/service/openai_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index 19f3fc6800..25af28973a 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -146,6 +146,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec if !isGrokVideoUsageResult(result, nil) { ApplyOpenAIImageBillingResolution(result) } + logServiceTierBillingDowngrade("service.openai_gateway", account, result.RequestID, ApplyOpenAIServiceTierBillingResolution(result)) // OpenAI input_tokens 是总输入，包含缓存读取和缓存写入明细。 // 将三类 token 拆成互斥桶，避免缓存写入同时按普通输入和 cache_write 重复计费。 diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index 2974964f96..124f8d76a7 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -1163,6 +1163,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( UpstreamModel: mappedModel, UpstreamResponseModel: responseModelObserver.Model(), UpstreamResponseModelConflict: responseModelObserver.Conflict(), + UpstreamResponseServiceTier: responseModelObserver.ServiceTier(), ServiceTier: extractOpenAIServiceTierFromBody(payload), ReasoningEffort: ApplyThinkingEnabledFallback(extractOpenAIReasoningEffortFromBody(payload, mappedModel, originalModel), payload, mappedModel), Stream: reqStream, diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 4ecd73d09a..dcd5989c99 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -773,6 +773,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( UpstreamModel: mappedModel, UpstreamResponseModel: responseModelObserver.Model(), UpstreamResponseModelConflict: responseModelObserver.Conflict(), + UpstreamResponseServiceTier: responseModelObserver.ServiceTier(), ImageCount: imageCounter.Count(), ImageOutputSizes: imageCounter.Sizes(), ServiceTier: extractOpenAIServiceTier(reqBody), diff --git a/backend/internal/service/openai_ws_http_bridge.go b/backend/internal/service/openai_ws_http_bridge.go index 7a43006615..962095b407 100644 --- a/backend/internal/service/openai_ws_http_bridge.go +++ b/backend/internal/service/openai_ws_http_bridge.go @@ -513,6 +513,7 @@ func (s *OpenAIGatewayService) proxyOpenAIWSHTTPBridgeTurn( UpstreamModel: mappedModel, UpstreamResponseModel: responseModelObserver.Model(), UpstreamResponseModelConflict: responseModelObserver.Conflict(), + UpstreamResponseServiceTier: responseModelObserver.ServiceTier(), ServiceTier: extractOpenAIServiceTierFromBody(body), ReasoningEffort: ApplyThinkingEnabledFallback(extractOpenAIReasoningEffortFromBody(body, mappedModel, originalModel), body, mappedModel), Stream: reqStream, diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay.go b/backend/internal/service/openai_ws_v2/passthrough_relay.go index 67c26f3e96..e28223e091 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay.go @@ -33,9 +33,12 @@ type Usage struct { } type RelayResult struct { - RequestModel string - ResponseModel string - ResponseModelConflict bool + RequestModel string + ResponseModel string + ResponseModelConflict bool + // ResponseServiceTier is the raw service_tier declared by the last terminal + // response event; "" when the upstream never declared one. + ResponseServiceTier string Usage Usage RequestID string TerminalEventType string @@ -50,6 +53,7 @@ type RelayTurnResult struct { RequestModel string ResponseModel string ResponseModelConflict bool + ResponseServiceTier string Usage Usage RequestID string TerminalEventType string @@ -96,19 +100,20 @@ type RelayTraceEvent struct { } type relayState struct { - usage Usage - turnUsage Usage - requestModelMu sync.RWMutex - requestModel string - pendingTurnStart atomic.Pointer[time.Time] - lastResponseID string - lastResponseModel string - responseConflict bool - terminalEventType string - firstTokenMs *int - turnTimingByID map[string]*relayTurnTiming - activeTurn *relayTurnTiming - pendingBareError *observedUpstreamEvent + usage Usage + turnUsage Usage + requestModelMu sync.RWMutex + requestModel string + pendingTurnStart atomic.Pointer[time.Time] + lastResponseID string + lastResponseModel string + lastResponseServiceTier string + responseConflict bool + terminalEventType string + firstTokenMs *int + turnTimingByID map[string]*relayTurnTiming + activeTurn *relayTurnTiming + pendingBareError *observedUpstreamEvent } type relayExitSignal struct { @@ -119,15 +124,16 @@ type relayExitSignal struct { } type observedUpstreamEvent struct { - terminal bool - eventType string - responseID string - usage Usage - startedAt time.Time - responseModel string - responseConflict bool - duration time.Duration - firstToken *int + terminal bool + eventType string + responseID string + usage Usage + startedAt time.Time + responseModel string + responseConflict bool + responseServiceTier string + duration time.Duration + firstToken *int } type relayTurnTiming struct { @@ -136,6 +142,9 @@ type relayTurnTiming struct { firstResponseModel string terminalResponseModel string responseModelConflict bool + // terminalResponseServiceTier is only taken from terminal events: earlier + // events echo the requested tier, not the one the upstream actually used. + terminalResponseServiceTier string } func Relay( @@ -757,6 +766,7 @@ func observeUpstreamMessage( if !isTerminalEvent(eventType) { return observed } + observeRelayTurnResponseServiceTier(turnTiming, firstRelayResponseServiceTier(message)) state.terminalEventType = eventType if eventType == "error" { // Some Responses servers emit error immediately before response.failed. @@ -815,8 +825,10 @@ func finalizeObservedRelayTerminal(state *relayState, observed observedUpstreamE if turnTiming, ok := openAIWSRelayDeleteTurnTiming(state, responseID); ok { observed.responseModel = relayTurnResponseModel(&turnTiming) observed.responseConflict = turnTiming.responseModelConflict + observed.responseServiceTier = turnTiming.terminalResponseServiceTier state.lastResponseModel = observed.responseModel state.responseConflict = observed.responseConflict + state.lastResponseServiceTier = observed.responseServiceTier duration := now.Sub(turnTiming.startAt) if duration < 0 { duration = 0 @@ -852,6 +864,7 @@ func emitTurnComplete( RequestModel: requestModel, ResponseModel: observed.responseModel, ResponseModelConflict: observed.responseConflict, + ResponseServiceTier: observed.responseServiceTier, Usage: observed.usage, RequestID: responseID, TerminalEventType: observed.eventType, @@ -908,6 +921,31 @@ func relayTurnResponseModel(turn *relayTurnTiming) string { return turn.firstResponseModel } +func firstRelayResponseServiceTier(message []byte) string { + if len(message) == 0 { + return "" + } + values := gjson.GetManyBytes(message, "response.service_tier", "service_tier") + for _, value := range values { + if value.Type != gjson.String { + continue + } + if tier := strings.TrimSpace(value.String()); tier != "" { + return tier + } + } + return "" +} + +func observeRelayTurnResponseServiceTier(turn *relayTurnTiming, tier string) { + if turn == nil { + return + } + if tier = strings.TrimSpace(tier); tier != "" { + turn.terminalResponseServiceTier = tier + } +} + func openAIWSRelayGetOrInitTurnTiming(state *relayState, responseID string, now time.Time) *relayTurnTiming { if state == nil { return nil @@ -1165,6 +1203,7 @@ func enrichResult(result *RelayResult, state *relayState, duration time.Duration result.RequestModel = state.currentRequestModel() result.ResponseModel = state.lastResponseModel result.ResponseModelConflict = state.responseConflict + result.ResponseServiceTier = state.lastResponseServiceTier result.Usage = state.usage result.RequestID = state.lastResponseID result.TerminalEventType = state.terminalEventType diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go index cd02c3858f..80fef47d44 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go @@ -744,3 +744,54 @@ func TestObserveUpstreamMessage_ResponseIDFallbackPolicy(t *testing.T) { require.True(t, observed.terminal) require.Equal(t, "resp_fallback", observed.responseID) } + +func TestObserveUpstreamMessage_ResponseServiceTierOnlyFromTerminalEvents(t *testing.T) { + t.Parallel() + + state := &relayState{requestModel: "gpt-5.6-sol"} + startAt := time.Unix(0, 0) + now := startAt + nowFn := func() time.Time { + now = now.Add(5 * time.Millisecond) + return now + } + + created := observeUpstreamMessage( + state, + []byte(`{"type":"response.created","response":{"id":"resp_1","model":"gpt-5.6-sol","service_tier":"priority"}}`), + startAt, + nowFn, + nil, + ) + require.False(t, created.terminal) + require.Equal(t, "", created.responseServiceTier) + + completed := observeUpstreamMessage( + state, + []byte(`{"type":"response.completed","response":{"id":"resp_1","model":"gpt-5.6-sol","service_tier":"default","usage":{"input_tokens":1,"output_tokens":2}}}`), + startAt, + nowFn, + nil, + ) + require.True(t, completed.terminal) + require.Equal(t, "default", completed.responseServiceTier, "the echoed priority tier must lose to the terminal declaration") + + var turn RelayTurnResult + emitTurnComplete(func(result RelayTurnResult) { turn = result }, state, completed) + require.Equal(t, "default", turn.ResponseServiceTier) + + var result RelayResult + enrichResult(&result, state, now.Sub(startAt)) + require.Equal(t, "default", result.ResponseServiceTier) + + // A later turn without any declaration must not inherit the previous one. + observeUpstreamMessage(state, []byte(`{"type":"response.created","response":{"id":"resp_2","model":"gpt-5.6-sol"}}`), startAt, nowFn, nil) + second := observeUpstreamMessage( + state, + []byte(`{"type":"response.completed","response":{"id":"resp_2","model":"gpt-5.6-sol","usage":{"input_tokens":3,"output_tokens":4}}}`), + startAt, + nowFn, + nil, + ) + require.Equal(t, "", second.responseServiceTier) +} diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 27cd4b44c3..c8fcd87525 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -1181,6 +1181,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( UpstreamModel: openAIWSDifferentModel(turnRequestModel, turnUpstreamModel), UpstreamResponseModel: turn.ResponseModel, UpstreamResponseModelConflict: turn.ResponseModelConflict, + UpstreamResponseServiceTier: normalizeObservedOpenAIServiceTier(turn.ResponseServiceTier), ServiceTier: usageMeta.serviceTier.Load(), ReasoningEffort: usageMeta.reasoningEffort.Load(), Stream: true, @@ -1306,6 +1307,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( UpstreamModel: openAIWSDifferentModel(resultRequestModel, resultUpstreamModel), UpstreamResponseModel: relayResult.ResponseModel, UpstreamResponseModelConflict: relayResult.ResponseModelConflict, + UpstreamResponseServiceTier: normalizeObservedOpenAIServiceTier(relayResult.ResponseServiceTier), ServiceTier: usageMeta.serviceTier.Load(), ReasoningEffort: usageMeta.reasoningEffort.Load(), Stream: true, diff --git a/backend/internal/service/service_tier_billing.go b/backend/internal/service/service_tier_billing.go new file mode 100644 index 0000000000..74cce9146e --- /dev/null +++ b/backend/internal/service/service_tier_billing.go @@ -0,0 +1,107 @@ +package service + +import ( + "log/slog" + "strings" +) + +// ServiceTierBillingResolution describes how the billable service tier of one +// request was settled between the tier the client asked for and the tier the +// upstream reports having used. +type ServiceTierBillingResolution struct { + Requested string // tier carried by the request sent upstream ("" when none) + Observed string // tier declared by the upstream response ("" when none) + Billing string // tier used for billing and the usage log + Downgraded bool // Billing is cheaper than Requested +} + +// ResolveBillingServiceTier picks the tier to bill. The upstream declaration is +// trusted only to lower the bill: a request served below the tier it asked for +// (OpenAI "priority" answered with service_tier "default", Anthropic speed=fast +// answered with usage.speed "standard") is billed at the cheaper tier. A response +// that claims a more expensive tier, an unknown one, or none at all leaves the +// requested tier in place, which is also the behaviour for upstreams that never +// declare a tier. +func ResolveBillingServiceTier(requested, observed string) ServiceTierBillingResolution { + requested = normalizeBillingServiceTier(requested) + observed = normalizeBillingServiceTier(observed) + resolution := ServiceTierBillingResolution{Requested: requested, Observed: observed, Billing: requested} + if observed == "" || observed == requested { + return resolution + } + observedRank, known := serviceTierCostRank(observed) + if !known { + return resolution + } + requestedRank, _ := serviceTierCostRank(requested) + if observedRank >= requestedRank { + return resolution + } + resolution.Billing = observed + resolution.Downgraded = true + return resolution +} + +// serviceTierCostRank orders tiers by their cost relative to the base rate, so a +// lower rank is always cheaper. Unknown tiers rank as the base rate and report +// known=false so callers can refuse to act on them. +func serviceTierCostRank(tier string) (rank int, known bool) { + switch normalizeBillingServiceTier(tier) { + case "flex": + return 0, true + case "", "default", "standard", "auto", "scale": + return 1, true + case "priority", "fast": + return 2, true + default: + return 1, false + } +} + +// ApplyOpenAIServiceTierBillingResolution lowers result.ServiceTier to the tier +// the upstream reports having used, so cost calculation and the usage log share +// one billable tier. The returned resolution is meant for the audit log. +func ApplyOpenAIServiceTierBillingResolution(result *OpenAIForwardResult) ServiceTierBillingResolution { + if result == nil { + return ServiceTierBillingResolution{} + } + resolution := ResolveBillingServiceTier(optionalStringValue(result.ServiceTier), result.UpstreamResponseServiceTier) + if resolution.Downgraded { + billing := resolution.Billing + result.ServiceTier = &billing + } + return resolution +} + +// ApplyForwardServiceTierBillingResolution is the ForwardResult counterpart of +// ApplyOpenAIServiceTierBillingResolution. +func ApplyForwardServiceTierBillingResolution(result *ForwardResult) ServiceTierBillingResolution { + if result == nil { + return ServiceTierBillingResolution{} + } + resolution := ResolveBillingServiceTier(optionalStringValue(result.ServiceTier), result.UpstreamResponseServiceTier) + if resolution.Downgraded { + billing := resolution.Billing + result.ServiceTier = &billing + } + return resolution +} + +// logServiceTierBillingDowngrade leaves an audit trail for every request billed +// below the tier it asked for; unchanged tiers are not logged. +func logServiceTierBillingDowngrade(component string, account *Account, requestID string, resolution ServiceTierBillingResolution) { + if !resolution.Downgraded { + return + } + attrs := []any{ + "component", component, + "request_id", strings.TrimSpace(requestID), + "requested_tier", resolution.Requested, + "response_tier", resolution.Observed, + "billed_tier", resolution.Billing, + } + if account != nil { + attrs = append(attrs, "platform", account.Platform, "account_id", account.ID) + } + slog.Info("billing.service_tier_downgraded", attrs...) +} diff --git a/backend/internal/service/service_tier_billing_test.go b/backend/internal/service/service_tier_billing_test.go new file mode 100644 index 0000000000..fc8242fa2b --- /dev/null +++ b/backend/internal/service/service_tier_billing_test.go @@ -0,0 +1,71 @@ +package service + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestResolveBillingServiceTier(t *testing.T) { + tests := []struct { + name string + requested string + observed string + billing string + downgraded bool + }{ + {name: "openai priority served as default", requested: "priority", observed: "default", billing: "default", downgraded: true}, + {name: "anthropic fast served as standard", requested: "fast", observed: "standard", billing: "standard", downgraded: true}, + {name: "priority honoured", requested: "priority", observed: "priority", billing: "priority"}, + {name: "no declaration keeps request", requested: "priority", observed: "", billing: "priority"}, + {name: "no request no declaration", requested: "", observed: "", billing: ""}, + {name: "response never raises the tier", requested: "", observed: "priority", billing: ""}, + {name: "flex never raised to default", requested: "flex", observed: "default", billing: "flex"}, + {name: "default echoed for untiered request", requested: "", observed: "default", billing: ""}, + {name: "unknown response tier ignored", requested: "priority", observed: "turbo", billing: "priority"}, + {name: "case and whitespace normalised", requested: " Priority ", observed: "DEFAULT", billing: "default", downgraded: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ResolveBillingServiceTier(tt.requested, tt.observed) + require.Equal(t, tt.billing, got.Billing) + require.Equal(t, tt.downgraded, got.Downgraded) + }) + } +} + +func TestApplyServiceTierBillingResolutionOnlyRewritesDowngrades(t *testing.T) { + t.Run("openai downgrade rewrites tier", func(t *testing.T) { + requested := "priority" + result := &OpenAIForwardResult{ServiceTier: &requested, UpstreamResponseServiceTier: "default"} + resolution := ApplyOpenAIServiceTierBillingResolution(result) + require.True(t, resolution.Downgraded) + require.NotNil(t, result.ServiceTier) + require.Equal(t, "default", *result.ServiceTier) + }) + + t.Run("openai honoured tier keeps pointer", func(t *testing.T) { + requested := "priority" + result := &OpenAIForwardResult{ServiceTier: &requested, UpstreamResponseServiceTier: "priority"} + require.False(t, ApplyOpenAIServiceTierBillingResolution(result).Downgraded) + require.Same(t, &requested, result.ServiceTier) + }) + + t.Run("openai untiered request stays nil", func(t *testing.T) { + result := &OpenAIForwardResult{UpstreamResponseServiceTier: "priority"} + require.False(t, ApplyOpenAIServiceTierBillingResolution(result).Downgraded) + require.Nil(t, result.ServiceTier) + }) + + t.Run("anthropic standard speed rewrites fast", func(t *testing.T) { + requested := "fast" + result := &ForwardResult{ServiceTier: &requested, UpstreamResponseServiceTier: "standard"} + require.True(t, ApplyForwardServiceTierBillingResolution(result).Downgraded) + require.Equal(t, "standard", *result.ServiceTier) + }) + + t.Run("nil results are ignored", func(t *testing.T) { + require.False(t, ApplyOpenAIServiceTierBillingResolution(nil).Downgraded) + require.False(t, ApplyForwardServiceTierBillingResolution(nil).Downgraded) + }) +} diff --git a/backend/internal/service/upstream_response_model.go b/backend/internal/service/upstream_response_model.go index b7c304e488..eb52a76d1f 100644 --- a/backend/internal/service/upstream_response_model.go +++ b/backend/internal/service/upstream_response_model.go @@ -20,10 +20,21 @@ const ( // channel explicitly configured with billing_model_source = response_model, // where a conflict flag makes billing fall back to the baseline model // (see responseModelBillingDeclaration). +// +// The same observer also records the service tier the upstream reports having +// used (OpenAI service_tier, Anthropic usage.speed). Billing consumes it through +// ResolveBillingServiceTier, which only ever lowers the tier a request asked for. type upstreamResponseModelObserver struct { first string terminal string conflict bool + + // firstTier holds the first non-terminal tier declaration; it is discarded + // when later non-terminal declarations disagree. terminalTier comes from a + // terminal event and always wins. + firstTier string + firstTierConflict bool + terminalTier string } func (o *upstreamResponseModelObserver) Observe(model string, terminal bool) { @@ -57,17 +68,98 @@ func normalizeObservedUpstreamResponseModel(model string) string { } func (o *upstreamResponseModelObserver) ObserveOpenAI(payload []byte, eventType string) { - model := firstValidTrimmedGJSONModel(payload, "response.model", "model") - o.Observe(model, isUpstreamResponseModelTerminalEvent(eventType)) + model := firstValidTrimmedGJSONString(payload, "response.model", "model") + terminal := isUpstreamResponseModelTerminalEvent(eventType) + o.Observe(model, terminal) + // Every payload that declares a service tier also declares a model, so + // model-free delta frames skip the extra lookups entirely. + if model == "" { + return + } + // Non-terminal Responses API events echo the requested tier rather than the + // tier actually used. Only terminal events and untyped payloads (chat + // completions chunks, non-streaming bodies) report the processing tier. + if !terminal && strings.TrimSpace(eventType) != "" { + return + } + tier := normalizeObservedOpenAIServiceTier(firstValidTrimmedGJSONString(payload, "response.service_tier", "service_tier")) + o.ObserveServiceTier(tier, terminal) } func (o *upstreamResponseModelObserver) ObserveAnthropic(payload []byte) { - model := firstValidTrimmedGJSONModel(payload, "message.model", "model") + model := firstValidTrimmedGJSONString(payload, "message.model", "model") o.Observe(model, false) + // usage.speed travels with the message object (message_start in streams, + // the top-level body otherwise), i.e. only in payloads that declare a model. + if model == "" { + return + } + tier := normalizeObservedAnthropicSpeed(firstValidTrimmedGJSONString(payload, "message.usage.speed", "usage.speed")) + o.ObserveServiceTier(tier, false) +} + +// ObserveServiceTier records a tier declared by the upstream response. A +// terminal declaration always wins; non-terminal declarations are only trusted +// when they agree with each other. +func (o *upstreamResponseModelObserver) ObserveServiceTier(tier string, terminal bool) { + if o == nil || tier == "" { + return + } + if terminal { + o.terminalTier = tier + return + } + if o.firstTier == "" { + o.firstTier = tier + return + } + if o.firstTier != tier { + o.firstTierConflict = true + } +} + +// ServiceTier returns the tier the upstream reports having used, or "" when the +// response never declared one unambiguously. +func (o *upstreamResponseModelObserver) ServiceTier() string { + if o == nil { + return "" + } + if o.terminalTier != "" { + return o.terminalTier + } + if o.firstTierConflict { + return "" + } + return o.firstTier +} + +// normalizeObservedOpenAIServiceTier maps a tier reported by an OpenAI response +// onto the billing vocabulary. "auto" never describes a processing tier and +// unknown values are ignored rather than guessed at. +func normalizeObservedOpenAIServiceTier(raw string) string { + switch value := strings.ToLower(strings.TrimSpace(raw)); value { + case "priority", "fast": + return OpenAIFastTierPriority + case "default", "flex", "scale": + return value + default: + return "" + } +} + +// normalizeObservedAnthropicSpeed maps Anthropic usage.speed onto the billing +// vocabulary: "fast" is the billable fast-mode tier, "standard" the base rate. +func normalizeObservedAnthropicSpeed(raw string) string { + switch value := strings.ToLower(strings.TrimSpace(raw)); value { + case "fast", "standard": + return value + default: + return "" + } } func (o *upstreamResponseModelObserver) ObserveGemini(payload []byte) { - model := firstValidTrimmedGJSONModel( + model := firstValidTrimmedGJSONString( payload, "modelVersion", "response.modelVersion", @@ -120,6 +212,10 @@ func observedUpstreamResponseModelConflict(c *gin.Context) bool { return upstreamResponseModelObserverFromContext(c).Conflict() } +func observedUpstreamResponseServiceTier(c *gin.Context) string { + return upstreamResponseModelObserverFromContext(c).ServiceTier() +} + func observeOpenAISSEBody(observer *upstreamResponseModelObserver, body string) { if observer == nil || strings.TrimSpace(body) == "" { return @@ -129,7 +225,7 @@ func observeOpenAISSEBody(observer *upstreamResponseModelObserver, body string) }) } -func firstValidTrimmedGJSONModel(payload []byte, paths ...string) string { +func firstValidTrimmedGJSONString(payload []byte, paths ...string) string { if len(payload) == 0 { return "" } @@ -138,14 +234,14 @@ func firstValidTrimmedGJSONModel(payload []byte, paths ...string) string { if !value.Exists() || value.Type != gjson.String { continue } - if model := strings.TrimSpace(value.String()); model != "" { + if text := strings.TrimSpace(value.String()); text != "" { // Validate only after finding a candidate. This avoids a full validation // pass on the common model-free delta path while still rejecting malformed - // payloads that appear to declare a model. + // payloads that appear to declare a value. if !gjson.ValidBytes(payload) { return "" } - return model + return text } } return "" diff --git a/backend/internal/service/upstream_response_model_bench_test.go b/backend/internal/service/upstream_response_model_bench_test.go index 93d6854f72..ec56fe4b46 100644 --- a/backend/internal/service/upstream_response_model_bench_test.go +++ b/backend/internal/service/upstream_response_model_bench_test.go @@ -35,7 +35,7 @@ func BenchmarkUpstreamResponseModelOpenAI(b *testing.B) { b.Run("optimized", func(b *testing.B) { b.ReportAllocs() for b.Loop() { - upstreamResponseModelBenchmarkSink = firstValidTrimmedGJSONModel(tt.payload, "response.model", "model") + upstreamResponseModelBenchmarkSink = firstValidTrimmedGJSONString(tt.payload, "response.model", "model") } }) }) @@ -52,7 +52,7 @@ func BenchmarkUpstreamResponseModelAntigravityWrapper(b *testing.B) { b.Run("optimized", func(b *testing.B) { b.ReportAllocs() for b.Loop() { - upstreamResponseModelBenchmarkSink = firstValidTrimmedGJSONModel( + upstreamResponseModelBenchmarkSink = firstValidTrimmedGJSONString( upstreamResponseModelBenchmarkWrapper, "modelVersion", "response.modelVersion", diff --git a/backend/internal/service/upstream_response_model_test.go b/backend/internal/service/upstream_response_model_test.go index 51374ec1f3..ea616d53ee 100644 --- a/backend/internal/service/upstream_response_model_test.go +++ b/backend/internal/service/upstream_response_model_test.go @@ -194,3 +194,79 @@ func TestUpstreamResponseModelObserverBoundsUntrustedModelName(t *testing.T) { require.Len(t, []rune(observer.Model()), upstreamResponseModelMaxLength) } + +func TestUpstreamResponseModelObserverServiceTierTerminalEventWins(t *testing.T) { + observer := &upstreamResponseModelObserver{} + + // response.created echoes the requested tier and must not count as a declaration. + observer.ObserveOpenAI([]byte(`{"type":"response.created","response":{"id":"resp_1","model":"gpt-5.6-sol","service_tier":"priority"}}`), "response.created") + require.Equal(t, "", observer.ServiceTier()) + + observer.ObserveOpenAI([]byte(`{"type":"response.completed","response":{"id":"resp_1","model":"gpt-5.6-sol","service_tier":"default","usage":{"input_tokens":1,"output_tokens":2}}}`), "response.completed") + require.Equal(t, "default", observer.ServiceTier()) + require.False(t, observer.Conflict(), "tier echo must not be reported as a model conflict") +} + +func TestUpstreamResponseModelObserverServiceTierUntypedPayloads(t *testing.T) { + t.Run("chat completions chunks agree", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveOpenAI([]byte(`{"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-5.4","service_tier":"default","choices":[]}`), "") + observer.ObserveOpenAI([]byte(`{"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-5.4","service_tier":"default","choices":[]}`), "") + require.Equal(t, "default", observer.ServiceTier()) + }) + + t.Run("disagreeing chunks are not trusted", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveOpenAI([]byte(`{"model":"gpt-5.4","service_tier":"priority"}`), "") + observer.ObserveOpenAI([]byte(`{"model":"gpt-5.4","service_tier":"default"}`), "") + require.Equal(t, "", observer.ServiceTier()) + }) + + t.Run("non-stream body normalises fast and drops auto", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveOpenAI([]byte(`{"id":"resp_1","object":"response","model":"gpt-5.6-sol","service_tier":"fast"}`), "") + require.Equal(t, "priority", observer.ServiceTier()) + + auto := &upstreamResponseModelObserver{} + auto.ObserveOpenAI([]byte(`{"id":"resp_2","object":"response","model":"gpt-5.6-sol","service_tier":"auto"}`), "") + require.Equal(t, "", auto.ServiceTier()) + }) + + t.Run("model-free deltas never declare a tier", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveOpenAI([]byte(`{"type":"response.output_text.delta","delta":"hi","service_tier":"default"}`), "response.output_text.delta") + require.Equal(t, "", observer.ServiceTier()) + }) +} + +func TestUpstreamResponseModelObserverServiceTierAnthropicSpeed(t *testing.T) { + t.Run("message_start reports standard", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveAnthropic([]byte(`{"type":"message_start","message":{"id":"msg_1","model":"claude-opus-5","usage":{"input_tokens":8,"output_tokens":1,"speed":"standard"}}}`)) + observer.ObserveAnthropic([]byte(`{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":12}}`)) + require.Equal(t, "standard", observer.ServiceTier()) + require.Equal(t, "claude-opus-5", observer.Model()) + }) + + t.Run("non-stream body reports fast", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveAnthropic([]byte(`{"id":"msg_1","type":"message","model":"claude-opus-5","usage":{"input_tokens":8,"output_tokens":12,"speed":"fast"}}`)) + require.Equal(t, "fast", observer.ServiceTier()) + }) + + t.Run("missing speed declares nothing", func(t *testing.T) { + observer := &upstreamResponseModelObserver{} + observer.ObserveAnthropic([]byte(`{"id":"msg_1","type":"message","model":"claude-opus-5","usage":{"input_tokens":8,"output_tokens":12}}`)) + require.Equal(t, "", observer.ServiceTier()) + }) +} + +func TestObservedUpstreamResponseServiceTierFromContext(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(nil) + require.Equal(t, "", observedUpstreamResponseServiceTier(c)) + + observer := beginUpstreamResponseModelObservation(c) + observer.ObserveOpenAI([]byte(`{"type":"response.completed","response":{"id":"resp_1","model":"gpt-5.6-sol","service_tier":"default"}}`), "response.completed") + require.Equal(t, "default", observedUpstreamResponseServiceTier(c)) +}