merge main 并修复与 #5730 的语义冲突

main 侧 #5730 新增的 openai_gateway_cn_fixes_test.go 按旧 11 参签名调用
calculateOpenAIRecordUsageCost;本分支为该函数新增了第 12 个参数
pricingAt。文本无冲突但 test build 会失败,此处按本分支对同类测试
调用点的既有处理方式补传 time.Time{}。
This commit is contained in:
shaw
2026-08-17 22:27:22 +08:00
20 changed files with 551 additions and 232 deletions
@@ -98,7 +98,7 @@ func NewGroupHandler(adminService service.AdminService, dashboardService *servic
type CreateGroupRequest struct {
Name string `json:"name" binding:"required"`
Description string `json:"description"`
Platform string `json:"platform" binding:"omitempty,oneof=anthropic openai gemini antigravity grok composite"`
Platform string `json:"platform" binding:"omitempty,oneof=anthropic openai gemini antigravity grok kimi zhipu deepseek composite"`
RateMultiplier float64 `json:"rate_multiplier"`
IsExclusive bool `json:"is_exclusive"`
SubscriptionType string `json:"subscription_type" binding:"omitempty,oneof=standard subscription"`
@@ -166,7 +166,7 @@ type CreateGroupRequest struct {
type UpdateGroupRequest struct {
Name string `json:"name"`
Description *string `json:"description"`
Platform string `json:"platform" binding:"omitempty,oneof=anthropic openai gemini antigravity grok composite"`
Platform string `json:"platform" binding:"omitempty,oneof=anthropic openai gemini antigravity grok kimi zhipu deepseek composite"`
RateMultiplier *float64 `json:"rate_multiplier"`
IsExclusive *bool `json:"is_exclusive"`
Status string `json:"status" binding:"omitempty,oneof=active inactive"`
@@ -0,0 +1,83 @@
//go:build unit
package admin
import (
"bytes"
"fmt"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
// 回归分组平台枚举:kimi/zhipu/deepseek 必须能通过 Create/Update 的 binding 校验
// (历史 bug:调度/路由链路已支持 CN 平台分组,但 oneof 白名单漏加三平台,导致
// 平台分组无法创建、CN 账号"无可用分组");非法值仍须被拒。
func bindGroupPlatformJSON(t *testing.T, target any, body string) error {
t.Helper()
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
c.Request = httptest.NewRequest("POST", "/", bytes.NewBufferString(body))
c.Request.Header.Set("Content-Type", "application/json")
return c.ShouldBindJSON(target)
}
func TestGroupPlatformBinding_AllowedPlatforms(t *testing.T) {
allowed := []string{
"anthropic", "openai", "gemini", "antigravity", "grok",
"kimi", "zhipu", "deepseek", "composite",
}
for _, platform := range allowed {
t.Run("create_"+platform, func(t *testing.T) {
var req CreateGroupRequest
body := fmt.Sprintf(`{"name":"g","platform":%q}`, platform)
require.NoError(t, bindGroupPlatformJSON(t, &req, body),
"platform %q 应通过 CreateGroupRequest 校验", platform)
require.Equal(t, platform, req.Platform)
})
t.Run("update_"+platform, func(t *testing.T) {
var req UpdateGroupRequest
body := fmt.Sprintf(`{"platform":%q}`, platform)
require.NoError(t, bindGroupPlatformJSON(t, &req, body),
"platform %q 应通过 UpdateGroupRequest 校验", platform)
require.Equal(t, platform, req.Platform)
})
}
}
func TestGroupPlatformBinding_RejectsInvalidPlatforms(t *testing.T) {
invalid := []string{
"moonshot", // 厂商别名,不是平台标识
"Kimi", // 大小写敏感
"openai ", // 尾随空格
"glm",
"bogus",
}
for _, platform := range invalid {
t.Run("create_"+platform, func(t *testing.T) {
var req CreateGroupRequest
body := fmt.Sprintf(`{"name":"g","platform":%q}`, platform)
require.Error(t, bindGroupPlatformJSON(t, &req, body),
"platform %q 应被 CreateGroupRequest 拒绝", platform)
})
t.Run("update_"+platform, func(t *testing.T) {
var req UpdateGroupRequest
body := fmt.Sprintf(`{"platform":%q}`, platform)
require.Error(t, bindGroupPlatformJSON(t, &req, body),
"platform %q 应被 UpdateGroupRequest 拒绝", platform)
})
}
}
// 守住 composite 路由目标不放行 CN:CN 平台不可作为 composite 路由目标
// (DetectModelPlatform/isConcreteRequestPlatform 均无 CN 分支,放行即打开半实现路径)。
func TestCompositeRouteTargetPlatform_StillExcludesCNProviders(t *testing.T) {
for _, platform := range []string{"kimi", "zhipu", "deepseek"} {
var req CompositeRouteRequest
body := fmt.Sprintf(`{"public_model":"m","target_platform":%q}`, platform)
require.Error(t, bindGroupPlatformJSON(t, &req, body),
"composite target_platform %q 应保持被拒", platform)
}
}
@@ -255,6 +255,48 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
if err == nil && result != nil && result.FirstTokenMs != nil {
service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
}
// #5148 对齐:错误返回携带的部分 result(流中断前上游已计量的 usage)照常
// 入账;failover 错误恒定 result=nil,不会重复计费。
submitChatUsage := func(res *service.OpenAIForwardResult) {
if res == nil {
return
}
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: res,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.chat_completions"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai_chat_completions.record_usage_failed", zap.Error(err))
}
})
}
if err != nil {
if result != nil && result.ImageCount > 0 {
reqLog.Warn("openai_chat_completions.forward_partial_error_with_image_result",
@@ -339,6 +381,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated),
zap.Error(err),
)
submitChatUsage(result)
return
}
}
@@ -348,42 +391,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(reqModel), true, nil)
}
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: result,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, result.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.chat_completions"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai_chat_completions.record_usage_failed", zap.Error(err))
}
})
submitChatUsage(result)
reqLog.Debug("openai_chat_completions.request_completed",
zap.Int64("account_id", account.ID),
zap.Int("switch_count", switchCount),
@@ -0,0 +1,29 @@
package handler
// CN 分组 /v1/messages 调度闸门回归(修复:正常途径创建的 CN 分组曾恒 403):
// sanitizeGroupMessagesDispatchFields 对非 openai 平台强制 AllowMessagesDispatch
// =false,故 CN 分组必须与 grok 一样在闸门处豁免,否则原生 Anthropic 直通
//(Claude Code 主用例)永远不可达。
import (
"testing"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
func TestAllowOpenAICompatibleMessagesDispatch_CNProvidersExempt(t *testing.T) {
require.True(t, allowOpenAICompatibleMessagesDispatch(nil), "无 key 保持放行")
for _, platform := range []string{service.PlatformKimi, service.PlatformZhipu, service.PlatformDeepseek, service.PlatformGrok} {
apiKey := &service.APIKey{Group: &service.Group{Platform: platform, AllowMessagesDispatch: false}}
require.True(t, allowOpenAICompatibleMessagesDispatch(apiKey),
"%s 分组必须豁免 allow_messages_dispatch 闸门", platform)
}
// 非回归:openai 分组仍受开关控制。
openaiOff := &service.APIKey{Group: &service.Group{Platform: service.PlatformOpenAI, AllowMessagesDispatch: false}}
require.False(t, allowOpenAICompatibleMessagesDispatch(openaiOff))
openaiOn := &service.APIKey{Group: &service.Group{Platform: service.PlatformOpenAI, AllowMessagesDispatch: true}}
require.True(t, allowOpenAICompatibleMessagesDispatch(openaiOn))
}
@@ -78,7 +78,7 @@ func (h *OpenAIGatewayHandler) CountTokens(c *gin.Context) {
zap.Any("group_id", apiKey.GroupID),
)
if apiKey.Group != nil && !apiKey.Group.AllowMessagesDispatch {
if !allowOpenAICompatibleMessagesDispatch(apiKey) {
h.anthropicErrorResponse(c, http.StatusForbidden, "permission_error",
"This group does not allow /v1/messages dispatch")
return
@@ -197,6 +197,13 @@ func allowOpenAICompatibleMessagesDispatch(apiKey *service.APIKey) bool {
if apiKey.Group.Platform == service.PlatformGrok {
return true
}
// 国产供应商分组与 grok 同语义:/v1/messages 就是其主要服务形态(anthropic
// 协议账号原生直通 Claude Code),无需 allow_messages_dispatch 开关授权——
// 该开关对非 openai 平台恒被 sanitizeGroupMessagesDispatchFields 置 false,
// 若不豁免,CN 分组将永远 403。
if service.IsCNProvider(apiKey.Group.Platform) {
return true
}
return apiKey.Group.AllowMessagesDispatch
}
@@ -597,6 +604,50 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
if err == nil && result != nil && result.FirstTokenMs != nil {
service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
}
// #5148 对齐:错误返回携带的部分 result(流中断前上游已计量的 usage)照常
// 入账;failover 错误恒定 result=nil,不会重复计费。
submitResponsesUsage := func(res *service.OpenAIForwardResult) {
if res == nil {
return
}
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
requestPayloadHash := service.HashUsageRequestPayload(body)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: res,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
RequestPayloadHash: requestPayloadHash,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.responses"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai.record_usage_failed", zap.Error(err))
}
})
}
if err != nil {
if result != nil && result.ImageCount > 0 {
reqLog.Warn("openai.forward_partial_error_with_image_result",
@@ -696,6 +747,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated),
zap.Error(err),
}
submitResponsesUsage(result)
if shouldLogOpenAIForwardFailureAsWarn(c, wroteFallback) {
reqLog.Warn("openai.forward_failed", fields...)
return
@@ -714,46 +766,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(reqModel), openAIForwardSucceededForScheduling(result), nil)
}
// 捕获请求信息(用于异步记录,避免在 goroutine 中访问 gin.Context)
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
requestPayloadHash := service.HashUsageRequestPayload(body)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
// 使用量记录通过有界 worker 池提交,避免请求热路径创建无界 goroutine。
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: result,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
RequestPayloadHash: requestPayloadHash,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, result.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.responses"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai.record_usage_failed", zap.Error(err))
}
})
submitResponsesUsage(result)
reqLog.Debug("openai.request_completed",
zap.Int64("account_id", account.ID),
zap.Int("switch_count", switchCount),
@@ -1145,6 +1159,51 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
if err == nil && result != nil && result.FirstTokenMs != nil {
service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
}
// Forward 与错误一起返回的部分结果:流中断/客户端断开排水前上游已计量的
// usage 照常入账,避免上游已产生消耗的请求完全漏记(#5148,对齐 anthropic
// 网关同名修复)。failover 错误恒定 result=nil,不会重复计费。
submitMessagesUsage := func(res *service.OpenAIForwardResult) {
if res == nil {
return
}
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
requestPayloadHash := service.HashUsageRequestPayload(body)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: res,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
RequestPayloadHash: requestPayloadHash,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMappingMsg, reqModel, res.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.messages"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai_messages.record_usage_failed", zap.Error(err))
}
})
}
if err != nil {
if result != nil && result.ImageCount > 0 {
reqLog.Warn("openai_messages.forward_partial_error_with_image_result",
@@ -1219,6 +1278,9 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
zap.Int64("account_id", account.ID),
zap.Error(err),
)
// 断开排水期间上游已计量的 usage 必须入账(此前直接 return 丢弃,
// payg 上游照常计费而平台漏记)。
submitMessagesUsage(result)
return
}
h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(currentRoutingModel), false, nil)
@@ -1228,6 +1290,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
zap.Bool("fallback_error_response_written", wroteFallback),
zap.Error(err),
)
submitMessagesUsage(result)
return
}
}
@@ -1237,44 +1300,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(currentRoutingModel), true, nil)
}
userAgent := c.GetHeader("User-Agent")
clientIP := ip.GetClientIP(c)
requestPayloadHash := service.HashUsageRequestPayload(body)
inboundEndpoint := GetInboundEndpoint(c)
upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
sessionID := service.ExtractClientSessionID(c)
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
Result: result,
APIKey: apiKey,
User: apiKey.User,
Account: account,
Subscription: subscription,
InboundEndpoint: inboundEndpoint,
UpstreamEndpoint: upstreamEndpoint,
UserAgent: userAgent,
IPAddress: clientIP,
RequestPayloadHash: requestPayloadHash,
APIKeyService: h.apiKeyService,
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: clientRequestedUsageFields(c, channelMappingMsg, reqModel, result.UpstreamModel),
PricingAt: pricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
logger.L().With(
zap.String("component", "handler.openai_gateway.messages"),
zap.Int64("user_id", subject.UserID),
zap.Int64("api_key_id", apiKey.ID),
zap.Any("group_id", apiKey.GroupID),
zap.String("model", reqModel),
zap.Int64("account_id", account.ID),
).Error("openai_messages.record_usage_failed", zap.Error(err))
}
})
submitMessagesUsage(result)
reqLog.Debug("openai_messages.request_completed",
zap.Int64("account_id", account.ID),
zap.Int("switch_count", switchCount),