mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
refactor(service): 纯移动拆分 openai_gateway_service.go（4872→1095行）
This commit is contained in:
shaw
@@ -0,0 +1,956 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/openai"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
// Forward forwards request to OpenAI API
|
||||
func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, account *Account, body []byte) (*OpenAIForwardResult, error) {
|
||||
startTime := time.Now()
|
||||
|
||||
restrictionResult := s.detectCodexClientRestriction(c, account, body)
|
||||
apiKeyID := getAPIKeyIDFromContext(c)
|
||||
logCodexCLIOnlyDetection(ctx, c, account, apiKeyID, restrictionResult, body)
|
||||
if restrictionResult.Enabled && !restrictionResult.Matched {
|
||||
MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalPolicyDenied)
|
||||
c.JSON(http.StatusForbidden, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "forbidden_error",
|
||||
"message": CodexClientRestrictionMessage(restrictionResult),
|
||||
},
|
||||
})
|
||||
return nil, errors.New("codex_cli_only restriction: only codex official clients are allowed")
|
||||
}
|
||||
|
||||
originalBody := body
|
||||
requestView := newOpenAIRequestView(body)
|
||||
reqModel, reqStream, promptCacheKey := requestView.Model, requestView.Stream, requestView.PromptCacheKey
|
||||
originalModel := reqModel
|
||||
|
||||
if account.Platform == PlatformGrok {
|
||||
_ = promptCacheKey
|
||||
return s.forwardGrokResponses(ctx, c, account, body, originalModel, reqStream, startTime)
|
||||
}
|
||||
|
||||
if account.Type == AccountTypeAPIKey && !openai_compat.ShouldUseResponsesAPI(account.Extra) {
|
||||
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body)
|
||||
}
|
||||
|
||||
compatMessagesBridge := isOpenAICompatMessagesBridgeBody(body)
|
||||
setOpenAICompatMessagesBridgeContext(c, compatMessagesBridge)
|
||||
|
||||
isCodexCLI := openai.IsCodexOfficialClientByHeaders(c.GetHeader("User-Agent"), c.GetHeader("originator")) || (s.cfg != nil && s.cfg.Gateway.ForceCodexCLI)
|
||||
wsDecision := s.getOpenAIWSProtocolResolver().Resolve(account)
|
||||
clientTransport := GetOpenAIClientTransport(c)
|
||||
// 仅允许 WS 入站请求走 WS 上游，避免出现 HTTP -> WS 协议混用。
|
||||
wsDecision = resolveOpenAIWSDecisionByClientTransport(wsDecision, clientTransport)
|
||||
if c != nil {
|
||||
c.Set("openai_ws_transport_decision", string(wsDecision.Transport))
|
||||
c.Set("openai_ws_transport_reason", wsDecision.Reason)
|
||||
}
|
||||
if wsDecision.Transport == OpenAIUpstreamTransportResponsesWebsocketV2 {
|
||||
logOpenAIWSModeDebug(
|
||||
"selected account_id=%d account_type=%s transport=%s reason=%s model=%s stream=%v",
|
||||
account.ID,
|
||||
account.Type,
|
||||
normalizeOpenAIWSLogValue(string(wsDecision.Transport)),
|
||||
normalizeOpenAIWSLogValue(wsDecision.Reason),
|
||||
reqModel,
|
||||
reqStream,
|
||||
)
|
||||
}
|
||||
// 当前仅支持 WSv2；WSv1 命中时直接返回错误，避免出现“配置可开但行为不确定”。
|
||||
if wsDecision.Transport == OpenAIUpstreamTransportResponsesWebsocket {
|
||||
if c != nil {
|
||||
MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalFeatureGate)
|
||||
c.JSON(http.StatusBadRequest, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "invalid_request_error",
|
||||
"message": "OpenAI WSv1 is temporarily unsupported. Please enable responses_websockets_v2.",
|
||||
},
|
||||
})
|
||||
}
|
||||
return nil, errors.New("openai ws v1 is temporarily unsupported; use ws v2")
|
||||
}
|
||||
passthroughEnabled := account.IsOpenAIPassthroughEnabled()
|
||||
if passthroughEnabled {
|
||||
// 透传分支只需要轻量提取字段，避免热路径全量 Unmarshal。
|
||||
reasoningEffort := extractOpenAIReasoningEffortFromBody(body, reqModel)
|
||||
// 国产模型默认 effort 补充：也要用 mappedModel 判定是否是 passback-required 上游。
|
||||
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, account.GetMappedModel(reqModel))
|
||||
return s.forwardOpenAIPassthrough(ctx, c, account, originalBody, reqModel, reasoningEffort, reqStream, startTime)
|
||||
}
|
||||
|
||||
bodyModified := false
|
||||
var reqBody map[string]any
|
||||
ensureReqBody := func() (map[string]any, error) {
|
||||
if requestView.HasPatches() {
|
||||
patchedBody, patchErr := requestView.ApplyPatches()
|
||||
if patchErr != nil {
|
||||
return nil, patchErr
|
||||
}
|
||||
body = patchedBody
|
||||
requestView = newOpenAIRequestView(body)
|
||||
reqBody = nil
|
||||
bodyModified = false
|
||||
}
|
||||
if reqBody != nil {
|
||||
return reqBody, nil
|
||||
}
|
||||
decoded, decodeErr := requestView.Decode(c)
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
reqBody = decoded
|
||||
return reqBody, nil
|
||||
}
|
||||
markPatchSet := func(path string, value any) {
|
||||
bodyModified = true
|
||||
if requestView.patchesDisabled {
|
||||
if reqBody != nil {
|
||||
setOpenAIRequestMapPath(reqBody, path, value)
|
||||
}
|
||||
return
|
||||
}
|
||||
requestView.MarkPatchSet(path, value)
|
||||
}
|
||||
markPatchDelete := func(path string) {
|
||||
bodyModified = true
|
||||
if requestView.patchesDisabled {
|
||||
if reqBody != nil {
|
||||
deleteOpenAIRequestMapPath(reqBody, path)
|
||||
}
|
||||
return
|
||||
}
|
||||
requestView.MarkPatchDelete(path)
|
||||
}
|
||||
disablePatch := func() {
|
||||
requestView.DisablePatches()
|
||||
}
|
||||
markDecodedModified := func() {
|
||||
bodyModified = true
|
||||
disablePatch()
|
||||
}
|
||||
|
||||
apiKey := getAPIKeyFromContext(c)
|
||||
imageGenerationAllowed := GroupAllowsImageGeneration(nil)
|
||||
if apiKey != nil {
|
||||
imageGenerationAllowed = GroupAllowsImageGeneration(apiKey.Group)
|
||||
}
|
||||
codexImageGenerationExplicitToolPolicy := codexImageGenerationExplicitToolPolicyAllow
|
||||
if isCodexCLI {
|
||||
codexImageGenerationExplicitToolPolicy = account.CodexImageGenerationExplicitToolPolicy()
|
||||
}
|
||||
codexImageGenerationBridgeEnabled := isCodexCLI && imageGenerationAllowed && codexImageGenerationExplicitToolPolicy != codexImageGenerationExplicitToolPolicyStrip && s.isCodexImageGenerationBridgeEnabled(ctx, account, apiKey)
|
||||
var imageIntent bool
|
||||
if isCodexCLI && codexImageGenerationExplicitToolPolicy == codexImageGenerationExplicitToolPolicyStrip {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if stripOpenAIImageGenerationTools(decoded) {
|
||||
markDecodedModified()
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Stripped /responses image_generation tool for Codex client by account policy")
|
||||
}
|
||||
imageIntent = IsImageGenerationIntentMap(openAIResponsesEndpoint, reqModel, decoded)
|
||||
} else {
|
||||
imageIntent = IsImageGenerationIntent(openAIResponsesEndpoint, reqModel, body)
|
||||
}
|
||||
if imageIntent && !imageGenerationAllowed {
|
||||
MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalFeatureGate)
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": gin.H{"type": "permission_error", "message": ImageGenerationPermissionMessage()}})
|
||||
return nil, errors.New("image generation disabled for group")
|
||||
}
|
||||
|
||||
instructions := gjson.GetBytes(body, "instructions")
|
||||
instructionsEmpty := !instructions.Exists() || instructions.Type != gjson.String || strings.TrimSpace(instructions.String()) == ""
|
||||
if instructionsEmpty && !compatMessagesBridge {
|
||||
markPatchSet("instructions", defaultCodexSynthInstructions(reqModel))
|
||||
}
|
||||
|
||||
billingModel := account.GetMappedModel(reqModel)
|
||||
if billingModel != reqModel {
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Model mapping applied: %s -> %s (account: %s, isCodexCLI: %v)", reqModel, billingModel, account.Name, isCodexCLI)
|
||||
reqModel = billingModel
|
||||
markPatchSet("model", billingModel)
|
||||
}
|
||||
upstreamModel := billingModel
|
||||
isCompactRequest := isOpenAIResponsesCompactPath(c)
|
||||
compactMapped := false
|
||||
if isCompactRequest {
|
||||
compactMappedModel := resolveOpenAICompactForwardModel(account, billingModel)
|
||||
if compactMappedModel != "" && compactMappedModel != billingModel {
|
||||
compactMapped = true
|
||||
upstreamModel = compactMappedModel
|
||||
reqModel = compactMappedModel
|
||||
markPatchSet("model", compactMappedModel)
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Compact model mapping applied: %s -> %s (account: %s, isCodexCLI: %v)", billingModel, compactMappedModel, account.Name, isCodexCLI)
|
||||
}
|
||||
}
|
||||
if !compactMapped {
|
||||
modelForNormalize := reqModel
|
||||
if modelForNormalize == "" {
|
||||
modelForNormalize = requestView.Model
|
||||
}
|
||||
upstreamModel = normalizeOpenAIModelForUpstream(account, modelForNormalize)
|
||||
if upstreamModel != "" && upstreamModel != modelForNormalize {
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Upstream model resolved: %s -> %s (account: %s, type: %s, isCodexCLI: %v)", modelForNormalize, upstreamModel, account.Name, account.Type, isCodexCLI)
|
||||
reqModel = upstreamModel
|
||||
markPatchSet("model", upstreamModel)
|
||||
}
|
||||
}
|
||||
if strings.TrimSpace(gjson.GetBytes(body, "reasoning.effort").String()) == "minimal" {
|
||||
markPatchSet("reasoning.effort", "none")
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Normalized reasoning.effort: minimal -> none (account: %s)", account.Name)
|
||||
}
|
||||
|
||||
imageIntent = imageIntent || IsImageGenerationIntent(openAIResponsesEndpoint, reqModel, nil) || isOpenAIImageGenerationModel(upstreamModel)
|
||||
if imageIntent && !imageGenerationAllowed {
|
||||
MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalFeatureGate)
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": gin.H{"type": "permission_error", "message": ImageGenerationPermissionMessage()}})
|
||||
return nil, errors.New("image generation disabled for group")
|
||||
}
|
||||
|
||||
// /responses/compact 是会话压缩请求：上游不接受 tool_choice（400 unknown_parameter），
|
||||
// 注入 image_generation 工具也没有意义，整块豁免。
|
||||
if imageGenerationAllowed && !isCompactRequest && (codexImageGenerationBridgeEnabled || isOpenAIImageGenerationModel(requestView.Model) || openAIRequestBodyImageGenerationToolNeedsNormalization(body) || isOpenAIImageGenerationModel(upstreamModel)) {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if codexImageGenerationBridgeEnabled && ensureOpenAIResponsesImageGenerationTool(decoded) {
|
||||
markDecodedModified()
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Injected /responses image_generation tool for Codex client")
|
||||
}
|
||||
if codexImageGenerationBridgeEnabled && ensureOpenAIResponsesImageGenerationToolChoiceAuto(decoded) {
|
||||
markDecodedModified()
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Set /responses image_generation tool_choice=auto for Codex client")
|
||||
}
|
||||
if normalizeOpenAIResponsesImageGenerationTools(decoded) {
|
||||
markDecodedModified()
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Normalized /responses image_generation tool payload")
|
||||
}
|
||||
if normalizeOpenAIResponsesImageOnlyModel(decoded) {
|
||||
markDecodedModified()
|
||||
if model, ok := decoded["model"].(string); ok {
|
||||
upstreamModel = strings.TrimSpace(model)
|
||||
}
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Normalized /responses image-only model request inbound_model=%s image_model=%s upstream_model=%s", requestView.Model, billingModel, upstreamModel)
|
||||
}
|
||||
if err := validateOpenAIResponsesImageModel(decoded, upstreamModel); err != nil {
|
||||
setOpsUpstreamError(c, http.StatusBadRequest, err.Error(), "")
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": gin.H{"type": "invalid_request_error", "message": err.Error(), "param": "model"}})
|
||||
return nil, err
|
||||
}
|
||||
if hasOpenAIImageGenerationTool(decoded) {
|
||||
imageIntent = true
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] /responses image_generation request inbound_model=%s mapped_model=%s account_type=%s", requestView.Model, upstreamModel, account.Type)
|
||||
}
|
||||
if codexImageGenerationBridgeEnabled && applyCodexImageGenerationBridgeInstructions(decoded) {
|
||||
markDecodedModified()
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Added Codex image_generation bridge instructions")
|
||||
}
|
||||
} else if imageGenerationAllowed && imageIntent && openAIRequestBodyHasImageGenerationTool(body) {
|
||||
// 完整 image_generation tool 只做 raw 计费读取，校验/桥接/旧字段迁移命中时才展开大 input map。
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] /responses image_generation request inbound_model=%s mapped_model=%s account_type=%s", requestView.Model, upstreamModel, account.Type)
|
||||
}
|
||||
|
||||
if isCodexSparkModel(upstreamModel) && openAIRequestBodyMayContainImageInput(body) {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if err := validateCodexSparkInput(decoded, upstreamModel); err != nil {
|
||||
setOpsUpstreamError(c, http.StatusBadRequest, err.Error(), "")
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": gin.H{"type": "invalid_request_error", "message": err.Error(), "param": "input"}})
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// gpt-5.3-codex-spark also rejects the image_generation tool (HTTP 400,
|
||||
// param=tools). Strip it here so both APIKey and OAuth /responses paths are
|
||||
// covered regardless of the image-generation feature gate.
|
||||
if isCodexSparkModel(upstreamModel) && openAIRequestBodyHasImageGenerationTool(body) {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if stripCodexSparkImageGenerationTools(decoded) {
|
||||
markDecodedModified()
|
||||
}
|
||||
}
|
||||
|
||||
if account.Type == AccountTypeOAuth {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
codexResult := codexTransformResult{}
|
||||
if compatMessagesBridge {
|
||||
codexResult = applyCodexOAuthTransformWithOptions(decoded, codexOAuthTransformOptions{IsCodexCLI: isCodexCLI, IsCompact: isCompactRequest, SkipDefaultInstructions: true, PreserveToolCallIDs: true})
|
||||
ensureCodexOAuthInstructionsField(decoded)
|
||||
markDecodedModified()
|
||||
} else {
|
||||
codexResult = applyCodexOAuthTransform(decoded, isCodexCLI, isCompactRequest)
|
||||
}
|
||||
if codexResult.Modified {
|
||||
markDecodedModified()
|
||||
}
|
||||
// 带真实 device_id 时补齐 client_metadata 安装标识，与真实 Codex 对齐（compact 形态不同，跳过）。
|
||||
if !isCompactRequest && applyCodexClientMetadata(decoded, account) {
|
||||
markDecodedModified()
|
||||
}
|
||||
if codexResult.NormalizedModel != "" {
|
||||
upstreamModel = codexResult.NormalizedModel
|
||||
}
|
||||
if codexResult.PromptCacheKey != "" {
|
||||
promptCacheKey = codexResult.PromptCacheKey
|
||||
}
|
||||
}
|
||||
|
||||
if !SupportsVerbosity(upstreamModel) && gjson.GetBytes(body, "text.verbosity").Exists() {
|
||||
markPatchDelete("text.verbosity")
|
||||
}
|
||||
|
||||
if !isCodexCLI {
|
||||
maxOutputTokens := gjson.GetBytes(body, "max_output_tokens")
|
||||
if maxOutputTokens.Exists() {
|
||||
switch account.Platform {
|
||||
case PlatformOpenAI:
|
||||
if account.Type == AccountTypeAPIKey {
|
||||
markPatchDelete("max_output_tokens")
|
||||
}
|
||||
case PlatformAnthropic:
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
delete(decoded, "max_output_tokens")
|
||||
if _, hasMaxTokens := decoded["max_tokens"]; !hasMaxTokens {
|
||||
decoded["max_tokens"] = maxOutputTokens.Value()
|
||||
}
|
||||
markDecodedModified()
|
||||
case PlatformGemini:
|
||||
markPatchDelete("max_output_tokens")
|
||||
default:
|
||||
markPatchDelete("max_output_tokens")
|
||||
}
|
||||
}
|
||||
if gjson.GetBytes(body, "max_completion_tokens").Exists() && (account.Type == AccountTypeAPIKey || account.Platform != PlatformOpenAI) {
|
||||
markPatchDelete("max_completion_tokens")
|
||||
}
|
||||
for _, unsupportedField := range []string{"prompt_cache_retention", "safety_identifier"} {
|
||||
if gjson.GetBytes(body, unsupportedField).Exists() {
|
||||
markPatchDelete(unsupportedField)
|
||||
}
|
||||
}
|
||||
}
|
||||
if wsDecision.Transport != OpenAIUpstreamTransportResponsesWebsocketV2 && gjson.GetBytes(body, "previous_response_id").Exists() {
|
||||
markPatchDelete("previous_response_id")
|
||||
}
|
||||
if openAIRequestBodyMayContainEmptyBase64InputImage(body) {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if sanitizeEmptyBase64InputImagesInOpenAIRequestBodyMap(decoded) {
|
||||
markDecodedModified()
|
||||
}
|
||||
}
|
||||
|
||||
if rawTier := requestView.ServiceTier; rawTier != "" {
|
||||
if normTier := normalizedOpenAIServiceTierValue(rawTier); normTier != "" {
|
||||
action, errMsg := s.evaluateOpenAIFastPolicy(ctx, account, upstreamModel, normTier)
|
||||
switch action {
|
||||
case BetaPolicyActionBlock:
|
||||
msg := errMsg
|
||||
if msg == "" {
|
||||
msg = fmt.Sprintf("openai service_tier=%s is not allowed for model %s", normTier, upstreamModel)
|
||||
}
|
||||
blocked := &OpenAIFastBlockedError{Message: msg}
|
||||
writeOpenAIFastPolicyBlockedResponse(c, blocked)
|
||||
return nil, blocked
|
||||
case BetaPolicyActionFilter:
|
||||
markPatchDelete("service_tier")
|
||||
case OpenAIFastPolicyActionForcePriority:
|
||||
if rawTier != OpenAIFastTierPriority {
|
||||
markPatchSet("service_tier", OpenAIFastTierPriority)
|
||||
}
|
||||
default:
|
||||
if normTier != rawTier {
|
||||
markPatchSet("service_tier", normTier)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if bodyModified {
|
||||
if requestView.HasPatches() {
|
||||
if patchedBody, patchErr := requestView.ApplyPatches(); patchErr == nil {
|
||||
body = patchedBody
|
||||
requestView = newOpenAIRequestView(body)
|
||||
reqBody = nil
|
||||
bodyModified = false
|
||||
}
|
||||
}
|
||||
if bodyModified {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
var marshalErr error
|
||||
body, marshalErr = marshalOpenAIUpstreamJSON(decoded)
|
||||
if marshalErr != nil {
|
||||
return nil, fmt.Errorf("serialize request body: %w", marshalErr)
|
||||
}
|
||||
requestView = newOpenAIRequestView(body)
|
||||
}
|
||||
}
|
||||
imageBillingModel := ""
|
||||
imageSizeTier := ""
|
||||
imageInputSize := ""
|
||||
if imageIntent {
|
||||
var imageCfg OpenAIResponsesImageBillingConfig
|
||||
var imageCfgErr error
|
||||
if reqBody != nil {
|
||||
imageCfg, imageCfgErr = resolveOpenAIResponsesImageBillingConfigDetailed(reqBody, billingModel)
|
||||
} else {
|
||||
imageCfg, imageCfgErr = resolveOpenAIResponsesImageBillingConfigDetailedFromBody(body, billingModel)
|
||||
}
|
||||
if imageCfgErr != nil {
|
||||
setOpsUpstreamError(c, http.StatusBadRequest, imageCfgErr.Error(), "")
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": gin.H{"type": "invalid_request_error", "message": imageCfgErr.Error(), "param": "size"}})
|
||||
return nil, imageCfgErr
|
||||
}
|
||||
imageBillingModel = imageCfg.Model
|
||||
imageSizeTier = imageCfg.SizeTier
|
||||
imageInputSize = imageCfg.InputSize
|
||||
}
|
||||
|
||||
// Get access token
|
||||
token, _, err := s.GetAccessToken(ctx, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 命中 WS 时仅走 WebSocket Mode；不再自动回退 HTTP。
|
||||
if wsDecision.Transport == OpenAIUpstreamTransportResponsesWebsocketV2 {
|
||||
// WS 分支需要结构化 payload 与重连恢复，命中后再触发 full-map decode。
|
||||
wsReqBody, err := ensureReqBody()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, hasPreviousResponseID := wsReqBody["previous_response_id"]
|
||||
logOpenAIWSModeDebug(
|
||||
"forward_start account_id=%d account_type=%s model=%s stream=%v has_previous_response_id=%v",
|
||||
account.ID,
|
||||
account.Type,
|
||||
upstreamModel,
|
||||
reqStream,
|
||||
hasPreviousResponseID,
|
||||
)
|
||||
maxAttempts := openAIWSReconnectRetryLimit + 1
|
||||
wsAttempts := 0
|
||||
var wsResult *OpenAIForwardResult
|
||||
var wsErr error
|
||||
wsLastFailureReason := ""
|
||||
wsPrevResponseRecoveryTried := false
|
||||
wsInvalidEncryptedContentRecoveryTried := false
|
||||
recoverPrevResponseNotFound := func(attempt int) bool {
|
||||
if wsPrevResponseRecoveryTried {
|
||||
return false
|
||||
}
|
||||
previousResponseID := openAIWSPayloadString(wsReqBody, "previous_response_id")
|
||||
if previousResponseID == "" {
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_prev_response_recovery_skip account_id=%d attempt=%d reason=missing_previous_response_id previous_response_id_present=false",
|
||||
account.ID,
|
||||
attempt,
|
||||
)
|
||||
return false
|
||||
}
|
||||
if HasFunctionCallOutput(wsReqBody) {
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_prev_response_recovery_skip account_id=%d attempt=%d reason=has_function_call_output previous_response_id_present=true",
|
||||
account.ID,
|
||||
attempt,
|
||||
)
|
||||
return false
|
||||
}
|
||||
delete(wsReqBody, "previous_response_id")
|
||||
wsPrevResponseRecoveryTried = true
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_prev_response_recovery account_id=%d attempt=%d action=drop_previous_response_id retry=1 previous_response_id=%s previous_response_id_kind=%s",
|
||||
account.ID,
|
||||
attempt,
|
||||
truncateOpenAIWSLogValue(previousResponseID, openAIWSIDValueMaxLen),
|
||||
normalizeOpenAIWSLogValue(ClassifyOpenAIPreviousResponseIDKind(previousResponseID)),
|
||||
)
|
||||
return true
|
||||
}
|
||||
recoverInvalidEncryptedContent := func(attempt int) bool {
|
||||
if wsInvalidEncryptedContentRecoveryTried {
|
||||
return false
|
||||
}
|
||||
removedReasoningItems := trimOpenAIEncryptedReasoningItems(wsReqBody)
|
||||
if !removedReasoningItems {
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_invalid_encrypted_content_recovery_skip account_id=%d attempt=%d reason=missing_encrypted_reasoning_items",
|
||||
account.ID,
|
||||
attempt,
|
||||
)
|
||||
return false
|
||||
}
|
||||
previousResponseID := openAIWSPayloadString(wsReqBody, "previous_response_id")
|
||||
hasFunctionCallOutput := HasFunctionCallOutput(wsReqBody)
|
||||
if previousResponseID != "" && !hasFunctionCallOutput {
|
||||
delete(wsReqBody, "previous_response_id")
|
||||
}
|
||||
wsInvalidEncryptedContentRecoveryTried = true
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_invalid_encrypted_content_recovery account_id=%d attempt=%d action=drop_encrypted_reasoning_items retry=1 previous_response_id_present=%v previous_response_id=%s previous_response_id_kind=%s has_function_call_output=%v dropped_previous_response_id=%v",
|
||||
account.ID,
|
||||
attempt,
|
||||
previousResponseID != "",
|
||||
truncateOpenAIWSLogValue(previousResponseID, openAIWSIDValueMaxLen),
|
||||
normalizeOpenAIWSLogValue(ClassifyOpenAIPreviousResponseIDKind(previousResponseID)),
|
||||
hasFunctionCallOutput,
|
||||
previousResponseID != "" && !hasFunctionCallOutput,
|
||||
)
|
||||
return true
|
||||
}
|
||||
retryBudget := s.openAIWSRetryTotalBudget()
|
||||
retryStartedAt := time.Now()
|
||||
wsRetryLoop:
|
||||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||||
wsAttempts = attempt
|
||||
wsResult, wsErr = s.forwardOpenAIWSV2(
|
||||
ctx,
|
||||
c,
|
||||
account,
|
||||
wsReqBody,
|
||||
token,
|
||||
wsDecision,
|
||||
isCodexCLI,
|
||||
reqStream,
|
||||
originalModel,
|
||||
upstreamModel,
|
||||
startTime,
|
||||
attempt,
|
||||
wsLastFailureReason,
|
||||
)
|
||||
if wsErr == nil {
|
||||
break
|
||||
}
|
||||
if c != nil && c.Writer != nil && c.Writer.Written() {
|
||||
break
|
||||
}
|
||||
|
||||
reason, retryable := classifyOpenAIWSReconnectReason(wsErr)
|
||||
if reason != "" {
|
||||
wsLastFailureReason = reason
|
||||
}
|
||||
// previous_response_not_found 说明续链锚点不可用：
|
||||
// 对非 function_call_output 场景，允许一次“去掉 previous_response_id 后重放”。
|
||||
if reason == "previous_response_not_found" && recoverPrevResponseNotFound(attempt) {
|
||||
continue
|
||||
}
|
||||
if reason == "invalid_encrypted_content" && recoverInvalidEncryptedContent(attempt) {
|
||||
continue
|
||||
}
|
||||
if retryable && attempt < maxAttempts {
|
||||
backoff := s.openAIWSRetryBackoff(attempt)
|
||||
if retryBudget > 0 && time.Since(retryStartedAt)+backoff > retryBudget {
|
||||
s.recordOpenAIWSRetryExhausted()
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_budget_exhausted account_id=%d attempts=%d max_retries=%d reason=%s elapsed_ms=%d budget_ms=%d",
|
||||
account.ID,
|
||||
attempt,
|
||||
openAIWSReconnectRetryLimit,
|
||||
normalizeOpenAIWSLogValue(reason),
|
||||
time.Since(retryStartedAt).Milliseconds(),
|
||||
retryBudget.Milliseconds(),
|
||||
)
|
||||
break
|
||||
}
|
||||
s.recordOpenAIWSRetryAttempt(backoff)
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_retry account_id=%d retry=%d max_retries=%d reason=%s backoff_ms=%d",
|
||||
account.ID,
|
||||
attempt,
|
||||
openAIWSReconnectRetryLimit,
|
||||
normalizeOpenAIWSLogValue(reason),
|
||||
backoff.Milliseconds(),
|
||||
)
|
||||
if backoff > 0 {
|
||||
timer := time.NewTimer(backoff)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if !timer.Stop() {
|
||||
<-timer.C
|
||||
}
|
||||
wsErr = wrapOpenAIWSFallback("retry_backoff_canceled", ctx.Err())
|
||||
break wsRetryLoop
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if retryable {
|
||||
s.recordOpenAIWSRetryExhausted()
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_exhausted account_id=%d attempts=%d max_retries=%d reason=%s",
|
||||
account.ID,
|
||||
attempt,
|
||||
openAIWSReconnectRetryLimit,
|
||||
normalizeOpenAIWSLogValue(reason),
|
||||
)
|
||||
} else if reason != "" {
|
||||
s.recordOpenAIWSNonRetryableFastFallback()
|
||||
logOpenAIWSModeInfo(
|
||||
"reconnect_stop account_id=%d attempt=%d reason=%s",
|
||||
account.ID,
|
||||
attempt,
|
||||
normalizeOpenAIWSLogValue(reason),
|
||||
)
|
||||
}
|
||||
break
|
||||
}
|
||||
if wsErr == nil {
|
||||
firstTokenMs := int64(0)
|
||||
hasFirstTokenMs := wsResult != nil && wsResult.FirstTokenMs != nil
|
||||
if hasFirstTokenMs {
|
||||
firstTokenMs = int64(*wsResult.FirstTokenMs)
|
||||
}
|
||||
requestID := ""
|
||||
if wsResult != nil {
|
||||
requestID = strings.TrimSpace(wsResult.RequestID)
|
||||
}
|
||||
logOpenAIWSModeDebug(
|
||||
"forward_succeeded account_id=%d request_id=%s stream=%v has_first_token_ms=%v first_token_ms=%d ws_attempts=%d",
|
||||
account.ID,
|
||||
requestID,
|
||||
reqStream,
|
||||
hasFirstTokenMs,
|
||||
firstTokenMs,
|
||||
wsAttempts,
|
||||
)
|
||||
wsResult.UpstreamModel = upstreamModel
|
||||
if wsResult.BillingModel == "" {
|
||||
wsResult.BillingModel = billingModel
|
||||
}
|
||||
if wsResult.ImageCount > 0 {
|
||||
wsResult.ImageSize = imageSizeTier
|
||||
wsResult.ImageInputSize = imageInputSize
|
||||
wsResult.BillingModel = imageBillingModel
|
||||
}
|
||||
return wsResult, nil
|
||||
}
|
||||
s.writeOpenAIWSFallbackErrorResponse(c, account, wsErr)
|
||||
return nil, wsErr
|
||||
}
|
||||
|
||||
httpInvalidEncryptedContentRetryTried := false
|
||||
for {
|
||||
// Build upstream request
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
upstreamReq, err := s.buildUpstreamRequest(upstreamCtx, c, account, body, token, reqStream, promptCacheKey, isCodexCLI)
|
||||
releaseUpstreamCtx()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Get proxy URL
|
||||
proxyURL := ""
|
||||
if account.ProxyID != nil && account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
|
||||
// Send request
|
||||
upstreamStart := time.Now()
|
||||
resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
|
||||
SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
|
||||
if err != nil {
|
||||
// Transport-level failure (proxy/DNS/TCP/TLS — no HTTP response). Convert to
|
||||
// a failover so the handler switches to a healthy account, and temporarily
|
||||
// unschedule the account on durable faults (e.g. rejected proxy credentials).
|
||||
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
|
||||
}
|
||||
|
||||
// Handle error response
|
||||
if resp.StatusCode >= 400 {
|
||||
respBody := s.readUpstreamErrorBody(resp)
|
||||
_ = resp.Body.Close()
|
||||
resp.Body = io.NopCloser(bytes.NewReader(respBody))
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
upstreamCode := extractUpstreamErrorCode(respBody)
|
||||
if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" {
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
return nil, decodeErr
|
||||
}
|
||||
if trimOpenAIEncryptedReasoningItems(decoded) {
|
||||
body, err = marshalOpenAIUpstreamJSON(decoded)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("serialize invalid_encrypted_content retry body: %w", err)
|
||||
}
|
||||
httpInvalidEncryptedContentRetryTried = true
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying non-WSv2 request once after invalid_encrypted_content (account: %s)", account.Name)
|
||||
continue
|
||||
}
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Skip non-WSv2 invalid_encrypted_content retry because encrypted reasoning items are missing (account: %s)", account.Name)
|
||||
}
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(respBody), maxBytes)
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "failover",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
|
||||
s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel)
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: respBody,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
|
||||
}
|
||||
}
|
||||
return s.handleErrorResponse(ctx, resp, c, account, body, billingModel)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
reasoningEffort := extractOpenAIReasoningEffortFromBody(body, originalModel)
|
||||
// 国产模型默认 effort 补充：此处 reqModel 已被 mapping 重写为 billingModel（见
|
||||
// line 2510-2515 的 GetMappedModel + reqModel 赋值），可直接作为 mappedModel。
|
||||
reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, reqModel)
|
||||
serviceTier := extractOpenAIServiceTierFromBody(body)
|
||||
// 上游接受后只保留计费需要的标量，避免响应处理期间继续保活完整 input/tools map。
|
||||
reqBody = nil
|
||||
|
||||
// Handle normal response
|
||||
var usage *OpenAIUsage
|
||||
var firstTokenMs *int
|
||||
responseID := ""
|
||||
imageCount := 0
|
||||
var imageOutputSizes []string
|
||||
if reqStream {
|
||||
streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, upstreamModel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usage = streamResult.usage
|
||||
firstTokenMs = streamResult.firstTokenMs
|
||||
responseID = strings.TrimSpace(streamResult.responseID)
|
||||
imageCount = streamResult.imageCount
|
||||
imageOutputSizes = streamResult.imageOutputSizes
|
||||
} else {
|
||||
nonStreamResult, err := s.handleNonStreamingResponse(ctx, resp, c, account, originalModel, upstreamModel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usage = nonStreamResult.usage
|
||||
responseID = strings.TrimSpace(nonStreamResult.responseID)
|
||||
imageCount = nonStreamResult.imageCount
|
||||
imageOutputSizes = nonStreamResult.imageOutputSizes
|
||||
}
|
||||
s.bindHTTPResponseAccount(ctx, c, account, responseID)
|
||||
|
||||
// Extract and save Codex usage snapshot from response headers (for OAuth accounts).
|
||||
// 排除 spark 影子:其 codex_* 仅由 QueryUsage(/wham/usage bengalfox)更新(外审第7轮 P1)。
|
||||
if account.Type == AccountTypeOAuth && !account.IsShadow() {
|
||||
if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil {
|
||||
s.updateCodexUsageSnapshot(ctx, account.ID, snapshot)
|
||||
}
|
||||
}
|
||||
|
||||
if usage == nil {
|
||||
usage = &OpenAIUsage{}
|
||||
}
|
||||
|
||||
forwardResult := &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
ResponseID: responseID,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
ServiceTier: serviceTier,
|
||||
ReasoningEffort: reasoningEffort,
|
||||
Stream: reqStream,
|
||||
OpenAIWSMode: false,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: firstTokenMs,
|
||||
}
|
||||
if imageCount > 0 {
|
||||
forwardResult.ImageCount = imageCount
|
||||
forwardResult.ImageSize = imageSizeTier
|
||||
forwardResult.ImageInputSize = imageInputSize
|
||||
forwardResult.ImageOutputSizes = imageOutputSizes
|
||||
forwardResult.BillingModel = imageBillingModel
|
||||
}
|
||||
return forwardResult, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.Context, account *Account, body []byte, token string, isStream bool, promptCacheKey string, isCodexCLI bool) (*http.Request, error) {
|
||||
// Determine target URL based on account type
|
||||
var targetURL string
|
||||
switch account.Type {
|
||||
case AccountTypeOAuth:
|
||||
// OAuth accounts use ChatGPT internal API
|
||||
targetURL = chatgptCodexURL
|
||||
case AccountTypeAPIKey:
|
||||
// API Key accounts use Platform API or custom base URL
|
||||
baseURL := account.GetOpenAIBaseURL()
|
||||
if baseURL == "" {
|
||||
targetURL = openaiPlatformAPIURL
|
||||
} else {
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
targetURL = buildOpenAIResponsesURL(validatedURL)
|
||||
}
|
||||
default:
|
||||
targetURL = openaiPlatformAPIURL
|
||||
}
|
||||
targetURL = appendOpenAIResponsesRequestPathSuffix(targetURL, openAIResponsesRequestPathSuffix(c))
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", targetURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
|
||||
|
||||
// Set authentication header
|
||||
req.Header.Set("authorization", "Bearer "+token)
|
||||
|
||||
// Set headers specific to OAuth accounts (ChatGPT internal API)
|
||||
if account.Type == AccountTypeOAuth {
|
||||
// Required: set Host for ChatGPT API (must use req.Host, not Header.Set)
|
||||
req.Host = "chatgpt.com"
|
||||
if err := resolveAndSetOpenAIChatGPTAccountHeaders(ctx, s.accountRepo, req.Header, account); err != nil {
|
||||
return nil, fmt.Errorf("resolve chatgpt account headers: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Whitelist passthrough headers
|
||||
for key, values := range c.Request.Header {
|
||||
lowerKey := strings.ToLower(key)
|
||||
if openaiAllowedHeaders[lowerKey] {
|
||||
for _, v := range values {
|
||||
req.Header.Add(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
if account.Type == AccountTypeOAuth {
|
||||
compatMessagesBridge := isOpenAICompatMessagesBridgeContext(c) || isOpenAICompatMessagesBridgeBody(body)
|
||||
// 清除客户端透传的 session 头，后续用隔离后的值重新设置，防止跨用户会话碰撞。
|
||||
clientConversationID := strings.TrimSpace(req.Header.Get("conversation_id"))
|
||||
req.Header.Del("conversation_id")
|
||||
req.Header.Del("session_id")
|
||||
|
||||
if compatMessagesBridge {
|
||||
req.Header.Del("OpenAI-Beta")
|
||||
req.Header.Del("originator")
|
||||
} else {
|
||||
req.Header.Set("OpenAI-Beta", "responses=experimental")
|
||||
req.Header.Set("originator", resolveOpenAIUpstreamOriginator(c, isCodexCLI))
|
||||
}
|
||||
apiKeyID := getAPIKeyIDFromContext(c)
|
||||
if isOpenAIResponsesCompactPath(c) {
|
||||
req.Header.Set("accept", "application/json")
|
||||
if req.Header.Get("version") == "" {
|
||||
req.Header.Set("version", codexCLIVersion)
|
||||
}
|
||||
compactSession := resolveOpenAICompactSessionID(c)
|
||||
req.Header.Set("session_id", isolateOpenAISessionID(apiKeyID, compactSession))
|
||||
} else {
|
||||
req.Header.Set("accept", "text/event-stream")
|
||||
}
|
||||
if promptCacheKey != "" {
|
||||
isolated := isolateOpenAISessionID(apiKeyID, promptCacheKey)
|
||||
req.Header.Set("session_id", isolated)
|
||||
if !compatMessagesBridge || clientConversationID != "" {
|
||||
req.Header.Set("conversation_id", isolated)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Apply custom User-Agent if configured
|
||||
customUA := account.GetOpenAIUserAgent()
|
||||
if customUA != "" {
|
||||
req.Header.Set("user-agent", customUA)
|
||||
}
|
||||
|
||||
// 若开启 ForceCodexCLI，则强制将上游 User-Agent 伪装为 Codex CLI。
|
||||
// 用于网关未透传/改写 User-Agent 时，仍能命中 Codex 侧识别逻辑。
|
||||
if s.cfg != nil && s.cfg.Gateway.ForceCodexCLI {
|
||||
req.Header.Set("user-agent", codexCLIUserAgent)
|
||||
}
|
||||
|
||||
// 浏览器型 UA 兜底：仅 OAuth（ChatGPT 内部接口）账号生效，若最终 user-agent 仍为浏览器
|
||||
// （Chrome/Firefox/Safari/Edge 等），替换为后台配置的 Codex UA，避免 Cloudflare 触发 JS 质询。
|
||||
s.overrideBrowserUserAgent(ctx, account, req)
|
||||
|
||||
// Ensure required headers exist
|
||||
if req.Header.Get("content-type") == "" {
|
||||
req.Header.Set("content-type", "application/json")
|
||||
}
|
||||
|
||||
// 账号级请求头覆写（仅 openai api_key 账号启用时生效；OAuth 路径 no-op）
|
||||
account.ApplyHeaderOverrides(req.Header)
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// overrideBrowserUserAgent 检查请求的最终 user-agent，若为浏览器 UA 则替换为后台配置的 Codex UA。
|
||||
// 用于规避 Cloudflare 对浏览器型 UA 在 ChatGPT 内部接口上的访问质询。
|
||||
// 影响范围严格限定：仅 OAuth（Codex/ChatGPT 内部接口）账号生效；API Key 等其他账号原样透传。
|
||||
// 仅在识别为浏览器（Mozilla/...）时改写，其他 CLI/工具 UA 不动。
|
||||
func (s *OpenAIGatewayService) overrideBrowserUserAgent(ctx context.Context, account *Account, req *http.Request) {
|
||||
if req == nil || account == nil {
|
||||
return
|
||||
}
|
||||
if account.Type != AccountTypeOAuth {
|
||||
return
|
||||
}
|
||||
currentUA := req.Header.Get("user-agent")
|
||||
if !openai.IsBrowserUserAgent(currentUA) {
|
||||
return
|
||||
}
|
||||
codexUA := DefaultOpenAICodexUserAgent
|
||||
if s != nil && s.settingService != nil {
|
||||
if v := strings.TrimSpace(s.settingService.GetOpenAICodexUserAgent(ctx)); v != "" {
|
||||
codexUA = v
|
||||
}
|
||||
}
|
||||
req.Header.Set("user-agent", codexUA)
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,597 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/openai"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/tidwall/gjson"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func logOpenAIInstructionsRequiredDebug(
|
||||
ctx context.Context,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
upstreamStatusCode int,
|
||||
upstreamMsg string,
|
||||
requestBody []byte,
|
||||
upstreamBody []byte,
|
||||
) {
|
||||
msg := strings.TrimSpace(upstreamMsg)
|
||||
if !isOpenAIInstructionsRequiredError(upstreamStatusCode, msg, upstreamBody) {
|
||||
return
|
||||
}
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
|
||||
accountID := int64(0)
|
||||
accountName := ""
|
||||
if account != nil {
|
||||
accountID = account.ID
|
||||
accountName = strings.TrimSpace(account.Name)
|
||||
}
|
||||
|
||||
userAgent := ""
|
||||
originator := ""
|
||||
if c != nil {
|
||||
userAgent = strings.TrimSpace(c.GetHeader("User-Agent"))
|
||||
originator = strings.TrimSpace(c.GetHeader("originator"))
|
||||
}
|
||||
|
||||
fields := []zap.Field{
|
||||
zap.String("component", "service.openai_gateway"),
|
||||
zap.Int64("account_id", accountID),
|
||||
zap.String("account_name", accountName),
|
||||
zap.Int("upstream_status_code", upstreamStatusCode),
|
||||
zap.String("upstream_error_message", msg),
|
||||
zap.String("request_user_agent", userAgent),
|
||||
zap.Bool("codex_official_client_match", openai.IsCodexOfficialClientByHeaders(userAgent, originator)),
|
||||
}
|
||||
fields = appendCodexCLIOnlyRejectedRequestFields(fields, c, requestBody)
|
||||
|
||||
logger.FromContext(ctx).With(fields...).Warn("OpenAI 上游返回 Instructions are required，已记录请求详情用于排查")
|
||||
}
|
||||
|
||||
func isOpenAIInstructionsRequiredError(upstreamStatusCode int, upstreamMsg string, upstreamBody []byte) bool {
|
||||
if upstreamStatusCode != http.StatusBadRequest {
|
||||
return false
|
||||
}
|
||||
|
||||
hasInstructionRequired := func(text string) bool {
|
||||
lower := strings.ToLower(strings.TrimSpace(text))
|
||||
if lower == "" {
|
||||
return false
|
||||
}
|
||||
if strings.Contains(lower, "instructions are required") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "required parameter: 'instructions'") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "required parameter: instructions") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "missing required parameter") && strings.Contains(lower, "instructions") {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(lower, "instruction") && strings.Contains(lower, "required")
|
||||
}
|
||||
|
||||
if hasInstructionRequired(upstreamMsg) {
|
||||
return true
|
||||
}
|
||||
if len(upstreamBody) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
errMsg := gjson.GetBytes(upstreamBody, "error.message").String()
|
||||
errMsgLower := strings.ToLower(strings.TrimSpace(errMsg))
|
||||
errCode := strings.ToLower(strings.TrimSpace(gjson.GetBytes(upstreamBody, "error.code").String()))
|
||||
errParam := strings.ToLower(strings.TrimSpace(gjson.GetBytes(upstreamBody, "error.param").String()))
|
||||
errType := strings.ToLower(strings.TrimSpace(gjson.GetBytes(upstreamBody, "error.type").String()))
|
||||
|
||||
if errParam == "instructions" {
|
||||
return true
|
||||
}
|
||||
if hasInstructionRequired(errMsg) {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(errCode, "missing_required_parameter") && strings.Contains(errMsgLower, "instructions") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(errType, "invalid_request") && strings.Contains(errMsgLower, "instructions") && strings.Contains(errMsgLower, "required") {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func isOpenAITransientProcessingError(upstreamStatusCode int, upstreamMsg string, upstreamBody []byte) bool {
|
||||
if upstreamStatusCode != http.StatusBadRequest && upstreamStatusCode != http.StatusServiceUnavailable {
|
||||
return false
|
||||
}
|
||||
|
||||
hasOpenAIServerOverloadedCode := func(payload []byte) bool {
|
||||
code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "error.code").String()))
|
||||
if code == "" {
|
||||
code = strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "response.error.code").String()))
|
||||
}
|
||||
return code == "server_is_overloaded" || code == "slow_down"
|
||||
}
|
||||
|
||||
if len(upstreamBody) > 0 && hasOpenAIServerOverloadedCode(upstreamBody) {
|
||||
return true
|
||||
}
|
||||
if upstreamStatusCode != http.StatusBadRequest {
|
||||
return false
|
||||
}
|
||||
|
||||
match := func(text string) bool {
|
||||
lower := strings.ToLower(strings.TrimSpace(text))
|
||||
if lower == "" {
|
||||
return false
|
||||
}
|
||||
if strings.Contains(lower, "an error occurred while processing your request") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "selected model is at capacity") {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(lower, "you can retry your request") &&
|
||||
strings.Contains(lower, "help.openai.com") &&
|
||||
strings.Contains(lower, "request id")
|
||||
}
|
||||
|
||||
if match(upstreamMsg) {
|
||||
return true
|
||||
}
|
||||
if len(upstreamBody) == 0 {
|
||||
return false
|
||||
}
|
||||
if match(gjson.GetBytes(upstreamBody, "error.message").String()) {
|
||||
return true
|
||||
}
|
||||
return match(string(upstreamBody))
|
||||
}
|
||||
|
||||
func isOpenAIContextWindowError(upstreamMsg string, upstreamBody []byte) bool {
|
||||
match := func(text string) bool {
|
||||
lower := strings.ToLower(strings.TrimSpace(text))
|
||||
if lower == "" {
|
||||
return false
|
||||
}
|
||||
if strings.Contains(lower, "context_too_large") || strings.Contains(lower, "context_length_exceeded") {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "maximum context length") || strings.Contains(lower, "max context length") {
|
||||
return true
|
||||
}
|
||||
hasExceeded := strings.Contains(lower, "exceed") || strings.Contains(lower, "too large") || strings.Contains(lower, "too long")
|
||||
if strings.Contains(lower, "context window") && hasExceeded {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(lower, "context length") && hasExceeded {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(lower, "token limit") &&
|
||||
strings.Contains(lower, "context") &&
|
||||
hasExceeded
|
||||
}
|
||||
|
||||
if match(upstreamMsg) {
|
||||
return true
|
||||
}
|
||||
if len(upstreamBody) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, path := range []string{
|
||||
"error.message",
|
||||
"response.error.message",
|
||||
"message",
|
||||
"error.code",
|
||||
"response.error.code",
|
||||
"code",
|
||||
} {
|
||||
if match(gjson.GetBytes(upstreamBody, path).String()) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return match(string(upstreamBody))
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) shouldFailoverUpstreamError(statusCode int) bool {
|
||||
switch statusCode {
|
||||
case 401, 402, 403, 429, 529:
|
||||
return true
|
||||
default:
|
||||
return statusCode >= 500
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) shouldFailoverOpenAIUpstreamResponse(statusCode int, upstreamMsg string, upstreamBody []byte) bool {
|
||||
if isOpenAIContextWindowError(upstreamMsg, upstreamBody) {
|
||||
return false
|
||||
}
|
||||
if s.shouldFailoverUpstreamError(statusCode) {
|
||||
return true
|
||||
}
|
||||
return isOpenAITransientProcessingError(statusCode, upstreamMsg, upstreamBody)
|
||||
}
|
||||
|
||||
func marshalOpenAIUpstreamJSON(v any) ([]byte, error) {
|
||||
var buf bytes.Buffer
|
||||
enc := json.NewEncoder(&buf)
|
||||
enc.SetEscapeHTML(false)
|
||||
if err := enc.Encode(v); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := buf.Bytes()
|
||||
if len(out) > 0 && out[len(out)-1] == '\n' {
|
||||
out = out[:len(out)-1]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func openAIUpstreamErrorBodyReadLimitForConfig(cfg *config.Config) int64 {
|
||||
limit := openAIUpstreamErrorBodyReadLimit
|
||||
if cfg != nil && cfg.Gateway.LogUpstreamErrorBody && cfg.Gateway.LogUpstreamErrorBodyMaxBytes > int(limit) {
|
||||
limit = int64(cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
|
||||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) readUpstreamErrorBody(resp *http.Response) []byte {
|
||||
if resp == nil || resp.Body == nil {
|
||||
return nil
|
||||
}
|
||||
cfg := (*config.Config)(nil)
|
||||
if s != nil {
|
||||
cfg = s.cfg
|
||||
}
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, openAIUpstreamErrorBodyReadLimitForConfig(cfg)))
|
||||
return body
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, responseBody []byte, requestedModel ...string) {
|
||||
if len(requestedModel) > 0 {
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody, requestedModel[0])
|
||||
return
|
||||
}
|
||||
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody)
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) handleErrorResponse(
|
||||
ctx context.Context,
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
requestBody []byte,
|
||||
requestedModel ...string,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
body := s.readUpstreamErrorBody(resp)
|
||||
|
||||
// cyber_policy 硬阻断：透传上游原始错误体给客户端（不重包成通用 502），不冷却账号。
|
||||
// 当前请求恒透传（需求1）；标记供 handler 事后写风控/邮件。400 cyber 不可 failover
|
||||
// （shouldFailoverUpstreamError(400)=false），故走到此处即可安全早返回。
|
||||
if hit, code, cyberMsg := detectOpenAICyberPolicy(body); hit {
|
||||
MarkOpsCyberPolicy(c, CyberPolicyMark{
|
||||
Code: code,
|
||||
Message: cyberMsg,
|
||||
Body: truncateString(string(body), 4096),
|
||||
UpstreamStatus: resp.StatusCode,
|
||||
})
|
||||
setOpsUpstreamError(c, resp.StatusCode, cyberMsg, truncateString(string(body), 2048))
|
||||
writeOpenAIPassthroughResponseHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
|
||||
contentType := resp.Header.Get("Content-Type")
|
||||
if contentType == "" {
|
||||
contentType = "application/json"
|
||||
}
|
||||
c.Data(resp.StatusCode, contentType, body)
|
||||
if cyberMsg == "" {
|
||||
return nil, fmt.Errorf("openai cyber_policy: %d", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("openai cyber_policy: %s", cyberMsg)
|
||||
}
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(body), maxBytes)
|
||||
}
|
||||
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
|
||||
logOpenAIInstructionsRequiredDebug(ctx, c, account, resp.StatusCode, upstreamMsg, requestBody, body)
|
||||
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
logger.LegacyPrintf("service.openai_gateway",
|
||||
"OpenAI upstream error %d (account=%d platform=%s type=%s): %s",
|
||||
resp.StatusCode,
|
||||
account.ID,
|
||||
account.Platform,
|
||||
account.Type,
|
||||
truncateForLog(body, s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes),
|
||||
)
|
||||
}
|
||||
|
||||
if status, errType, errMsg, matched := applyErrorPassthroughRule(
|
||||
c,
|
||||
PlatformOpenAI,
|
||||
resp.StatusCode,
|
||||
body,
|
||||
http.StatusBadGateway,
|
||||
"upstream_error",
|
||||
"Upstream request failed",
|
||||
); matched {
|
||||
MarkResponseCommitted(c)
|
||||
c.JSON(status, gin.H{
|
||||
"error": gin.H{
|
||||
"type": errType,
|
||||
"message": errMsg,
|
||||
},
|
||||
})
|
||||
if upstreamMsg == "" {
|
||||
upstreamMsg = errMsg
|
||||
}
|
||||
if upstreamMsg == "" {
|
||||
return nil, fmt.Errorf("upstream error: %d (passthrough rule matched)", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("upstream error: %d (passthrough rule matched) message=%s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
|
||||
// Check custom error codes
|
||||
if !account.ShouldHandleErrorCode(resp.StatusCode) {
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "http_error",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
MarkResponseCommitted(c)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{
|
||||
"error": gin.H{
|
||||
"type": "upstream_error",
|
||||
"message": "Upstream gateway error",
|
||||
},
|
||||
})
|
||||
if upstreamMsg == "" {
|
||||
return nil, fmt.Errorf("upstream error: %d (not in custom error codes)", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("upstream error: %d (not in custom error codes) message=%s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
|
||||
// Handle upstream error (mark account status)
|
||||
var reqModel string
|
||||
if len(requestedModel) > 0 {
|
||||
reqModel = strings.TrimSpace(requestedModel[0])
|
||||
}
|
||||
if reqModel == "" {
|
||||
reqModel, _, _ = extractOpenAIRequestMetaFromBody(requestBody)
|
||||
}
|
||||
shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, reqModel)
|
||||
kind := "http_error"
|
||||
if shouldDisable {
|
||||
kind = "failover"
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: kind,
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
if shouldDisable {
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: body,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
|
||||
MarkResponseCommitted(c)
|
||||
|
||||
// Return appropriate error response
|
||||
var errType, errMsg string
|
||||
var statusCode int
|
||||
|
||||
switch resp.StatusCode {
|
||||
case 401:
|
||||
statusCode = http.StatusBadGateway
|
||||
errType = "upstream_error"
|
||||
errMsg = "Upstream authentication failed, please contact administrator"
|
||||
case 402:
|
||||
statusCode = http.StatusBadGateway
|
||||
errType = "upstream_error"
|
||||
errMsg = "Upstream payment required: insufficient balance or billing issue"
|
||||
case 403:
|
||||
statusCode = http.StatusBadGateway
|
||||
errType = "upstream_error"
|
||||
errMsg = "Upstream access forbidden, please contact administrator"
|
||||
case 429:
|
||||
statusCode = http.StatusTooManyRequests
|
||||
errType = "rate_limit_error"
|
||||
errMsg = "Upstream rate limit exceeded, please retry later"
|
||||
default:
|
||||
statusCode = http.StatusBadGateway
|
||||
errType = "upstream_error"
|
||||
errMsg = "Upstream request failed"
|
||||
}
|
||||
if isOpenAIContextWindowError(upstreamMsg, body) && upstreamMsg != "" {
|
||||
errMsg = upstreamMsg
|
||||
}
|
||||
|
||||
c.JSON(statusCode, gin.H{
|
||||
"error": gin.H{
|
||||
"type": errType,
|
||||
"message": errMsg,
|
||||
},
|
||||
})
|
||||
|
||||
if upstreamMsg == "" {
|
||||
return nil, fmt.Errorf("upstream error: %d", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("upstream error: %d message=%s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
|
||||
// compatErrorWriter is the signature for format-specific error writers used by
|
||||
// the compat paths (Chat Completions and Anthropic Messages).
|
||||
type compatErrorWriter func(c *gin.Context, statusCode int, errType, message string)
|
||||
|
||||
// handleCompatErrorResponse is the shared non-failover error handler for the
|
||||
// Chat Completions and Anthropic Messages compat paths. It mirrors the logic of
|
||||
// handleErrorResponse (passthrough rules, ShouldHandleErrorCode, rate-limit
|
||||
// tracking, secondary failover) but delegates the final error write to the
|
||||
// format-specific writer function.
|
||||
func (s *OpenAIGatewayService) handleCompatErrorResponse(
|
||||
resp *http.Response,
|
||||
c *gin.Context,
|
||||
account *Account,
|
||||
writeError compatErrorWriter,
|
||||
requestedModel ...string,
|
||||
) (*OpenAIForwardResult, error) {
|
||||
body := s.readUpstreamErrorBody(resp)
|
||||
|
||||
// cyber_policy：兼容路径（Chat Completions / Anthropic）以各自格式回写错误，
|
||||
// 不原样透传 responses 格式的 cyber body（否则对下游格式不合法）。cyber 是上游网络
|
||||
// 安全策略拦截，不冷却账号，故标记后直接以兼容格式回写错误并返回，跳过下方
|
||||
// handleOpenAIAccountUpstreamError（避免自定义 temp-unschedulable 规则误冷却）。
|
||||
if hit, code, cyberMsg := detectOpenAICyberPolicy(body); hit {
|
||||
MarkOpsCyberPolicy(c, CyberPolicyMark{
|
||||
Code: code,
|
||||
Message: cyberMsg,
|
||||
Body: truncateString(string(body), 4096),
|
||||
UpstreamStatus: resp.StatusCode,
|
||||
})
|
||||
setOpsUpstreamError(c, resp.StatusCode, cyberMsg, truncateString(string(body), 2048))
|
||||
clientMsg := cyberMsg
|
||||
if clientMsg == "" {
|
||||
clientMsg = "Request blocked by upstream cyber-security policy"
|
||||
}
|
||||
writeError(c, resp.StatusCode, "invalid_request_error", clientMsg)
|
||||
if cyberMsg == "" {
|
||||
return nil, fmt.Errorf("openai cyber_policy: %d", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("openai cyber_policy: %s", cyberMsg)
|
||||
}
|
||||
|
||||
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
|
||||
if upstreamMsg == "" {
|
||||
upstreamMsg = fmt.Sprintf("Upstream error: %d", resp.StatusCode)
|
||||
}
|
||||
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
|
||||
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 2048
|
||||
}
|
||||
upstreamDetail = truncateString(string(body), maxBytes)
|
||||
}
|
||||
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
|
||||
|
||||
// Apply error passthrough rules
|
||||
if status, errType, errMsg, matched := applyErrorPassthroughRule(
|
||||
c, account.Platform, resp.StatusCode, body,
|
||||
http.StatusBadGateway, "api_error", "Upstream request failed",
|
||||
); matched {
|
||||
MarkResponseCommitted(c)
|
||||
writeError(c, status, errType, errMsg)
|
||||
if upstreamMsg == "" {
|
||||
upstreamMsg = errMsg
|
||||
}
|
||||
if upstreamMsg == "" {
|
||||
return nil, fmt.Errorf("upstream error: %d (passthrough rule matched)", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("upstream error: %d (passthrough rule matched) message=%s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
|
||||
// Check custom error codes — if the account does not handle this status,
|
||||
// return a generic error without exposing upstream details.
|
||||
if !account.ShouldHandleErrorCode(resp.StatusCode) {
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: "http_error",
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
MarkResponseCommitted(c)
|
||||
writeError(c, http.StatusInternalServerError, "api_error", "Upstream gateway error")
|
||||
if upstreamMsg == "" {
|
||||
return nil, fmt.Errorf("upstream error: %d (not in custom error codes)", resp.StatusCode)
|
||||
}
|
||||
return nil, fmt.Errorf("upstream error: %d (not in custom error codes) message=%s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
|
||||
// Track rate limits and decide whether to trigger secondary failover.
|
||||
var modelForCooldown string
|
||||
if len(requestedModel) > 0 {
|
||||
modelForCooldown = requestedModel[0]
|
||||
}
|
||||
shouldDisable := s.handleOpenAIAccountUpstreamError(
|
||||
c.Request.Context(), account, resp.StatusCode, resp.Header, body, modelForCooldown,
|
||||
)
|
||||
kind := "http_error"
|
||||
if shouldDisable {
|
||||
kind = "failover"
|
||||
}
|
||||
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
|
||||
Platform: account.Platform,
|
||||
AccountID: account.ID,
|
||||
AccountName: account.Name,
|
||||
UpstreamStatusCode: resp.StatusCode,
|
||||
UpstreamRequestID: resp.Header.Get("x-request-id"),
|
||||
Kind: kind,
|
||||
Message: upstreamMsg,
|
||||
Detail: upstreamDetail,
|
||||
})
|
||||
if shouldDisable {
|
||||
return nil, &UpstreamFailoverError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ResponseBody: body,
|
||||
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
|
||||
}
|
||||
}
|
||||
|
||||
MarkResponseCommitted(c)
|
||||
|
||||
// Map status code to error type and write response
|
||||
errType := "api_error"
|
||||
switch {
|
||||
case resp.StatusCode == 400:
|
||||
errType = "invalid_request_error"
|
||||
case resp.StatusCode == 404:
|
||||
errType = "not_found_error"
|
||||
case resp.StatusCode == 429:
|
||||
errType = "rate_limit_error"
|
||||
case resp.StatusCode >= 500:
|
||||
errType = "api_error"
|
||||
}
|
||||
|
||||
writeError(c, resp.StatusCode, errType, upstreamMsg)
|
||||
return nil, fmt.Errorf("upstream error: %d %s", resp.StatusCode, upstreamMsg)
|
||||
}
|
||||
Reference in New Issue
Block a user