From 2e212d18e3128c0506f0e1bf20e92980ca758668 Mon Sep 17 00:00:00 2001 From: northya Date: Tue, 2 Jun 2026 14:00:16 +0800 Subject: [PATCH] fix: handle failed responses in chat completions compat --- .../openai_gateway_chat_completions.go | 27 +++++- .../openai_gateway_chat_completions_test.go | 88 +++++++++++++++++++ 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 6e91d85c83..9a534c1bb3 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -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) diff --git a/backend/internal/service/openai_gateway_chat_completions_test.go b/backend/internal/service/openai_gateway_chat_completions_test.go index 9a5ea711c2..58aea1d716 100644 --- a/backend/internal/service/openai_gateway_chat_completions_test.go +++ b/backend/internal/service/openai_gateway_chat_completions_test.go @@ -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)