fix(openai): 原生 Responses 路径不再把上游确定性 400 归一成可重试 502

上游返回 400(invalid_function_parameters、missing_required_parameter 等)时,
`/v1/responses` 原生路径把它包成 `502 upstream_error / "Upstream request failed"`,
真实的 message/code/param 全部丢弃。

502 属于可重试类,下游网关会把这个永远不会成功的请求反复重放(issue #5479 实测
30 个失败请求被放大成 60 次上游调用),客户端也拿不到「哪个字段非法」的线索。

同一个 service 上的兄弟路径早已做对:
- `handleCompatErrorResponse`(ChatCompletions / Anthropic)回真实状态码 +
  invalid_request_error + 真实 message;
- `/v1/images/generations` 还额外透传 code/param。

原生 Responses 是唯一漏掉的一条,本次对齐。

改动只放行 400:走到该分支说明 `shouldFailoverOpenAIUpstreamResponse` 已判定这个
400 不可 failover,即 server_is_overloaded / at capacity 这类可重试的 400 不会到达。
401/402/403(运营方凭据问题)、404/405(可能是 base_url 配错)、429 一律维持原状。

Fixes #5479
This commit is contained in:
li
2026-08-10 11:25:46 +08:00
parent 10a4c6e3ad
commit 591d47fb9b
4 changed files with 376 additions and 1 deletions
@@ -12,6 +12,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
type stubCodexRestrictionDetector struct {
@@ -365,7 +366,12 @@ func TestOpenAIGatewayService_Forward_LogsInstructionsRequiredDetails(t *testing
_, err := svc.Forward(context.Background(), c, account, body)
require.Error(t, err)
require.Equal(t, http.StatusBadGateway, rec.Code)
// missing_required_parameter 是确定性的请求错误:换账号、重试都不会变。按真实的
// 400 回写并保留 param/code,客户端才知道该补哪个字段(而不是收到可重试的 502)。
require.Equal(t, http.StatusBadRequest, rec.Code)
require.Equal(t, "invalid_request_error", gjson.Get(rec.Body.String(), "error.type").String())
require.Equal(t, "missing_required_parameter", gjson.Get(rec.Body.String(), "error.code").String())
require.Equal(t, "instructions", gjson.Get(rec.Body.String(), "error.param").String())
require.Contains(t, err.Error(), "upstream error: 400")
require.True(t, logSink.ContainsMessageAtLevel("OpenAI 上游返回 Instructions are required,已记录请求详情用于排查", "warn"))
@@ -488,6 +488,24 @@ func (s *OpenAIGatewayService) handleErrorResponse(
MarkResponseCommitted(c)
// 上游 400 是确定性的请求错误:同一份请求体换账号、重试多少次都会失败。归一成
// 502 upstream_error 会让下游网关把它当成可重试的上游故障反复重放(#5479 实测
// 30 个失败请求被放大成 60 次上游调用),同时抹掉客户端定位问题所需的 code/param。
//
// 走到这里说明 shouldFailoverOpenAIUpstreamResponse 已判定该 400 不可 failover,
// 即 server_is_overloaded / at capacity 这类可重试的 400 不会到达此处。
//
// 兄弟路径早已这么做:handleCompatErrorResponse(ChatCompletions / Anthropic)
// 回真实状态码 + invalid_request_error + 真实 message;/v1/images 还额外透传
// code/param。原生 Responses 是唯一漏掉的一条。
if isOpenAIDeterministicClientError(resp.StatusCode) {
writeOpenAIUpstreamClientError(c, resp.StatusCode, body, upstreamMsg)
if upstreamMsg == "" {
return nil, fmt.Errorf("upstream error: %d", resp.StatusCode)
}
return nil, fmt.Errorf("upstream error: %d message=%s", resp.StatusCode, upstreamMsg)
}
// Return appropriate error response
var errType, errMsg string
var statusCode int
@@ -0,0 +1,57 @@
package service
import (
"net/http"
"strings"
"github.com/gin-gonic/gin"
"github.com/tidwall/gjson"
)
// openAIUpstreamClientErrorFallbackType 是上游没给 error.type 时的兜底值。
// 与 handleCompatErrorResponse 对 400 的取值保持一致。
const openAIUpstreamClientErrorFallbackType = "invalid_request_error"
// openAIUpstreamClientErrorFallbackMessage 是上游连 message 都没给时的兜底文案。
// 仍然比 "Upstream request failed" 明确:它说明拒绝来自请求本身,而不是链路故障。
const openAIUpstreamClientErrorFallbackMessage = "Upstream rejected the request"
// isOpenAIDeterministicClientError 判断上游状态码是否表示「请求本身非法」。
//
// 只认 400:同一份请求体换任何账号、重试多少次都会得到同样的结果。
// - 401/402/403 是网关运营方的凭据/账单问题,继续包成 502,不向客户端暴露上游账号状态。
// - 404/405 既可能是模型不存在,也可能是上游 base_url 配错,同属运营方问题,保持现状。
// - 429 已有独立分支映射成 429。
// - 413 在更上面就按 request-body-too-large 走 failover 了。
func isOpenAIDeterministicClientError(statusCode int) bool {
return statusCode == http.StatusBadRequest
}
// writeOpenAIUpstreamClientError 以 OpenAI 错误体形状回写确定性客户端错误。
//
// 保留上游的 type/code/param:客户端靠 param 定位是哪个字段非法(上游会给出形如
// input[8].tools[1].tools[2].parameters 的路径),靠 code 判断是否值得重试。归一成
// {type:"upstream_error", message:"Upstream request failed"} 会把这些信息全部抹掉。
//
// upstreamMsg 由调用方传入,调用方已做过 sanitizeUpstreamErrorMessage 与
// redactAgentIdentitySensitiveBody;这里不重复清洗,也不回落读取原始 body 的
// message,避免绕开那两道脱敏。
func writeOpenAIUpstreamClientError(c *gin.Context, statusCode int, body []byte, upstreamMsg string) {
errorPayload := gin.H{"type": openAIUpstreamClientErrorFallbackType}
if errType := strings.TrimSpace(gjson.GetBytes(body, "error.type").String()); errType != "" {
errorPayload["type"] = errType
}
if code := strings.TrimSpace(extractUpstreamErrorCode(body)); code != "" {
errorPayload["code"] = code
}
if param := strings.TrimSpace(gjson.GetBytes(body, "error.param").String()); param != "" {
errorPayload["param"] = param
}
message := strings.TrimSpace(upstreamMsg)
if message == "" {
message = openAIUpstreamClientErrorFallbackMessage
}
errorPayload["message"] = message
c.JSON(statusCode, gin.H{"error": errorPayload})
}
@@ -0,0 +1,294 @@
package service
import (
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/model"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
// issue #5479 上游返回的原始错误体:Codex Desktop 的 automation_update 工具定义沉进
// 会话历史后,OpenAI 每一轮都确定性地拒收。
const openAIInvalidFunctionParametersBody = `{"error":{` +
`"message":"Invalid schema for function 'automation_update': schema must be a JSON Schema of 'type: \"object\"', got 'type: \"None\"'.",` +
`"type":"invalid_request_error",` +
`"param":"input[8].tools[1].tools[2].parameters",` +
`"code":"invalid_function_parameters"}}`
func newOpenAIUpstreamErrorTestContext(t *testing.T) (*gin.Context, *httptest.ResponseRecorder) {
t.Helper()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
return c, rec
}
func newOpenAIUpstreamErrorResponse(statusCode int, body string) *http.Response {
return &http.Response{
StatusCode: statusCode,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(body)),
}
}
func newOpenAIUpstreamErrorTestAccount() *Account {
return &Account{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Name: "acct"}
}
// 主复现:原生 Responses 路径必须回真实的 400 与上游诊断信息,而不是可重试的 502。
//
// 归一成 502 时下游网关(CCH 等)会把确定性的 Schema 错误当成临时上游故障重试,
// issue #5479 实测 30 个失败请求被放大成 60 次上游调用。
func TestHandleErrorResponse_Deterministic400IsNotRewrappedAs502(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
_, err := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, openAIInvalidFunctionParametersBody),
c, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, rec.Code, "确定性 400 不得被包成可重试的 502")
body := rec.Body.String()
require.Equal(t, "invalid_request_error", gjson.Get(body, "error.type").String())
require.Equal(t, "invalid_function_parameters", gjson.Get(body, "error.code").String())
require.Equal(t, "input[8].tools[1].tools[2].parameters", gjson.Get(body, "error.param").String(),
"param 是客户端定位哪个字段非法的唯一线索")
require.Contains(t, gjson.Get(body, "error.message").String(), "Invalid schema for function 'automation_update'")
require.NotContains(t, body, "Upstream request failed")
// 确定性请求错误不该换号重试——换任何账号都是同样的结果。
var failoverErr *UpstreamFailoverError
require.False(t, errors.As(err, &failoverErr), "400 不得触发 failover")
}
// 不变式:同一份上游错误体,原生 Responses 与 ChatCompletions/Anthropic 兼容路径
// 必须给出同样的状态码和同样的 message。这两条路径在同一个 service 上,之前一条对
// 一条错,正是本次修复的根因;锁死对称性避免将来只改一边。
func TestHandleErrorResponse_MatchesCompatSiblingForDeterministic400(t *testing.T) {
svc := &OpenAIGatewayService{cfg: &config.Config{}}
nativeCtx, nativeRec := newOpenAIUpstreamErrorTestContext(t)
_, nativeErr := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, openAIInvalidFunctionParametersBody),
nativeCtx, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, nativeErr)
compatCtx, _ := newOpenAIUpstreamErrorTestContext(t)
var compatStatus int
var compatType, compatMsg string
writeError := func(_ *gin.Context, statusCode int, errType, message string) {
compatStatus, compatType, compatMsg = statusCode, errType, message
}
_, compatErr := svc.handleCompatErrorResponse(
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, openAIInvalidFunctionParametersBody),
compatCtx, newOpenAIUpstreamErrorTestAccount(), writeError,
)
require.Error(t, compatErr)
require.Equal(t, compatStatus, nativeRec.Code, "两条路径的状态码必须一致")
require.Equal(t, compatType, gjson.Get(nativeRec.Body.String(), "error.type").String(),
"两条路径的 error.type 必须一致")
require.Equal(t, compatMsg, gjson.Get(nativeRec.Body.String(), "error.message").String(),
"两条路径的 message 必须一致")
}
// 上游只给 message、没有 type/code/param 时,仍要回 400 + 真实 message,
// 缺失字段用 OpenAI 惯例兜底,不得凭空编造 code/param。
func TestHandleErrorResponse_Deterministic400WithoutUpstreamMetadata(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
_, err := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, `{"error":{"message":"Invalid 'input': expected an array."}}`),
c, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, rec.Code)
body := rec.Body.String()
require.Equal(t, "invalid_request_error", gjson.Get(body, "error.type").String())
require.Equal(t, "Invalid 'input': expected an array.", gjson.Get(body, "error.message").String())
require.False(t, gjson.Get(body, "error.code").Exists(), "上游没给 code 就不要编一个")
require.False(t, gjson.Get(body, "error.param").Exists(), "上游没给 param 就不要编一个")
}
// 上游回非 JSON(反代的 HTML 错误页等)时不得 panic,也不得回空 message。
func TestHandleErrorResponse_Deterministic400WithNonJSONBody(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
_, err := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, `<html><body>400 Bad Request</body></html>`),
c, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, rec.Code)
body := rec.Body.String()
require.Equal(t, "invalid_request_error", gjson.Get(body, "error.type").String())
require.NotEmpty(t, gjson.Get(body, "error.message").String())
}
// 作用域守卫:本次只放行 400。其余落到 default 的状态码必须维持原样,
// 避免后续有人顺手把 404/422/5xx 一起改掉。
func TestHandleErrorResponse_NonDeterministicStatusesKeepGeneric502(t *testing.T) {
cases := []struct {
name string
statusCode int
body string
wantStatus int
wantType string
wantMsg string
}{
// 404/405 可能是上游 base_url 配错(运营方问题),不当成客户端错误暴露。
{"not_found", http.StatusNotFound, `{"error":{"message":"Unknown request URL"}}`,
http.StatusBadGateway, "upstream_error", "Upstream request failed"},
{"unprocessable", http.StatusUnprocessableEntity, `{"error":{"message":"Invalid schema for field messages"}}`,
http.StatusBadGateway, "upstream_error", "Upstream request failed"},
// 401/402/403 是网关运营方的凭据/账单问题,必须继续对客户端屏蔽上游账号状态。
{"unauthorized", http.StatusUnauthorized, `{"error":{"message":"Incorrect API key provided: sk-abc"}}`,
http.StatusBadGateway, "upstream_error", "Upstream authentication failed, please contact administrator"},
{"forbidden", http.StatusForbidden, `{"error":{"message":"Your account is deactivated"}}`,
http.StatusBadGateway, "upstream_error", "Upstream access forbidden, please contact administrator"},
// 429 保持独立映射。
{"rate_limited", http.StatusTooManyRequests, `{"error":{"message":"Rate limit reached"}}`,
http.StatusTooManyRequests, "rate_limit_error", "Upstream rate limit exceeded, please retry later"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
_, err := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(tc.statusCode, tc.body),
c, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, err)
require.Equal(t, tc.wantStatus, rec.Code)
require.Equal(t, tc.wantType, gjson.Get(rec.Body.String(), "error.type").String())
require.Equal(t, tc.wantMsg, gjson.Get(rec.Body.String(), "error.message").String())
})
}
}
// 顺序守卫:管理员配置的错误透传规则在更上游命中,新分支不得抢在它前面。
func TestHandleErrorResponse_PassthroughRuleStillWinsOver400Branch(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
ruleSvc := &ErrorPassthroughService{}
ruleSvc.setLocalCache([]*model.ErrorPassthroughRule{
newNonFailoverPassthroughRule(http.StatusBadRequest, "automation_update", http.StatusTeapot, "自定义文案"),
})
BindErrorPassthroughService(c, ruleSvc)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
_, err := svc.handleErrorResponse(
context.Background(),
newOpenAIUpstreamErrorResponse(http.StatusBadRequest, openAIInvalidFunctionParametersBody),
c, newOpenAIUpstreamErrorTestAccount(), nil,
)
require.Error(t, err)
require.Equal(t, http.StatusTeapot, rec.Code, "命中透传规则时必须按规则的状态码回写")
require.Equal(t, "自定义文案", gjson.Get(rec.Body.String(), "error.message").String())
}
func TestIsOpenAIDeterministicClientError(t *testing.T) {
require.True(t, isOpenAIDeterministicClientError(http.StatusBadRequest))
for _, status := range []int{
http.StatusUnauthorized, http.StatusPaymentRequired, http.StatusForbidden,
http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusRequestEntityTooLarge,
http.StatusUnprocessableEntity, http.StatusTooManyRequests,
http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable,
} {
require.False(t, isOpenAIDeterministicClientError(status), "status %d", status)
}
}
func TestWriteOpenAIUpstreamClientError_PayloadShape(t *testing.T) {
cases := []struct {
name string
body string
upstreamMsg string
wantType string
wantCode string
wantParam string
wantMessage string
}{
{
name: "full_metadata",
body: openAIInvalidFunctionParametersBody,
upstreamMsg: "Invalid schema for function 'automation_update'",
wantType: "invalid_request_error",
wantCode: "invalid_function_parameters",
wantParam: "input[8].tools[1].tools[2].parameters",
wantMessage: "Invalid schema for function 'automation_update'",
},
{
name: "upstream_type_preserved",
body: `{"error":{"type":"invalid_prompt","message":"blocked"}}`,
upstreamMsg: "blocked",
wantType: "invalid_prompt",
wantMessage: "blocked",
},
{
name: "empty_body_falls_back",
body: ``,
upstreamMsg: "",
wantType: "invalid_request_error",
wantMessage: openAIUpstreamClientErrorFallbackMessage,
},
{
// 调用方传入的 message 已脱敏,必须原样使用,不得回落读取原始 body。
name: "sanitized_message_wins_over_raw_body",
body: `{"error":{"message":"failed for key=secret123"}}`,
upstreamMsg: "failed for key=***",
wantType: "invalid_request_error",
wantMessage: "failed for key=***",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
c, rec := newOpenAIUpstreamErrorTestContext(t)
writeOpenAIUpstreamClientError(c, http.StatusBadRequest, []byte(tc.body), tc.upstreamMsg)
require.Equal(t, http.StatusBadRequest, rec.Code)
body := rec.Body.String()
require.Equal(t, tc.wantType, gjson.Get(body, "error.type").String())
require.Equal(t, tc.wantMessage, gjson.Get(body, "error.message").String())
if tc.wantCode == "" {
require.False(t, gjson.Get(body, "error.code").Exists())
} else {
require.Equal(t, tc.wantCode, gjson.Get(body, "error.code").String())
}
if tc.wantParam == "" {
require.False(t, gjson.Get(body, "error.param").Exists())
} else {
require.Equal(t, tc.wantParam, gjson.Get(body, "error.param").String())
}
require.NotContains(t, body, "secret123", "原始 body 里的敏感串不得泄漏")
})
}
}