seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 12:02:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(billing): preserve final reasoning effort across forwarding paths

Browse Source
This commit is contained in:
haruka
2026-09-21 19:58:02 +08:00
parent b8d52fad34
commit e47255715d
12 changed files with 552 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/antigravity_gateway_gemini.go
+1
View File
@@ -467,6 +467,7 @@ handleSuccess:
UpstreamResponseModel: observedUpstreamResponseModel(c),
UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c),
Stream: stream,
ReasoningEffort: extractGeminiReasoningEffortFromBody(injectedBody),
Duration: time.Since(startTime),
FirstTokenMs: firstTokenMs,
ClientDisconnect: clientDisconnect,
backend/internal/service/gateway_image_reasoning_pricing_test.go
+143
View File
@@ -0,0 +1,143 @@
//go:build unit
package service
import (
"context"
"fmt"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestRecordUsage_ImageReasoningPricing(t *testing.T) {
for _, platform := range []string{PlatformAnthropic, PlatformOpenAI} {
for _, source := range []string{PricingSourceChannel, PricingSourceGroup} {
for _, mode := range []BillingMode{BillingModeImage, BillingModePerRequest} {
for _, independent := range []bool{false, true} {
for _, effort := range []string{"high", "low", ""} {
name := fmt.Sprintf("%s/%s/%s/independent=%t/effort=%s", platform, source, mode, independent, effort)
t.Run(name, func(t *testing.T) {
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
userRepo := &openAIRecordUsageUserRepoStub{}
subRepo := &openAIRecordUsageSubRepoStub{}
groupID := int64(77)
model := "gemini-3-pro-image-preview"
price := 0.25
pricing := ChannelModelPricing{
Models: []string{model}, BillingMode: mode, PerRequestPrice: &price,
ReasoningEffortMultipliers: map[string]float64{"high": 2},
}
resolver := newOpenAIImageChannelPricingResolverForTest(t, groupID, model, price)
cache := resolver.channelService.cache.Load().(*channelCache)
cache.pricingByGroupModel[channelModelKey{groupID: groupID, model: model}] = &pricing
group := &Group{
ID: groupID, Platform: platform, Status: StatusActive, Hydrated: true,
RateMultiplier: 0.5, ImageRateIndependent: independent, ImageRateMultiplier: 0.25,
}
if source == PricingSourceGroup {
group.ModelPricing = []ChannelModelPricing{pricing}
// A matching group card owns the multiplier; the channel must not stack on top.
pricing.ReasoningEffortMultipliers = map[string]float64{"high": 3}
}
apiKey := &APIKey{ID: 10, GroupID: &groupID, Group: group}
account := &Account{ID: 30, Platform: platform, Type: AccountTypeAPIKey}
user := &User{ID: 20}
var forwardedEffort *string
if effort != "" {
forwardedEffort = &effort
}
if platform == PlatformAnthropic {
svc := newGatewayRecordUsageServiceForTest(usageRepo, userRepo, subRepo)
svc.resolver = resolver
require.NoError(t, svc.RecordUsage(context.Background(), &RecordUsageInput{
Result: &ForwardResult{
RequestID: "image_reasoning_pricing", Model: model, Duration: time.Second,
ImageCount: 2, ImageSize: "1K", ReasoningEffort: forwardedEffort,
},
APIKey: apiKey, User: user, Account: account,
}))
} else {
svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, subRepo, nil)
svc.resolver = resolver
require.NoError(t, svc.RecordUsage(context.Background(), &OpenAIRecordUsageInput{
Result: &OpenAIForwardResult{
RequestID: "image_reasoning_pricing", Model: model, Duration: time.Second,
ImageCount: 2, ImageSize: "1K", ReasoningEffort: forwardedEffort,
},
APIKey: apiKey, User: user, Account: account,
}))
}
wantTotal := 0.5
if effort == "high" {
wantTotal *= 2
}
wantRate := 0.5
if independent {
wantRate = 0.25
}
require.NotNil(t, usageRepo.lastLog)
require.InDelta(t, wantTotal, usageRepo.lastLog.TotalCost, 1e-12)
require.InDelta(t, wantTotal*wantRate, usageRepo.lastLog.ActualCost, 1e-12)
require.InDelta(t, wantTotal*wantRate, userRepo.lastAmount, 1e-12)
require.Equal(t, forwardedEffort, usageRepo.lastLog.ReasoningEffort)
require.Equal(t, 2, usageRepo.lastLog.ImageCount)
require.Equal(t, string(mode), *usageRepo.lastLog.BillingMode)
})
}
}
}
}
}
}
func TestCalculateRecordUsageCost_MediaReasoningPricing(t *testing.T) {
for _, source := range []string{PricingSourceChannel, PricingSourceGroup} {
for _, media := range []string{"gateway_audio", "openai_audio", "openai_video"} {
t.Run(source+"/"+media, func(t *testing.T) {
groupID := int64(77)
model, effort := "media-model", "high"
price := 0.25
resolver := newOpenAIImageChannelPricingResolverForTest(t, groupID, model, price)
cache := resolver.channelService.cache.Load().(*channelCache)
pricing := cache.pricingByGroupModel[channelModelKey{groupID: groupID, model: model}]
pricing.BillingMode = BillingModePerRequest
pricing.ReasoningEffortMultipliers = map[string]float64{"high": 2}
wantTotal := 1.0 // Two units at $0.25, with high=2.
if media == "openai_video" {
pricing.BillingMode = BillingModeVideo
wantTotal = 5 // Two five-second videos at $0.25/second, with high=2.
}
apiKey := &APIKey{GroupID: &groupID, Group: &Group{ID: groupID, Hydrated: true}}
if source == PricingSourceGroup {
pricing.Models = []string{model}
apiKey.Group.ModelPricing = []ChannelModelPricing{*pricing}
pricing.ReasoningEffortMultipliers = map[string]float64{"high": 3}
}
var cost *CostBreakdown
if media == "gateway_audio" {
svc := &GatewayService{billingService: resolver.billingService, resolver: resolver}
cost = svc.calculateRecordUsageCost(context.Background(), &ForwardResult{
ReasoningEffort: &effort, AudioUsage: &AudioUsage{Mode: "tts", DurationOrUnits: 2},
}, apiKey, model, 0.5, 0.5, time.Time{})
} else {
svc := &OpenAIGatewayService{billingService: resolver.billingService, resolver: resolver}
result := &OpenAIForwardResult{ReasoningEffort: &effort}
if media == "openai_audio" {
result.AudioUsage = &AudioUsage{Mode: "tts", DurationOrUnits: 2}
} else {
result.VideoCount, result.VideoDurationSeconds = 2, 5
}
var err error
cost, err = svc.calculateOpenAIRecordUsageCost(context.Background(), result, apiKey,
[]string{model}, 0.5, 0.5, 0.5, 0.5, UsageTokens{}, "", nil, time.Time{})
require.NoError(t, err)
}
require.NotNil(t, cost)
require.InDelta(t, wantTotal, cost.TotalCost, 1e-12)
require.InDelta(t, wantTotal*0.5, cost.ActualCost, 1e-12)
})
}
}
}
backend/internal/service/gateway_usage_billing.go
+13 -10
View File
@@ -903,6 +903,7 @@ func (s *GatewayService) calculateRecordUsageCost(
Ctx: ctx, Model: billingModel, GroupID: &gid, Group: apiKey.Group,
UsageUnits: result.AudioUsage.DurationOrUnits, SizeTier: result.AudioUsage.Mode,
RateMultiplier: multiplier, Resolver: s.resolver, Resolved: resolved,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
})
if err == nil {
return cost
@@ -1027,6 +1028,7 @@ func (s *GatewayService) calculateImageCost(
Ctx: ctx, Model: billingModel, GroupID: &gid, Group: apiKey.Group,
RequestCount: result.ImageCount, SizeTier: sizeTier,
RateMultiplier: multiplier, Resolver: s.resolver, Resolved: resolved,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
})
if err == nil {
return cost
@@ -1044,16 +1046,17 @@ func (s *GatewayService) calculateImageCost(
}
gid := apiKey.Group.ID
cost, err := s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
Tokens: tokens,
RequestCount: result.ImageCount,
SizeTier: sizeTier,
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
Tokens: tokens,
RequestCount: result.ImageCount,
SizeTier: sizeTier,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
})
if err != nil {
logger.LegacyPrintf("service.gateway", "Calculate image token cost failed: %v", err)
backend/internal/service/gemini_chat_completions_compat_service.go
+2 -6
View File
@@ -58,7 +58,7 @@ func (s *GeminiMessagesCompatService) ForwardAsChatCompletions(
return nil, fmt.Errorf("marshal chat completions compat request: %w", err)
}
return s.forwardClaudeBodyAsChatCompletions(ctx, c, account, claudeBody, originalModel, clientStream, includeUsage, startTime, body)
return s.forwardClaudeBodyAsChatCompletions(ctx, c, account, claudeBody, originalModel, clientStream, includeUsage, startTime)
}
func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
@@ -70,7 +70,6 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
clientStream bool,
includeUsage bool,
startTime time.Time,
originalChatBody []byte,
) (*ForwardResult, error) {
var req struct {
Model string `json:"model"`
@@ -208,10 +207,7 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
c.Header("x-request-id", requestID)
}
reasoningEffort := extractCCReasoningEffortFromBody(originalChatBody, mappedModel)
// 国产模型默认 effort 补充（本路径上游是 Gemini，不会命中 passback-required）。
// 保持与 OpenAI 网关路径调用模式一致，便于未来上游变异时语义一致。
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, originalChatBody, mappedModel)
reasoningEffort := extractGeminiReasoningEffortFromBody(geminiReq)
if resp.StatusCode >= 400 {
respBody := s.readUpstreamErrorBody(resp)
backend/internal/service/gemini_messages_compat_service.go
+1
View File
@@ -1668,6 +1668,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
UpstreamResponseModel: observedUpstreamResponseModel(c),
UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c),
Stream: stream,
ReasoningEffort: extractGeminiReasoningEffortFromBody(body),
Duration: time.Since(startTime),
FirstTokenMs: firstTokenMs,
ImageCount: imageCount,
backend/internal/service/gemini_native_reasoning_pricing_test.go
+190
View File
@@ -0,0 +1,190 @@
//go:build unit
package service
import (
"context"
"encoding/json"
"io"
"net/http"
"strings"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
func TestGeminiNativeReasoningPricingUsesExplicitForwardedLevel(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, provider := range []string{"api_key", "oauth_ai_studio", "oauth_code_assist", "antigravity"} {
for _, stream := range []bool{false, true} {
for _, tc := range []struct {
name string
thinking map[string]any
wantEffort string
multiplier float64
}{
{"high", map[string]any{"thinkingLevel": "high"}, "high", 2},
{"uppercase_high", map[string]any{"thinkingLevel": "HIGH"}, "high", 2},
{"snake_case_high", map[string]any{"thinking_level": "high"}, "high", 2},
{"minimal", map[string]any{"thinkingLevel": "minimal"}, "minimal", 1.1},
{"low", map[string]any{"thinkingLevel": "low"}, "low", 1.2},
{"medium", map[string]any{"thinkingLevel": "medium"}, "medium", 1.5},
{"missing_level", nil, "", 1},
{"budget_only", map[string]any{"thinkingBudget": 24576}, "", 1},
{"unspecified_level", map[string]any{"thinkingLevel": "THINKING_LEVEL_UNSPECIFIED"}, "", 1},
} {
mode := "buffered"
if stream {
mode = "streaming"
}
t.Run(provider+"/"+mode+"/"+tc.name, func(t *testing.T) {
const model = "gemini-3.1-pro-high"
request := map[string]any{
"contents": []any{map[string]any{"role": "user", "parts": []any{map[string]string{"text": "hello"}}}},
}
if tc.name == "snake_case_high" {
request["generation_config"] = map[string]any{"thinking_config": tc.thinking}
} else if tc.thinking != nil {
request["generationConfig"] = map[string]any{"thinkingConfig": tc.thinking}
}
body, err := json.Marshal(request)
require.NoError(t, err)
action := "generateContent"
if stream {
action = "streamGenerateContent"
}
path := "/v1beta/models/" + model + ":" + action
c, recorder := newAntigravityCompatContext(http.MethodPost, path, body)
responseBody := `{"candidates":[{"content":{"parts":[{"text":"ok"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":5}}`
wrapped := provider == "oauth_code_assist" || provider == "antigravity"
if wrapped {
responseBody = `{"response":` + responseBody + `}`
}
contentType := "application/json"
if stream || wrapped {
contentType = "text/event-stream"
responseBody = "data: " + responseBody + "\n\n"
}
upstream := &queuedHTTPUpstreamStub{responses: []*http.Response{{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{contentType}},
Body: io.NopCloser(strings.NewReader(responseBody)),
}}}
account := geminiSignalTestAccount()
var result *ForwardResult
if provider == "antigravity" {
account = newAntigravityCompatAccount(AccountTypeOAuth)
svc := newAntigravityCompatService(config.GatewayConfig{}, upstream)
result, err = svc.ForwardGemini(context.Background(), c, account, model, action, stream, body, false)
} else {
if provider != "api_key" {
account.Type = AccountTypeOAuth
account.Credentials["access_token"] = "test-token"
}
if provider == "oauth_code_assist" {
account.Credentials["project_id"] = "test-project"
}
svc := &GeminiMessagesCompatService{
httpUpstream: upstream, cfg: &config.Config{}, tokenProvider: &GeminiTokenProvider{},
}
result, err = svc.ForwardNative(context.Background(), c, account, model, action, stream, body)
}
require.NoError(t, err)
require.Equal(t, http.StatusOK, recorder.Code)
require.Len(t, upstream.requestBodies, 1)
forwardedPath := "generationConfig.thinkingConfig.thinkingLevel"
requestedLevel, _ := tc.thinking["thinkingLevel"].(string)
if tc.name == "snake_case_high" {
forwardedPath = "generation_config.thinking_config.thinking_level"
requestedLevel, _ = tc.thinking["thinking_level"].(string)
}
if wrapped {
forwardedPath = "request." + forwardedPath
}
require.Equal(t, requestedLevel, gjson.GetBytes(upstream.requestBodies[0], forwardedPath).String())
require.Equal(t, tc.wantEffort, optionalStringValue(result.ReasoningEffort))
if tc.wantEffort == "" {
require.Nil(t, result.ReasoningEffort)
}
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
userRepo := &openAIRecordUsageUserRepoStub{}
billing := newGatewayRecordUsageServiceForTest(usageRepo, userRepo, &openAIRecordUsageSubRepoStub{})
billing.resolver = NewModelPricingResolver(nil, billing.billingService)
groupID := int64(77)
apiKey := &APIKey{ID: 10, GroupID: &groupID, Group: &Group{
ID: groupID, Platform: account.Platform, Status: StatusActive, Hydrated: true, RateMultiplier: 0.5,
ModelPricing: []ChannelModelPricing{{
Models: []string{model}, BillingMode: BillingModeToken,
InputPrice: testPtrFloat64(0.05), OutputPrice: testPtrFloat64(0.1),
ReasoningEffortMultipliers: map[string]float64{"minimal": 1.1, "low": 1.2, "medium": 1.5, "high": 2},
}},
}}
require.NoError(t, billing.RecordUsage(context.Background(), &RecordUsageInput{
Result: result, APIKey: apiKey, User: &User{ID: 20}, Account: account,
}))
require.NotNil(t, usageRepo.lastLog)
require.Equal(t, tc.wantEffort, optionalStringValue(usageRepo.lastLog.ReasoningEffort))
require.InDelta(t, tc.multiplier, usageRepo.lastLog.TotalCost, 1e-12)
require.InDelta(t, tc.multiplier*0.5, usageRepo.lastLog.ActualCost, 1e-12)
require.InDelta(t, tc.multiplier*0.5, userRepo.lastAmount, 1e-12)
})
}
}
}
}
func TestGeminiChatCompatReasoningPricingIgnoresUnforwardedEffort(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, stream := range []bool{false, true} {
mode := "buffered"
if stream {
mode = "streaming"
}
t.Run(mode, func(t *testing.T) {
const model = "gemini-3.1-pro"
body, err := json.Marshal(map[string]any{
"model": model, "stream": stream, "reasoning_effort": "high",
"messages": []map[string]string{{"role": "user", "content": "hello"}},
})
require.NoError(t, err)
responseBody := `{"candidates":[{"content":{"parts":[{"text":"ok"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":5}}`
contentType := "application/json"
if stream {
contentType = "text/event-stream"
responseBody = "data: " + responseBody + "\n\n"
}
upstream := &queuedHTTPUpstreamStub{responses: []*http.Response{{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{contentType}},
Body: io.NopCloser(strings.NewReader(responseBody)),
}}}
svc := &GeminiMessagesCompatService{httpUpstream: upstream, cfg: &config.Config{}}
c, recorder := newAntigravityCompatContext(http.MethodPost, "/v1/chat/completions", body)
result, err := svc.ForwardAsChatCompletions(context.Background(), c, geminiSignalTestAccount(), body)
require.NoError(t, err)
require.Equal(t, http.StatusOK, recorder.Code)
require.Len(t, upstream.requestBodies, 1)
require.False(t, gjson.GetBytes(upstream.requestBodies[0], "generationConfig.thinkingConfig").Exists())
require.Nil(t, result.ReasoningEffort)
billing := NewBillingService(&config.Config{}, nil)
cost, err := billing.CalculateTokenCostForRequest(TokenCostRequest{
Ctx: context.Background(), Model: model,
Group: &Group{ID: 77, Platform: PlatformGemini, ModelPricing: []ChannelModelPricing{{
Models: []string{model}, BillingMode: BillingModeToken,
InputPrice: testPtrFloat64(0.05), OutputPrice: testPtrFloat64(0.1),
ReasoningEffortMultipliers: map[string]float64{"high": 2},
}}},
Tokens: UsageTokens{InputTokens: result.Usage.InputTokens, OutputTokens: result.Usage.OutputTokens},
ReasoningEffort: optionalStringValue(result.ReasoningEffort), RateMultiplier: 1,
Resolver: NewModelPricingResolver(nil, billing),
})
require.NoError(t, err)
require.InDelta(t, 1, cost.TotalCost, 1e-12)
})
}
}
backend/internal/service/gemini_reasoning_effort.go
+32
View File
@@ -0,0 +1,32 @@
package service
import (
"strings"
"github.com/tidwall/gjson"
)
// extractGeminiReasoningEffortFromBody reads only an explicit level from the
// forwarded Gemini request. A budget or an omitted level does not define a
// billing level, even when the upstream model enables thinking by default.
func extractGeminiReasoningEffortFromBody(body []byte) *string {
config := gjson.GetBytes(body, "generationConfig")
if !config.Exists() {
config = gjson.GetBytes(body, "generation_config")
}
thinking := config.Get("thinkingConfig")
if !thinking.Exists() {
thinking = config.Get("thinking_config")
}
level := thinking.Get("thinkingLevel")
if !level.Exists() {
level = thinking.Get("thinking_level")
}
effort := strings.ToLower(strings.TrimSpace(level.String()))
switch effort {
case "minimal", "low", "medium", "high":
return &effort
default:
return nil
}
}
backend/internal/service/openai_gateway_chat_completions_anthropic_native.go
+5 -3
View File
@@ -23,6 +23,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/Wei-Shaw/sub2api/internal/util/responseheaders"
"github.com/gin-gonic/gin"
"github.com/tidwall/gjson"
"go.uber.org/zap"
)
@@ -111,7 +112,7 @@ func (s *OpenAIGatewayService) forwardChatCompletionsViaNativeAnthropic(
}
upstreamCtx, releaseUpstreamCtx := detachStreamUpstreamContext(ctx, reqStream)
upstreamReq, _, err := s.buildNativeAnthropicUpstreamRequest(upstreamCtx, c, account, anthropicBody, apiKey, targetURL, body)
upstreamReq, forwardedBody, err := s.buildNativeAnthropicUpstreamRequest(upstreamCtx, c, account, anthropicBody, apiKey, targetURL, body)
releaseUpstreamCtx()
if err != nil {
return nil, fmt.Errorf("build upstream request: %w", err)
@@ -132,8 +133,9 @@ func (s *OpenAIGatewayService) forwardChatCompletionsViaNativeAnthropic(
return nil, fmt.Errorf("upstream error: %d %s", resp.StatusCode, upstreamMsg)
}
reasoningEffort := extractCCReasoningEffortFromBody(body, upstreamModel, billingModel, originalModel)
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, billingModel)
// Bill the Anthropic effort actually sent upstream (for example xhigh → max).
reasoningEffort := NormalizeClaudeOutputEffort(gjson.GetBytes(forwardedBody, "output_config.effort").String())
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, forwardedBody, upstreamModel)
if clientStream {
return s.handleCCStreamingFromNativeAnthropic(resp, c, originalModel, billingModel, upstreamModel, reasoningEffort, startTime, includeUsage)
backend/internal/service/openai_gateway_converted_reasoning_pricing_test.go
+133
View File
@@ -0,0 +1,133 @@
//go:build unit
package service
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
func TestOpenAINativeAnthropicReasoningPricingUsesForwardedEffort(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, endpoint := range []string{"responses", "chat/completions"} {
for _, stream := range []bool{false, true} {
for _, tc := range []struct {
effort string
wantEffort string
wantCost float64
}{
{"xhigh", "max", 3},
{"high", "high", 2},
{"", "", 1},
} {
t.Run(fmt.Sprintf("%s/stream=%t/effort=%s", endpoint, stream, tc.effort), func(t *testing.T) {
request := map[string]any{"model": "k3", "stream": stream}
if endpoint == "responses" {
request["input"] = "hello"
if tc.effort != "" {
request["reasoning"] = map[string]string{"effort": tc.effort}
}
} else {
request["messages"] = []map[string]string{{"role": "user", "content": "hello"}}
if tc.effort != "" {
request["reasoning_effort"] = tc.effort
}
}
body, err := json.Marshal(request)
require.NoError(t, err)
resp := nativeAnthropicStreamResponse()
responseBody, err := io.ReadAll(resp.Body)
require.NoError(t, err)
resp.Body = io.NopCloser(strings.NewReader(strings.ReplaceAll(string(responseBody), `"input_tokens":93`, `"input_tokens":1000000`)))
upstream := &httpUpstreamRecorder{resp: resp}
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
c := adaptiveProtocolTestContext("/v1/"+endpoint, body)
account := nativeAnthropicTestAccount()
account.Credentials["base_url"] = "http://anthropic.example"
var result *OpenAIForwardResult
if endpoint == "responses" {
result, err = svc.Forward(context.Background(), c, account, body)
} else {
result, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
}
require.NoError(t, err)
require.Equal(t, "/v1/messages", upstream.lastReq.URL.Path)
require.Equal(t, tc.wantEffort, gjson.GetBytes(upstream.lastBody, "output_config.effort").String())
require.Equal(t, tc.wantEffort, optionalStringValue(result.ReasoningEffort))
assertConvertedReasoningPrice(t, result, tc.wantCost)
})
}
}
}
}
func TestOpenAIMessagesGLMReasoningPricingUsesForwardedEffort(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, stream := range []bool{false, true} {
for _, tc := range []struct {
effort string
maxPolicy string
wantEffort string
wantCost float64
}{
{"medium", "", "high", 2},
{"xhigh", "", "max", 3},
{"xhigh", "medium", "medium", 1.5},
} {
t.Run(fmt.Sprintf("stream=%t/effort=%s/policy=%s", stream, tc.effort, tc.maxPolicy), func(t *testing.T) {
body := []byte(fmt.Sprintf(`{"model":"glm-5.2","stream":%t,"max_tokens":32,"output_config":{"effort":%q},"messages":[{"role":"user","content":"hello"}]}`, stream, tc.effort))
responseBody := `{"id":"chatcmpl_glm","object":"chat.completion","model":"glm-5.2","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1000000,"completion_tokens":2,"total_tokens":1000002}}`
contentType := "application/json"
if stream {
contentType = "text/event-stream"
responseBody = "data: " + `{"id":"chatcmpl_glm","object":"chat.completion.chunk","model":"glm-5.2","choices":[{"index":0,"delta":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1000000,"completion_tokens":2,"total_tokens":1000002}}` + "\n\ndata: [DONE]\n\n"
}
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{contentType}},
Body: io.NopCloser(strings.NewReader(responseBody)),
}}
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
ctx := context.Background()
if tc.maxPolicy != "" {
ctx = WithOpenAIReasoningEffortPolicy(ctx, tc.maxPolicy, nil, "")
}
result, err := svc.ForwardAsAnthropic(ctx, adaptiveProtocolTestContext("/v1/messages", body), forceChatMessagesFallbackAccount(), body, "", "")
require.NoError(t, err)
require.Equal(t, "/v1/chat/completions", upstream.lastReq.URL.Path)
require.Equal(t, tc.wantEffort, gjson.GetBytes(upstream.lastBody, "reasoning_effort").String())
require.Equal(t, tc.wantEffort, optionalStringValue(result.ReasoningEffort))
assertConvertedReasoningPrice(t, result, tc.wantCost)
})
}
}
}
func assertConvertedReasoningPrice(t *testing.T, result *OpenAIForwardResult, wantCost float64) {
t.Helper()
require.Equal(t, 1_000_000, result.Usage.InputTokens)
billing := NewBillingService(rawChatCompletionsTestConfig(), nil)
group := &Group{ID: 1, Platform: PlatformOpenAI, ModelPricing: []ChannelModelPricing{{
Models: []string{result.BillingModel}, BillingMode: BillingModeToken,
InputPrice: testPtrFloat64(1e-6), OutputPrice: testPtrFloat64(0),
ReasoningEffortMultipliers: map[string]float64{"medium": 1.5, "high": 2, "xhigh": 2, "max": 3},
}}}
cost, err := billing.CalculateTokenCostForRequest(TokenCostRequest{
Ctx: context.Background(), Model: result.BillingModel, Group: group,
Tokens: UsageTokens{InputTokens: result.Usage.InputTokens, OutputTokens: result.Usage.OutputTokens},
ReasoningEffort: optionalStringValue(result.ReasoningEffort), RateMultiplier: 1,
Resolver: NewModelPricingResolver(nil, billing),
})
require.NoError(t, err)
require.InDelta(t, wantCost, cost.TotalCost, 1e-12)
require.InDelta(t, wantCost, cost.ActualCost, 1e-12)
}
backend/internal/service/openai_gateway_messages_chat_fallback.go
+4 -7
View File
@@ -12,7 +12,6 @@ import (
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/Wei-Shaw/sub2api/internal/util/responseheaders"
"github.com/gin-gonic/gin"
"github.com/tidwall/gjson"
"go.uber.org/zap"
)
@@ -67,9 +66,6 @@ func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions(
chatReq.StreamOptions = &apicompat.ChatStreamOptions{IncludeUsage: true}
}
convertedEffort := chatReq.ReasoningEffort
reasoningEffort := &convertedEffort
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, billingModel)
serviceTier := extractOpenAIServiceTierFromBody(body)
chatBody, err := json.Marshal(chatReq)
@@ -90,11 +86,12 @@ func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions(
}
if changed {
chatBody = policyBody
if effectiveEffort := strings.TrimSpace(gjson.GetBytes(chatBody, "reasoning_effort").String()); effectiveEffort != "" {
reasoningEffort = &effectiveEffort
}
}
}
// Provider normalization and policy caps can both change the converted effort.
// Use the final outbound value for usage logs and billing.
reasoningEffort := extractOpenAIReasoningEffortFromBody(chatBody, upstreamModel, billingModel, originalModel)
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, chatBody, upstreamModel)
// Unlike forwardResponsesViaRawChatCompletions, applyOpenAIFastPolicyToBody
// is intentionally skipped: Anthropic Messages bodies carry no service_tier,
// so the converted Chat Completions body never contains one and the policy
backend/internal/service/openai_gateway_responses_anthropic_native.go
+4 -4
View File
@@ -75,9 +75,6 @@ func (s *OpenAIGatewayService) forwardResponsesViaNativeAnthropic(
upstreamModel := normalizeOpenAIModelForUpstream(account, billingModel)
anthropicReq.Model = upstreamModel
reasoningEffort := ExtractResponsesReasoningEffortFromBody(body, upstreamModel, billingModel, originalModel)
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, billingModel)
// 5. Force upstream streaming（客户端原始终决定响应格式；
// 上游恒为流式，非流式由缓冲路径组装）。
anthropicReq.Stream = true
@@ -116,11 +113,14 @@ func (s *OpenAIGatewayService) forwardResponsesViaNativeAnthropic(
}
upstreamCtx, releaseUpstreamCtx := detachStreamUpstreamContext(ctx, reqStream)
upstreamReq, _, err := s.buildNativeAnthropicUpstreamRequest(upstreamCtx, c, account, anthropicBody, apiKey, targetURL, body)
upstreamReq, forwardedBody, err := s.buildNativeAnthropicUpstreamRequest(upstreamCtx, c, account, anthropicBody, apiKey, targetURL, body)
releaseUpstreamCtx()
if err != nil {
return nil, fmt.Errorf("build upstream request: %w", err)
}
// Bill the Anthropic effort actually sent upstream (for example xhigh → max).
reasoningEffort := NormalizeClaudeOutputEffort(gjson.GetBytes(forwardedBody, "output_config.effort").String())
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, forwardedBody, upstreamModel)
resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account)
if err != nil {
backend/internal/service/openai_gateway_usage.go
+24 -19
View File
@@ -590,6 +590,7 @@ func (s *OpenAIGatewayService) calculateOpenAIRecordUsageCost(
Ctx: ctx, Model: billingModel, GroupID: &gid, Group: apiKey.Group,
UsageUnits: result.AudioUsage.DurationOrUnits, SizeTier: result.AudioUsage.Mode,
RateMultiplier: webSearchMultiplier, Resolver: s.resolver, Resolved: resolved,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
})
}
cfg := groupAudioPriceConfigFromAPIKey(apiKey)
@@ -751,6 +752,7 @@ func (s *OpenAIGatewayService) calculateOpenAIImageCost(
Ctx: ctx, Model: billingModel, GroupID: &gid, Group: apiKey.Group,
RequestCount: result.ImageCount, SizeTier: sizeTier,
RateMultiplier: multiplier, Resolver: s.resolver, Resolved: resolved,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
})
if err == nil {
return cost
@@ -771,15 +773,16 @@ func (s *OpenAIGatewayService) calculateOpenAIImageCost(
(resolved.Mode == BillingModePerRequest || resolved.Mode == BillingModeImage) {
gid := apiKey.Group.ID
cost, err := s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
RequestCount: result.ImageCount,
SizeTier: sizeTier,
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
RequestCount: result.ImageCount,
SizeTier: sizeTier,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
})
if err == nil {
return cost
@@ -810,6 +813,7 @@ func (s *OpenAIGatewayService) calculateOpenAIVideoCost(
Ctx: ctx, Model: billingModel, GroupID: &gid, Group: apiKey.Group,
UsageUnits: float64(videoCount * durationSeconds), SizeTier: resolution,
RateMultiplier: multiplier, Resolver: s.resolver, Resolved: resolved,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
})
if err == nil {
return cost
@@ -835,16 +839,17 @@ func (s *OpenAIGatewayService) calculateOpenAIVideoCost(
units = float64(videoCount * durationSeconds)
}
cost, err := s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
RequestCount: videoCount,
UsageUnits: units,
SizeTier: resolution,
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Group: apiKey.Group,
RequestCount: videoCount,
UsageUnits: units,
SizeTier: resolution,
ReasoningEffort: optionalStringValue(result.ReasoningEffort),
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
})
if err == nil {
cost.BillingMode = string(BillingModeVideo)