mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-20 00:18:55 +00:00
fix(gateway): retry rejected structured input through chat endpoint
This commit is contained in:
wucm667
@@ -7,6 +7,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -431,6 +432,18 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
|
||||
)
|
||||
return s.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel)
|
||||
}
|
||||
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 &&
|
||||
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.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel)
|
||||
}
|
||||
if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil {
|
||||
return nil, foErr
|
||||
}
|
||||
@@ -483,6 +496,30 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
|
||||
return result, handleErr
|
||||
}
|
||||
|
||||
var responsesInputStringValidationError = regexp.MustCompile(`['"]loc['"]\s*:\s*\(\s*['"]body['"]\s*,\s*['"]input['"]\s*,\s*['"]str['"]\s*\)\s*,\s*['"]msg['"]\s*:\s*['"]Input should be a valid string['"]`)
|
||||
|
||||
func isConvertedResponsesInputStringRejection(status int, responseBody []byte) bool {
|
||||
if status != http.StatusBadRequest {
|
||||
return false
|
||||
}
|
||||
message := gjson.GetBytes(responseBody, "error.message").String()
|
||||
if responsesInputStringValidationError.MatchString(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 strings.Contains(message, "expected a string") &&
|
||||
(strings.Contains(message, "got an array") || strings.Contains(message, "got an object"))
|
||||
}
|
||||
|
||||
func normalizeResponsesRequestServiceTier(req *apicompat.ResponsesRequest) {
|
||||
if req == nil {
|
||||
return
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -13,6 +14,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 +34,290 @@ 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":"240 validation errors: ... '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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
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))
|
||||
|
||||
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 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
|
||||
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: "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'"}}`},
|
||||
} {
|
||||
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.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)
|
||||
require.True(t, strings.HasSuffix(upstream.requests[0].URL.Path, "/responses"))
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user