Merge pull request #7810 from wucm667/fix/issue-6247-responses-chat-fallback

fix(gateway): retry rejected structured input through chat endpoint
This commit is contained in:
Wesley Liddick
2026-10-13 15:01:20 +08:00
committed by GitHub
7 changed files with 771 additions and 9 deletions
@@ -67,13 +67,26 @@ func (s *OpenAIGatewayService) newStreamHeaderWriter(c *gin.Context, upstream ht
// (下游 handleXxxErrorResponse 需要再次读取),返回原始错误体与脱敏后的
// 上游错误消息。
func (s *OpenAIGatewayService) readOpenAIUpstreamError(resp *http.Response) ([]byte, string) {
respBody := s.readUpstreamErrorBody(resp)
respBody, upstreamMsg, _ := s.readOpenAIUpstreamErrorWithCompleteness(resp)
return respBody, upstreamMsg
}
func (s *OpenAIGatewayService) readOpenAIUpstreamErrorWithCompleteness(resp *http.Response) ([]byte, string, bool) {
limit := openAIUpstreamErrorBodyReadLimitForConfig(nil)
if s != nil {
limit = openAIUpstreamErrorBodyReadLimitForConfig(s.cfg)
}
respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, limit+1))
complete := readErr == nil && int64(len(respBody)) <= limit
if int64(len(respBody)) > limit {
respBody = respBody[:limit]
}
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
return respBody, upstreamMsg
return respBody, upstreamMsg, complete
}
// failoverOpenAIUpstreamHTTPError 对 >=400 的上游响应做 failover 判定:命中时
@@ -7,6 +7,8 @@ import (
"errors"
"fmt"
"net/http"
"regexp"
"strconv"
"strings"
"sync/atomic"
"time"
@@ -370,7 +372,7 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
// 8. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, upstreamMsg := s.readOpenAIUpstreamError(resp)
respBody, upstreamMsg, completeErrorBody := s.readOpenAIUpstreamErrorWithCompleteness(resp)
if !agentIdentityTaskRecoveryWasTried(ctx) && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, respBody) {
expectedTaskID := account.GetCredential("task_id")
if err := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); err != nil {
@@ -387,7 +389,24 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
zap.Int("upstream_status", resp.StatusCode),
zap.String("upstream_message", upstreamMsg),
)
return s.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel)
return s.forwardAsRawChatCompletionsWithStartTime(ctx, c, account, body, defaultMappedModel, startTime)
}
responsesMode, _ := account.Extra[openai_compat.ExtraKeyResponsesMode].(string)
if account.IsOpenAIApiKey() &&
openai_compat.NormalizeResponsesSupportMode(responsesMode) == openai_compat.ResponsesSupportModeAuto &&
openai_compat.ResolveResponsesSupport(account.Extra) == openai_compat.ResponsesSupportYes &&
!isResponsesShape &&
(ctx == nil || ctx.Err() == nil) &&
(c.Request == nil || c.Request.Context().Err() == nil) &&
!IsResponseCommitted(c) &&
!c.Writer.Written() &&
completeErrorBody &&
isConvertedResponsesInputStringRejection(resp.StatusCode, respBody) {
logger.L().Info("openai chat_completions: structured Responses input rejected, falling back to raw chat completions",
zap.Int64("account_id", account.ID),
zap.Int("upstream_status", resp.StatusCode),
)
return s.forwardAsRawChatCompletionsWithStartTime(ctx, c, account, body, defaultMappedModel, startTime)
}
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
return nil, foErr
@@ -441,6 +460,155 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
return result, handleErr
}
var responsesInputStringValidationError = regexp.MustCompile(`^['"]loc['"]\s*:\s*\(\s*['"]body['"]\s*,\s*['"]input['"]\s*,\s*['"]str['"]\s*\)$`)
var responsesStringValidationMessage = regexp.MustCompile(`^['"]msg['"]\s*:\s*['"]Input should be a valid string['"]$`)
var responsesStringValidationType = regexp.MustCompile(`^['"]type['"]\s*:\s*['"]string_type['"]$`)
var responsesValidationErrors = regexp.MustCompile(`(?s)^([1-9][0-9]*) validation errors?(?: for [^\r\n:]+)?:\s*(\[.*\])$`)
var responsesInvalidInputTypeMessage = regexp.MustCompile(`(?i)^(?:Invalid type for ['"]input['"]:\s*)?Expected a string(?: for input)?, but got an (?:array|object) instead\.?$`)
func isResponsesInputStringValidationMessage(message string) bool {
match := responsesValidationErrors.FindStringSubmatch(strings.TrimSpace(message))
if match == nil {
return false
}
entries, ok := splitResponsesValidationFields(match[2][1 : len(match[2])-1])
count, err := strconv.Atoi(match[1])
if !ok || err != nil || count != len(entries) {
return false
}
matched := false
for _, entry := range entries {
if !strings.HasPrefix(entry, "{") || !strings.HasSuffix(entry, "}") {
return false
}
fields, ok := splitResponsesValidationFields(entry[1 : len(entry)-1])
if !ok {
return false
}
location, message, stringType := false, false, false
keys := make(map[string]bool, len(fields))
for _, field := range fields {
key, value, found := strings.Cut(field, ":")
key = strings.TrimSpace(key)
if !found || len(key) < 3 || (key[0] != '\'' && key[0] != '"') || key[len(key)-1] != key[0] || strings.TrimSpace(value) == "" || keys[key[1:len(key)-1]] {
return false
}
keys[key[1:len(key)-1]] = true
location = location || responsesInputStringValidationError.MatchString(field)
message = message || responsesStringValidationMessage.MatchString(field)
stringType = stringType || responsesStringValidationType.MatchString(field)
}
matched = matched || (location && message && stringType)
}
return matched
}
func splitResponsesValidationFields(value string) ([]string, bool) {
var fields []string
var brackets []byte
var quote byte
escaped := false
start := 0
for index := 0; index < len(value); index++ {
character := value[index]
if quote != 0 {
if escaped {
escaped = false
} else if character == '\\' {
escaped = true
} else if character == quote {
quote = 0
}
continue
}
switch character {
case '\'', '"':
quote = character
case '{', '[', '(':
brackets = append(brackets, character)
case '}', ']', ')':
if len(brackets) == 0 {
return nil, false
}
opening := brackets[len(brackets)-1]
if (character == '}' && opening != '{') || (character == ']' && opening != '[') || (character == ')' && opening != '(') {
return nil, false
}
brackets = brackets[:len(brackets)-1]
case ',':
if len(brackets) == 0 {
field := strings.TrimSpace(value[start:index])
if field == "" {
return nil, false
}
fields = append(fields, field)
start = index + 1
}
}
}
if quote != 0 || len(brackets) != 0 {
return nil, false
}
if field := strings.TrimSpace(value[start:]); field != "" {
fields = append(fields, field)
}
return fields, len(fields) > 0
}
func isConvertedResponsesInputStringRejection(status int, responseBody []byte) bool {
if status != http.StatusBadRequest || !json.Valid(responseBody) {
return false
}
envelope := gjson.ParseBytes(responseBody)
for _, object := range []gjson.Result{envelope, envelope.Get("error")} {
if !object.IsObject() {
return false
}
keys := make(map[string]bool)
unambiguous := true
object.ForEach(func(key, value gjson.Result) bool {
name := key.String()
if keys[name] {
unambiguous = false
return false
}
keys[name] = true
switch name {
case "usage", "output", "output_text", "choices", "tool_calls", "response", "status", "incomplete_details":
unambiguous = false
case "code":
code := strings.ToLower(strings.TrimSpace(value.String()))
unambiguous = code == "" || code == "invalid_type" || code == "invalid_request_error"
case "param":
param := strings.ToLower(strings.TrimSpace(value.String()))
unambiguous = param == "" || param == "input"
case "type":
errorType := strings.ToLower(strings.TrimSpace(value.String()))
unambiguous = errorType == "" || errorType == "invalid_type" || errorType == "invalid_request_error"
}
return unambiguous
})
if !unambiguous {
return false
}
}
message := gjson.GetBytes(responseBody, "error.message").String()
if isResponsesInputStringValidationMessage(message) {
return true
}
code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.code").String()))
typ := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.type").String()))
if code != "invalid_type" && typ != "invalid_type" {
return false
}
param := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.param").String()))
if param != "input" {
return false
}
message = strings.ToLower(strings.TrimSpace(message))
return responsesInvalidInputTypeMessage.MatchString(message)
}
func normalizeResponsesRequestServiceTier(req *apicompat.ResponsesRequest) {
if req == nil {
return
@@ -61,7 +61,17 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions(
body []byte,
defaultMappedModel string,
) (*OpenAIForwardResult, error) {
startTime := time.Now()
return s.forwardAsRawChatCompletionsWithStartTime(ctx, c, account, body, defaultMappedModel, time.Now())
}
func (s *OpenAIGatewayService) forwardAsRawChatCompletionsWithStartTime(
ctx context.Context,
c *gin.Context,
account *Account,
body []byte,
defaultMappedModel string,
startTime time.Time,
) (*OpenAIForwardResult, error) {
// 1. Parse minimal fields needed for routing/billing
originalModel := gjson.GetBytes(body, "model").String()
@@ -3,7 +3,9 @@ package service
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
@@ -13,6 +15,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
"github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
@@ -32,6 +35,546 @@ func (w *openAIChatFailingWriter) Write(p []byte) (int, error) {
return w.ResponseWriter.Write(p)
}
func TestForwardAsChatCompletions_ResponsesSupportedFallsBackWhenStructuredInputRequiresString(t *testing.T) {
for name, rejection := range map[string]string{
"OpenAI invalid type": `{"error":{"type":"invalid_request_error","code":"invalid_type","param":"input","message":"Invalid type for 'input': expected a string, but got an array instead."}}`,
"vLLM validation": `{"error":{"message":"1 validation error: [{'type':'string_type', 'loc':('body','input','str'), 'msg':'Input should be a valid string', 'input':[{'role':'user','content':'hi'}, {'role':'assistant','content':[{'type':'output_text'}]}, ResponseFunctionToolCall(type='function_call'), {'type':'function_call_output'}]}]"}}`,
} {
t.Run(name, func(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"},{"role":"assistant","content":[{"type":"text","text":"checking"},{"type":"thinking","thinking":"think"}],"tool_calls":[{"id":"call_1","type":"function","function":{"name":"lookup","arguments":"{}"}}]},{"role":"tool","tool_call_id":"call_1","content":"done"},{"role":"user","content":"what now"}],"stream":false}`)
originalBody := append([]byte(nil), body...)
logs, releaseLogs := captureStructuredLog(t)
defer releaseLogs()
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{
StatusCode: http.StatusBadRequest,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(
rejection,
)),
},
{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_raw_fallback"}},
Body: io.NopCloser(strings.NewReader(
`{"id":"chatcmpl_fallback","object":"chat.completion","model":"gpt-5.4","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`,
)),
},
}}
svc := &OpenAIGatewayService{
cfg: structuredInputRetryTestConfig(),
httpUpstream: upstream,
}
account := structuredInputRetryTestAccount()
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
require.NoError(t, err)
require.NotNil(t, result)
require.Len(t, upstream.requests, 2)
require.Equal(t, "/v1/responses", upstream.requests[0].URL.Path)
require.True(t, gjson.GetBytes(upstream.bodies[0], "input").IsArray())
require.Equal(t, "/v1/chat/completions", upstream.requests[1].URL.Path)
require.JSONEq(t, string(body), string(upstream.bodies[1]))
require.Equal(t, originalBody, body)
require.Equal(t, "/v1/chat/completions", result.UpstreamEndpoint)
require.Equal(t, "/v1/chat/completions", GetActualOpenAIUpstreamEndpoint(c))
require.Equal(t, "ok", gjson.Get(rec.Body.String(), "choices.0.message.content").String())
require.True(t, logs.ContainsMessage("structured Responses input rejected"))
logs.mu.Lock()
defer logs.mu.Unlock()
for _, event := range logs.events {
require.NotContains(t, event.Fields, "upstream_message")
require.NotContains(t, event.Fields, "upstream_body")
require.NotContains(t, event.Message, rejection)
}
})
}
}
type openAIChatFallbackBoundaryReader struct {
io.Reader
beforeRead func()
}
func (reader *openAIChatFallbackBoundaryReader) Read(buffer []byte) (int, error) {
if reader.beforeRead != nil {
reader.beforeRead()
reader.beforeRead = nil
}
return reader.Reader.Read(buffer)
}
func TestForwardAsChatCompletions_StructuredInputRetrySafetyBoundaries(t *testing.T) {
rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`
validation := `"message":"1 validation error: [{'type':'string_type','loc':('body','input','str'),'msg':'Input should be a valid string'}]"`
for _, test := range []struct {
name string
body string
}{
{name: "duplicate error", body: strings.TrimSuffix(rejection, "}") + `,"error":{"code":"rate_limit_exceeded"}}`},
{name: "duplicate code", body: strings.Replace(rejection, `"code":"invalid_type"`, `"code":"invalid_type","code":"rate_limit_exceeded"`, 1)},
{name: "duplicate escaped param", body: strings.Replace(rejection, `"param":"input"`, `"param":"input","p\u0061ram":"tools"`, 1)},
{name: "duplicate message", body: strings.Replace(rejection, `"message":"Expected a string, but got an array instead."`, `"message":"Expected a string, but got an array instead.","message":"Input is too long."`, 1)},
{name: "contradictory error type", body: strings.Replace(rejection, `"error":{`, `"error":{"type":"authentication_error",`, 1)},
{name: "contradictory envelope type", body: strings.TrimSuffix(rejection, "}") + `,"type":"rate_limit_error"}`},
{name: "validation contradictory type", body: `{"error":{"type":"rate_limit_error",` + validation + `}}`},
{name: "validation contradictory envelope type", body: `{"type":"authentication_error","error":{` + validation + `}}`},
{name: "validation contradictory code", body: `{"error":{"code":"rate_limit_exceeded",` + validation + `}}`},
{name: "validation contradictory param", body: `{"error":{"param":"tools",` + validation + `}}`},
{name: "contradictory envelope code", body: strings.TrimSuffix(rejection, "}") + `,"code":"rate_limit_exceeded"}`},
{name: "contradictory envelope param", body: strings.TrimSuffix(rejection, "}") + `,"param":"tools"}`},
{name: "type cannot override code", body: strings.Replace(rejection, `"code":"invalid_type"`, `"code":"rate_limit_exceeded","type":"invalid_type"`, 1)},
{name: "usage", body: strings.TrimSuffix(rejection, "}") + `,"usage":{"input_tokens":12,"output_tokens":3}}`},
{name: "error usage", body: strings.Replace(rejection, `"error":{`, `"error":{"usage":{"input_tokens":12},`, 1)},
{name: "output", body: strings.TrimSuffix(rejection, "}") + `,"output":[{"type":"message","content":[{"type":"output_text","text":"partial"}]}]}`},
{name: "output text", body: strings.TrimSuffix(rejection, "}") + `,"output_text":"partial"}`},
{name: "tool calls", body: strings.TrimSuffix(rejection, "}") + `,"tool_calls":[{"id":"call_partial","type":"function","function":{"name":"lookup","arguments":"{}"}}]}`},
{name: "choices", body: strings.TrimSuffix(rejection, "}") + `,"choices":[{"delta":{"content":"partial"}}]}`},
{name: "nested response", body: strings.TrimSuffix(rejection, "}") + `,"response":{"status":"incomplete","usage":{"input_tokens":12},"output":[]}}`},
{name: "partial status", body: strings.TrimSuffix(rejection, "}") + `,"status":"in_progress"}`},
{name: "incomplete details", body: strings.TrimSuffix(rejection, "}") + `,"incomplete_details":{"reason":"max_output_tokens"}}`},
{name: "partial event", body: strings.TrimSuffix(rejection, "}") + `,"type":"response.output_text.delta","delta":"partial"}`},
{name: "caller canceled"},
{name: "request canceled"},
{name: "response committed"},
{name: "writer committed"},
} {
t.Run(test.name, func(t *testing.T) {
body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
requestCtx, cancelRequest := context.WithCancel(context.Background())
defer cancelRequest()
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)).WithContext(requestCtx)
errorBody := test.body
if errorBody == "" {
errorBody = rejection
}
reader := &openAIChatFallbackBoundaryReader{Reader: strings.NewReader(errorBody), beforeRead: func() {
switch test.name {
case "caller canceled":
cancel()
case "request canceled":
cancelRequest()
case "response committed":
MarkResponseCommitted(c)
case "writer committed":
c.Writer.WriteHeaderNow()
}
}}
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusBadRequest,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(reader),
}}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
result, err := svc.ForwardAsChatCompletions(ctx, c, structuredInputRetryTestAccount(), body, "", "")
require.Error(t, err)
require.Nil(t, result)
require.Len(t, upstream.requests, 1)
require.Equal(t, "/v1/responses", upstream.requests[0].URL.Path)
})
}
}
func TestConvertedResponsesInputStringRejectionIsNarrow(t *testing.T) {
matching := []byte(`{"error":{"code":"invalid_type","param":"input","message":"Expected a string for input, but got an object instead."}}`)
require.True(t, isConvertedResponsesInputStringRejection(http.StatusBadRequest, matching))
require.True(t, isConvertedResponsesInputStringRejection(http.StatusBadRequest, []byte(`{"error":{"type":"invalid_type","param":"input","message":"Invalid type for 'input': expected a string, but got an array instead."}}`)))
for _, tc := range []struct {
name string
status int
body string
}{
{name: "nested content validation", status: http.StatusBadRequest, body: `{"error":{"message":"'loc':('body','input',0,'content','str'), 'msg':'Input should be a valid string'"}}`},
{name: "other field validation", status: http.StatusBadRequest, body: `{"error":{"message":"'loc':('body','model','str'), 'msg':'Input should be a valid string'"}}`},
{name: "generic string validation", status: http.StatusBadRequest, body: `{"error":{"message":"Input should be a valid string"}}`},
{name: "wrong status", status: http.StatusUnprocessableEntity, body: string(matching)},
{name: "authentication", status: http.StatusUnauthorized, body: string(matching)},
{name: "authorization", status: http.StatusForbidden, body: string(matching)},
{name: "rate limit", status: http.StatusTooManyRequests, body: string(matching)},
{name: "server error", status: http.StatusInternalServerError, body: string(matching)},
{name: "wrong parameter", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_type","param":"tools","message":"Expected a string, but got an array instead."}}`},
{name: "wrong code", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_request_error","param":"input","message":"Expected a string, but got an array instead."}}`},
{name: "unrelated input error", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_type","param":"input","message":"Input is too long."}}`},
} {
t.Run(tc.name, func(t *testing.T) {
require.False(t, isConvertedResponsesInputStringRejection(tc.status, []byte(tc.body)))
})
}
}
func TestConvertedResponsesInputStringRejectionValidationBoundaries(t *testing.T) {
validation := `{'type': 'string_type', 'loc': ('body', 'input', 'str'), 'msg': 'Input should be a valid string', 'input': [{'role': 'user', 'content': 'hello'}]}`
for _, test := range []struct {
name string
message string
want bool
}{
{name: "complete Pydantic error", message: "1 validation error: [" + validation + "]", want: true},
{name: "multiple Pydantic errors", message: "2 validation errors for Request: [" + strings.Replace(validation, "'input', 'str'", "'model', 'str'", 1) + ", " + validation + "]", want: true},
{name: "double quoted keys", message: "1 validation error: [" + strings.ReplaceAll(validation, "'", `"`) + "]", want: true},
{name: "surrounding whitespace", message: " \n1 validation error: [" + validation + "]\n ", want: true},
{name: "escaped quote in history", message: `1 validation error: [{'type':'string_type','loc':('body','input','str'),'msg':'Input should be a valid string','input':[{'content':'can\'t [close] this }'}]}]`, want: true},
{name: "bare loc and msg", message: `'loc':('body','input','str'), 'msg':'Input should be a valid string'`},
{name: "echoed dictionary in unrelated error", message: "Invalid history: " + validation},
{name: "echoed fake loc and msg in input", message: `1 validation error: [{'type':'string_type','loc':('body','model','str'),'msg':'Input should be a valid string','input':{'loc':('body','input','str'),'msg':'Input should be a valid string'}}]`},
{name: "echoed fake loc and msg in quoted history", message: `1 validation error: [{'type':'string_type','loc':('body','input',0,'content','str'),'msg':'Input should be a valid string','input':"'loc':('body','input','str'), 'msg':'Input should be a valid string'"}]`},
{name: "echoed complete error in quoted history", message: `1 validation error: [{'type':'string_type','loc':('body','model','str'),'msg':'Input should be a valid string','input':"1 validation error: [{'type':'string_type','loc':('body','input','str'),'msg':'Input should be a valid string'}]"}]`},
{name: "truncated error list", message: "1 validation error: [" + validation},
{name: "truncated quoted input", message: `1 validation error: [{'type':'string_type','loc':('body','input','str'),'msg':'Input should be a valid string','input':'history}]`},
{name: "mismatched closing bracket", message: "1 validation error: [" + strings.TrimSuffix(validation, "}") + ")]"},
{name: "elided validation list", message: "240 validation errors: ... " + validation},
{name: "incorrect error count", message: "2 validation errors: [" + validation + "]"},
{name: "overflowed error count", message: "999999999999999999999999 validation errors: [" + validation + "]"},
{name: "duplicate location", message: "1 validation error: [" + strings.Replace(validation, "'input':", "'loc': ('body', 'model', 'str'), 'input':", 1) + "]"},
{name: "wrong validation type", message: "1 validation error: [" + strings.Replace(validation, "string_type", "value_error", 1) + "]"},
{name: "missing validation type", message: "1 validation error: [" + strings.Replace(validation, "'type': 'string_type', ", "", 1) + "]"},
{name: "location and message in different errors", message: "2 validation errors: [" + strings.Replace(validation, "Input should be a valid string", "Invalid history", 1) + ", " + strings.Replace(validation, "'input', 'str'", "'model', 'str'", 1) + "]"},
{name: "trailing echoed text", message: "1 validation error: [" + validation + "] echoed history"},
{name: "structured string array", message: "Expected a string, but got an array instead.", want: true},
{name: "structured object", message: "Invalid type for 'input': expected a string, but got an object instead.", want: true},
{name: "structured echoed history", message: "Invalid history containing 'expected a string' and 'got an array'"},
{name: "structured truncated message", message: "Expected a string, but got an array"},
} {
t.Run(test.name, func(t *testing.T) {
body, err := json.Marshal(map[string]any{"error": map[string]string{"message": test.message, "code": "invalid_type", "param": "input"}})
require.NoError(t, err)
require.Equal(t, test.want, isConvertedResponsesInputStringRejection(http.StatusBadRequest, body))
if test.want {
require.False(t, isConvertedResponsesInputStringRejection(http.StatusBadRequest, body[:len(body)-1]))
require.False(t, isConvertedResponsesInputStringRejection(http.StatusBadRequest, append(body, []byte(" trailing")...)))
}
})
}
}
func TestForwardAsChatCompletions_StructuredInputRetryRequiresCompleteErrorBody(t *testing.T) {
rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`
limit := int(openAIUpstreamErrorBodyReadLimitForConfig(structuredInputRetryTestConfig()))
for _, test := range []struct {
name string
payload string
readErr error
unknown bool
want bool
}{
{name: "complete JSON followed by read failure", payload: rejection, readErr: io.ErrUnexpectedEOF},
{name: "truncated JSON", payload: rejection[:len(rejection)-1]},
{name: "exact read limit", payload: rejection + strings.Repeat(" ", limit-len(rejection)), want: true},
{name: "one byte beyond read limit", payload: rejection + strings.Repeat(" ", limit-len(rejection)+1)},
{name: "valid JSON prefix before discarded suffix", payload: rejection + strings.Repeat(" ", limit-len(rejection)) + "trailing"},
{name: "unknown capability still retries after read failure", payload: rejection, readErr: io.ErrUnexpectedEOF, unknown: true, want: true},
{name: "unknown capability still retries beyond read limit", payload: rejection + strings.Repeat(" ", limit-len(rejection)+1), unknown: true, want: true},
} {
t.Run(test.name, func(t *testing.T) {
body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}]}`)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
errorBody := io.NopCloser(strings.NewReader(test.payload))
if test.readErr != nil {
errorBody = &openAIChatStreamReadErrorCloser{payload: []byte(test.payload), err: test.readErr}
}
account := structuredInputRetryTestAccount()
status := http.StatusBadRequest
if test.unknown {
delete(account.Extra, openai_compat.ExtraKeyResponsesSupported)
status = http.StatusNotFound
}
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{StatusCode: status, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: errorBody},
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"id":"chatcmpl_retry","choices":[{"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`))},
}}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
if test.want {
require.NoError(t, err)
require.NotNil(t, result)
require.Len(t, upstream.requests, 2)
} else {
require.Error(t, err)
require.Nil(t, result)
require.Len(t, upstream.requests, 1)
}
})
}
}
type openAIChatFallbackDelayReader struct {
io.Reader
elapsed time.Duration
}
func (reader *openAIChatFallbackDelayReader) Read(buffer []byte) (int, error) {
if reader.elapsed == 0 {
started := time.Now()
time.Sleep(30 * time.Millisecond)
reader.elapsed = time.Since(started)
}
return reader.Reader.Read(buffer)
}
func TestForwardAsChatCompletions_RawFallbackPreservesResponsesTiming(t *testing.T) {
for _, unknown := range []bool{false, true} {
for _, stream := range []bool{false, true} {
t.Run(fmt.Sprintf("unknown=%t/stream=%t", unknown, stream), func(t *testing.T) {
body := []byte(fmt.Sprintf(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":%t}`, stream))
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
account := structuredInputRetryTestAccount()
status := http.StatusBadRequest
if unknown {
delete(account.Extra, openai_compat.ExtraKeyResponsesSupported)
status = http.StatusNotFound
}
initialBody := &openAIChatFallbackDelayReader{Reader: strings.NewReader(`{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`)}
responseBody := `{"id":"chatcmpl_retry","choices":[{"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`
contentType := "application/json"
if stream {
responseBody = "data: {\"id\":\"chatcmpl_retry\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: {\"choices\":[],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1,\"total_tokens\":2}}\n\ndata: [DONE]\n\n"
contentType = "text/event-stream"
}
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{StatusCode: status, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(initialBody)},
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{contentType}}, Body: io.NopCloser(strings.NewReader(responseBody))},
}}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
require.NoError(t, err)
require.NotNil(t, result)
require.Len(t, upstream.requests, 2)
require.Equal(t, "/v1/responses", upstream.requests[0].URL.Path)
require.Equal(t, "/v1/chat/completions", upstream.requests[1].URL.Path)
require.GreaterOrEqual(t, result.Duration, initialBody.elapsed)
if stream {
require.NotNil(t, result.FirstTokenMs)
require.GreaterOrEqual(t, *result.FirstTokenMs, int(initialBody.elapsed.Milliseconds()))
}
})
}
}
}
func TestForwardAsChatCompletions_StructuredInputRejectionDoesNotRetryIneligibleRequests(t *testing.T) {
rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`
for _, test := range []struct {
name string
mode string
unknown bool
platform string
protocol string
endpoint string
accountType string
responsesBody bool
nativeIngress bool
status int
rejection string
}{
{name: "forced Responses", mode: "force_responses"},
{name: "forced Responses without probe", mode: "force_responses", unknown: true},
{name: "unknown support", unknown: true},
{name: "Responses shaped Chat ingress", responsesBody: true},
{name: "native Responses ingress", responsesBody: true, nativeIngress: true},
{name: "native CN Responses protocol", platform: PlatformDeepseek},
{name: "adaptive Chat ingress stays raw", platform: PlatformKimi, protocol: APIProtocolAdaptive, endpoint: "/chat/completions"},
{name: "adaptive Responses shaped ingress stays Responses", platform: PlatformKimi, protocol: APIProtocolAdaptive, responsesBody: true},
{name: "adaptive Responses shaped ingress converts to Chat", platform: PlatformZhipu, protocol: APIProtocolAdaptive, responsesBody: true, endpoint: "/chat/completions"},
{name: "model routed Responses is not OpenAI retry eligible", platform: PlatformCommandCode, protocol: APIProtocolResponses},
{name: "OAuth", accountType: AccountTypeOAuth},
{name: "authentication", status: http.StatusUnauthorized},
{name: "authorization", status: http.StatusForbidden},
{name: "rate limit", status: http.StatusTooManyRequests},
{name: "unrelated validation", rejection: `{"error":{"code":"invalid_type","param":"tools","message":"Expected a string, but got an array instead."}}`},
{name: "nested validation", rejection: `{"error":{"message":"'loc':('body','input',0,'content','str'), 'msg':'Input should be a valid string'"}}`},
{name: "echoed validation", rejection: `{"error":{"message":"1 validation error: [{'type':'string_type','loc':('body','model','str'),'msg':'Input should be a valid string','input':{'loc':('body','input','str'),'msg':'Input should be a valid string'}}]"}}`},
{name: "echoed structured error", rejection: `{"error":{"code":"invalid_type","param":"input","message":"Invalid history containing expected a string but got an array"}}`},
} {
t.Run(test.name, func(t *testing.T) {
body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`)
if test.responsesBody {
body = []byte(`{"model":"gpt-5.4","input":[{"role":"user","content":"hello"}],"stream":false}`)
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
account := structuredInputRetryTestAccount()
account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true, openai_compat.ExtraKeyResponsesMode: test.mode}
if test.unknown {
delete(account.Extra, openai_compat.ExtraKeyResponsesSupported)
}
if test.platform != "" {
account.Platform = test.platform
account.Credentials["api_protocol"] = APIProtocolResponses
}
if test.protocol != "" {
account.Credentials["api_protocol"] = test.protocol
}
if test.accountType != "" {
account.Type = test.accountType
account.Credentials["access_token"] = "test-token"
}
status := test.status
if status == 0 {
status = http.StatusBadRequest
}
errorBody := rejection
if test.rejection != "" {
errorBody = test.rejection
}
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: status,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(errorBody)),
}}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
var result *OpenAIForwardResult
var err error
if test.nativeIngress {
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body))
result, err = svc.Forward(context.Background(), c, account, body)
} else {
result, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
}
require.Error(t, err)
require.Nil(t, result)
require.Len(t, upstream.requests, 1)
endpoint := test.endpoint
if endpoint == "" {
endpoint = "/responses"
}
require.True(t, strings.HasSuffix(upstream.requests[0].URL.Path, endpoint))
var failover *UpstreamFailoverError
if status == http.StatusUnauthorized || status == http.StatusForbidden || status == http.StatusTooManyRequests {
require.ErrorAs(t, err, &failover)
require.Equal(t, status, failover.StatusCode)
} else {
require.False(t, errors.As(err, &failover))
require.True(t, c.Writer.Written())
}
})
}
}
func TestForwardAsChatCompletions_StructuredInputRetryPreservesStreamingAndMapping(t *testing.T) {
for _, useAccountMapping := range []bool{false, true} {
t.Run(fmt.Sprintf("account_mapping=%t", useAccountMapping), func(t *testing.T) {
body := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hello"}],"stream":true,"stream_options":{"include_usage":false},"vendor_option":"preserve"}`)
originalBody := append([]byte(nil), body...)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
account := structuredInputRetryTestAccount()
account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true, openai_compat.ExtraKeyResponsesMode: "auto"}
mappedModel := "gpt-5.4"
if useAccountMapping {
account.Credentials["model_mapping"] = map[string]any{"client-model": mappedModel}
}
stream := "data: {\"id\":\"chatcmpl_retry\",\"object\":\"chat.completion.chunk\",\"model\":\"gpt-5.4\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: {\"choices\":[],\"usage\":{\"prompt_tokens\":9,\"completion_tokens\":4,\"total_tokens\":13}}\n\ndata: [DONE]\n\n"
upstream := &httpUpstreamRecorder{responses: []*http.Response{
{StatusCode: http.StatusBadRequest, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`))},
{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(stream))},
}}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
defaultMappedModel := mappedModel
if useAccountMapping {
defaultMappedModel = ""
}
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", defaultMappedModel)
require.NoError(t, err)
require.Len(t, upstream.requests, 2)
for _, outbound := range upstream.bodies {
require.Equal(t, mappedModel, gjson.GetBytes(outbound, "model").String())
require.True(t, gjson.GetBytes(outbound, "stream").Bool())
}
require.JSONEq(t, gjson.GetBytes(body, "messages").Raw, gjson.GetBytes(upstream.bodies[1], "messages").Raw)
require.Equal(t, "preserve", gjson.GetBytes(upstream.bodies[1], "vendor_option").String())
require.True(t, gjson.GetBytes(upstream.bodies[1], "stream_options.include_usage").Bool())
require.Equal(t, originalBody, body)
require.True(t, result.Stream)
require.Equal(t, mappedModel, result.BillingModel)
require.Equal(t, mappedModel, result.UpstreamModel)
require.Equal(t, 9, result.Usage.InputTokens)
require.Equal(t, 4, result.Usage.OutputTokens)
require.Equal(t, "/v1/chat/completions", result.UpstreamEndpoint)
require.Contains(t, rec.Header().Get("Content-Type"), "text/event-stream")
require.Contains(t, rec.Body.String(), "data: [DONE]")
})
}
}
func TestForwardAsChatCompletions_StructuredInputRetryIsBoundedAndPreservesRawErrors(t *testing.T) {
for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusTooManyRequests} {
t.Run(fmt.Sprintf("status=%d", status), func(t *testing.T) {
rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`
response := func(status int) *http.Response {
return &http.Response{StatusCode: status, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(rejection))}
}
body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`)
account := structuredInputRetryTestAccount()
account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true}
forward := func(retry bool) (*httptest.ResponseRecorder, error) {
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
upstream := &httpUpstreamRecorder{responses: []*http.Response{response(status)}}
if retry {
upstream.responses = append([]*http.Response{response(http.StatusBadRequest)}, upstream.responses...)
}
svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream}
var result *OpenAIForwardResult
var err error
if retry {
result, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "")
require.Len(t, upstream.requests, 2)
require.Equal(t, "/v1/chat/completions", upstream.requests[1].URL.Path)
} else {
result, err = svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "")
require.Len(t, upstream.requests, 1)
}
require.Nil(t, result)
require.Error(t, err)
return rec, err
}
baseline, baselineErr := forward(false)
retried, retryErr := forward(true)
require.Equal(t, baseline.Code, retried.Code)
require.Equal(t, baseline.Body.String(), retried.Body.String())
require.Equal(t, baselineErr.Error(), retryErr.Error())
require.IsType(t, baselineErr, retryErr)
})
}
}
func structuredInputRetryTestConfig() *config.Config {
return &config.Config{Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{
Enabled: false,
AllowInsecureHTTP: true,
}}}
}
func structuredInputRetryTestAccount() *Account {
return &Account{
ID: 101,
Name: "raw-openai-apikey",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "sk-test", "base_url": "http://upstream.example"},
Extra: map[string]any{openai_compat.ExtraKeyResponsesSupported: true},
}
}
type openAIChatStreamReadErrorCloser struct {
payload []byte
err error
@@ -1322,7 +1865,7 @@ func TestGPT61SolOnlyChatFallbackRejectsToolsAndDisabledReasoning(t *testing.T)
svc := &OpenAIGatewayService{cfg: &config.Config{}}
var err error
if responses {
_, err = svc.forwardResponsesViaRawChatCompletions(context.Background(), c, account, []byte(body))
_, err = svc.forwardResponsesViaRawChatCompletions(context.Background(), c, account, []byte(body), time.Now())
} else {
_, err = svc.forwardAnthropicViaRawChatCompletions(context.Background(), c, account, []byte(body), "")
}
@@ -190,7 +190,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
// 账号映射未命中时以去除首尾空白的请求模型兜底,计费名与上游模型名一致。
return s.forwardResponsesViaNativeAnthropic(ctx, c, account, body, reqModel)
case APIProtocolChatCompletions:
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body)
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body, startTime)
}
SetActualOpenAIUpstreamEndpoint(c, openAIResponsesUpstreamEndpoint)
if account.IsOpenAI() && (account.IsOpenAIApiKey() || account.IsOpenAIOAuthLike()) {
@@ -29,9 +29,8 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions(
c *gin.Context,
account *Account,
body []byte,
startTime time.Time,
) (*OpenAIForwardResult, error) {
startTime := time.Now()
var responsesReq apicompat.ResponsesRequest
if err := json.Unmarshal(body, &responsesReq); err != nil {
writeOpenAIResponsesFallbackError(c, http.StatusBadRequest, "invalid_request_error", "Failed to parse request body")
@@ -5,6 +5,7 @@ package service
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"net/http/httptest"
@@ -19,6 +20,34 @@ import (
"github.com/tidwall/gjson"
)
func TestForwardResponses_ChatFallbackPreservesProvidedStartTime(t *testing.T) {
for _, stream := range []bool{false, true} {
t.Run(fmt.Sprintf("stream=%t", stream), func(t *testing.T) {
body := []byte(fmt.Sprintf(`{"model":"gpt-5.4","input":"hello","stream":%t}`, stream))
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body))
responseBody := `{"id":"chatcmpl_timing","choices":[{"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`
contentType := "application/json"
if stream {
responseBody = "data: {\"id\":\"chatcmpl_timing\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: {\"choices\":[],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1,\"total_tokens\":2}}\n\ndata: [DONE]\n\n"
contentType = "text/event-stream"
}
upstream := &httpUpstreamRecorder{resp: &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{contentType}}, Body: io.NopCloser(strings.NewReader(responseBody))}}
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
startTime := time.Now().Add(-time.Minute)
result, err := svc.forwardResponsesViaRawChatCompletions(context.Background(), c, forceChatResponsesFallbackAccount(), body, startTime)
require.NoError(t, err)
require.NotNil(t, result)
require.GreaterOrEqual(t, result.Duration, time.Minute)
if stream {
require.NotNil(t, result.FirstTokenMs)
require.GreaterOrEqual(t, *result.FirstTokenMs, int(time.Minute.Milliseconds()))
}
})
}
}
func TestForwardResponses_ForceChatCompletionsRoutesNonStreamingToChatCompletions(t *testing.T) {
gin.SetMode(gin.TestMode)