opensource/sub2api
3
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-09-21 14:19:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2972 from northya/fix/chat-completions-response-failed

Browse Source 
fix: 修复 Chat Completions 兼容层误将 response.failed 返回为成功响应
This commit is contained in:
Wesley Liddick
2026-06-05 21:51:23 +08:00
committed by GitHub
parent 8775047f84 2e212d18e3
commit 1cecd2716c
2 changed files with 114 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/openai_gateway_chat_completions.go
+26 -1
View File
@@ -292,7 +292,7 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions(
if clientStream {
result, handleErr = s.handleChatStreamingResponse(resp, c, account, originalModel, billingModel, upstreamModel, startTime, len(body))
} else {
result, handleErr = s.handleChatBufferedStreamingResponse(resp, c, originalModel, billingModel, upstreamModel, startTime)
result, handleErr = s.handleChatBufferedStreamingResponse(resp, c, account, originalModel, billingModel, upstreamModel, startTime)
}
// Propagate ServiceTier and ReasoningEffort to result for billing
@@ -352,6 +352,13 @@ func normalizedOpenAIServiceTierValue(raw string) string {
return *normalized
}
func openAICompatFailedResponseMessage(resp *apicompat.ResponsesResponse) string {
if resp == nil || resp.Error == nil {
return ""
}
return strings.TrimSpace(resp.Error.Message)
}
// handleChatCompletionsErrorResponse reads an upstream error and returns it in
// OpenAI Chat Completions error format.
func (s *OpenAIGatewayService) handleChatCompletionsErrorResponse(
@@ -369,6 +376,7 @@ func (s *OpenAIGatewayService) handleChatCompletionsErrorResponse(
func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse(
resp *http.Response,
c *gin.Context,
account *Account,
originalModel string,
billingModel string,
upstreamModel string,
@@ -385,6 +393,10 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse(
writeChatCompletionsError(c, http.StatusBadGateway, "api_error", "Upstream stream ended without a terminal response event")
return nil, fmt.Errorf("upstream stream ended without terminal event")
}
if strings.TrimSpace(finalResponse.Status) == "failed" {
payload, _ := json.Marshal(gin.H{"type": "response.failed", "response": finalResponse})
return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, openAICompatFailedResponseMessage(finalResponse))
}
// When the terminal event has an empty output array, reconstruct from
// accumulated delta events so the client receives the full content.
@@ -451,6 +463,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
clientOutputStarted := false
pendingSSE := make([]string, 0, 4)
refusalDetector := newOpenAIChatSilentRefusalDetector(requestBodyLen)
var streamFailoverErr *UpstreamFailoverError
scanner := bufio.NewScanner(resp.Body)
maxLineSize := defaultMaxLineSize
@@ -512,6 +525,12 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
usage = copyOpenAIUsageFromResponsesUsage(event.Response.Usage)
}
}
if strings.TrimSpace(event.Type) == "response.failed" {
payloadBytes := []byte(payload)
message := extractOpenAISSEErrorMessage(payloadBytes)
streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message)
return true
}
chunks := apicompat.ResponsesEventToChatChunks(&event, state)
if !clientDisconnected {
@@ -562,6 +581,12 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
}
finalizeStream := func() (*OpenAIForwardResult, error) {
if streamFailoverErr != nil {
if c == nil || c.Writer == nil || !c.Writer.Written() {
return nil, streamFailoverErr
}
return resultWithUsage(), streamFailoverErr
}
if finalChunks := apicompat.FinalizeResponsesChatStream(state); len(finalChunks) > 0 && !clientDisconnected {
for _, chunk := range finalChunks {
refusalDetector.ObserveChatChunk(chunk)
backend/internal/service/openai_gateway_chat_completions_test.go
+88
View File
@@ -180,6 +180,94 @@ func TestForwardAsChatCompletions_ClientDisconnectDrainsUpstreamUsage(t *testing
require.Equal(t, 4, result.Usage.CacheReadInputTokens)
}
func TestForwardAsChatCompletions_BufferedResponseFailedTriggersFailover(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
body := []byte(`{"model":"gpt-5.5","messages":[{"role":"user","content":"large prompt"}],"stream":false}`)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
upstreamBody := strings.Join([]string{
`event: response.failed`,
`data: {"type":"response.failed","response":{"id":"resp_failed","object":"response","model":"gpt-5.5","status":"failed","output":[],"error":{"code":"upstream_error","message":"input exceeds the context window"}}}`,
"",
}, "\n")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_chat_failed_buffered"}},
Body: io.NopCloser(strings.NewReader(upstreamBody)),
}}
svc := &OpenAIGatewayService{httpUpstream: upstream}
account := &Account{
ID: 1,
Name: "openai-oauth",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
"access_token": "oauth-token",
"chatgpt_account_id": "chatgpt-acc",
},
}
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "gpt-5.5")
require.Error(t, err)
require.Nil(t, result)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode)
require.Contains(t, string(failoverErr.ResponseBody), "input exceeds the context window")
require.False(t, c.Writer.Written())
}
func TestForwardAsChatCompletions_StreamResponseFailedTriggersFailoverBeforeFlush(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
body := []byte(`{"model":"gpt-5.5","messages":[{"role":"user","content":"` + strings.Repeat("large prompt ", 6000) + `"}],"stream":true}`)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
upstreamBody := strings.Join([]string{
`data: {"type":"response.created","response":{"id":"resp_failed","model":"gpt-5.5","status":"in_progress","output":[]}}`,
"",
`event: response.failed`,
`data: {"type":"response.failed","response":{"id":"resp_failed","object":"response","model":"gpt-5.5","status":"failed","output":[],"error":{"code":"upstream_error","message":"input exceeds the context window"}}}`,
"",
}, "\n")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_chat_failed_stream"}},
Body: io.NopCloser(strings.NewReader(upstreamBody)),
}}
svc := &OpenAIGatewayService{httpUpstream: upstream}
account := &Account{
ID: 1,
Name: "openai-oauth",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
"access_token": "oauth-token",
"chatgpt_account_id": "chatgpt-acc",
},
}
result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "gpt-5.5")
require.Error(t, err)
require.Nil(t, result)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode)
require.Contains(t, string(failoverErr.ResponseBody), "input exceeds the context window")
require.False(t, c.Writer.Written())
}
func TestForwardAsChatCompletions_StreamsUsageWithoutClientStreamOptions(t *testing.T) {
gin.SetMode(gin.TestMode)