mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-19 16:17:56 +00:00
fix(gateway): preserve Responses semantics in chat fallback
This commit is contained in:
@@ -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 判定:命中时
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -413,7 +414,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 {
|
||||
@@ -430,19 +431,20 @@ 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 &&
|
||||
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.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel)
|
||||
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
|
||||
@@ -496,14 +498,107 @@ 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['"]`)
|
||||
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 {
|
||||
if status != http.StatusBadRequest || !json.Valid(responseBody) {
|
||||
return false
|
||||
}
|
||||
message := gjson.GetBytes(responseBody, "error.message").String()
|
||||
if responsesInputStringValidationError.MatchString(message) {
|
||||
if isResponsesInputStringValidationMessage(message) {
|
||||
return true
|
||||
}
|
||||
code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.code").String()))
|
||||
@@ -516,8 +611,7 @@ func isConvertedResponsesInputStringRejection(status int, responseBody []byte) b
|
||||
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"))
|
||||
return responsesInvalidInputTypeMessage.MatchString(message)
|
||||
}
|
||||
|
||||
func normalizeResponsesRequestServiceTier(req *apicompat.ResponsesRequest) {
|
||||
|
||||
@@ -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,6 +3,7 @@ package service
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -37,7 +38,7 @@ func (w *openAIChatFailingWriter) Write(p []byte) (int, error) {
|
||||
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'}]"}}`,
|
||||
"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)
|
||||
@@ -100,6 +101,7 @@ func TestForwardAsChatCompletions_ResponsesSupportedFallsBackWhenStructuredInput
|
||||
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
|
||||
@@ -124,6 +126,159 @@ func TestConvertedResponsesInputStringRejectionIsNarrow(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
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 {
|
||||
@@ -149,6 +304,8 @@ func TestForwardAsChatCompletions_StructuredInputRejectionDoesNotRetryIneligible
|
||||
{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}`)
|
||||
@@ -1608,7 +1765,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), "")
|
||||
}
|
||||
|
||||
@@ -167,7 +167,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
case APIProtocolResponses:
|
||||
break
|
||||
default:
|
||||
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body)
|
||||
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body, startTime)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -196,7 +196,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
}
|
||||
|
||||
if shouldForwardOpenAIResponsesViaRawChatCompletions(account) {
|
||||
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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user