mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-08-31 01:13:06 +08:00
Merge pull request #6111 from feeeei/fix/request_billing
fix(billing): bill fast mode by the tier upstream actually served
This commit is contained in:
Wesley Liddick
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
// 图片生成计费字段（图片生成模型使用）
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
// 用于粘性会话切换时的特殊计费处理
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 重复计费。
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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...)
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
@@ -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 ""
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user