From 1da3501af5fb13f9005f10e4d630c210017aa2b2 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Thu, 9 Jul 2026 17:59:38 +0800 Subject: [PATCH] fix: apply error passthrough to OpenAI response.failed streams --- .../service/openai_gateway_passthrough.go | 124 +++++++++++++++- .../openai_gateway_response_handling.go | 71 +++++++-- .../service/openai_gateway_service_test.go | 136 ++++++++++++++++++ 3 files changed, 312 insertions(+), 19 deletions(-) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index c4e0702eb0..93588bdd69 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -614,6 +614,104 @@ func openAIStreamDataStartsClientOutput(data, eventType string) bool { return !openAIStreamEventIsPreamble(eventType) } +func openAIStreamFailedEventSemanticStatus(payload []byte, message string) int { + if isOpenAIContextWindowError(message, payload) { + return http.StatusBadRequest + } + + code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "response.error.code").String())) + if code == "" { + code = strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "error.code").String())) + } + errType := strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "response.error.type").String())) + if errType == "" { + errType = strings.ToLower(strings.TrimSpace(gjson.GetBytes(payload, "error.type").String())) + } + combined := strings.TrimSpace(errType + " " + code + " " + strings.ToLower(strings.TrimSpace(message))) + switch { + case strings.Contains(errType, "invalid_request"): + return http.StatusBadRequest + case strings.Contains(combined, "rate_limit"): + return http.StatusTooManyRequests + case strings.Contains(combined, "authentication") || strings.Contains(combined, "unauthorized") || strings.Contains(combined, "invalid_api_key"): + return http.StatusUnauthorized + case strings.Contains(combined, "permission") || strings.Contains(combined, "forbidden") || strings.Contains(combined, "access denied"): + return http.StatusForbidden + case code == "server_is_overloaded" || code == "slow_down": + return http.StatusServiceUnavailable + default: + return http.StatusBadGateway + } +} + +func openAIStreamFailedEventPassthroughBody(payload []byte, failedMessage string) []byte { + if len(payload) == 0 || !gjson.ValidBytes(payload) { + return payload + } + if gjson.GetBytes(payload, "error").Exists() { + return payload + } + responseError := gjson.GetBytes(payload, "response.error") + if !responseError.Exists() { + if strings.TrimSpace(failedMessage) == "" { + return payload + } + body, err := marshalOpenAIUpstreamJSON(gin.H{ + "error": gin.H{ + "message": failedMessage, + }, + }) + if err != nil { + return payload + } + return body + } + + errorPayload := gin.H{} + if errType := strings.TrimSpace(gjson.Get(responseError.Raw, "type").String()); errType != "" { + errorPayload["type"] = errType + } + if code := strings.TrimSpace(gjson.Get(responseError.Raw, "code").String()); code != "" { + errorPayload["code"] = code + } + if param := strings.TrimSpace(gjson.Get(responseError.Raw, "param").String()); param != "" { + errorPayload["param"] = param + } + message := strings.TrimSpace(gjson.Get(responseError.Raw, "message").String()) + if message == "" { + message = strings.TrimSpace(failedMessage) + } + if message != "" { + errorPayload["message"] = message + } + if len(errorPayload) == 0 { + return payload + } + body, err := marshalOpenAIUpstreamJSON(gin.H{"error": errorPayload}) + if err != nil { + return payload + } + return body +} + +func applyOpenAIStreamFailedErrorPassthroughRule( + c *gin.Context, + payload []byte, + failedMessage string, +) (status int, errType string, errMsg string, matched bool) { + ruleBody := openAIStreamFailedEventPassthroughBody(payload, failedMessage) + upstreamStatus := openAIStreamFailedEventSemanticStatus(payload, failedMessage) + return applyErrorPassthroughRule( + c, + PlatformOpenAI, + upstreamStatus, + ruleBody, + http.StatusBadGateway, + "upstream_error", + "Upstream request failed", + ) +} + func openAIStreamFailedEventShouldFailover(payload []byte, message string) bool { if isOpenAIContextWindowError(message, payload) { return false @@ -822,9 +920,23 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( UpstreamInTok: usage.InputTokens, UpstreamOutTok: usage.OutputTokens, }) - } else if !openAIStreamClientOutputStarted(c, clientOutputStarted) && openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { - return resultWithUsage(), - s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage) + } + if !openAIStreamClientOutputStarted(c, clientOutputStarted) { + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, dataBytes, failedMessage); matched { + MarkResponseCommitted(c) + c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8") + c.JSON(status, gin.H{ + "error": gin.H{ + "type": errType, + "message": errMsg, + }, + }) + return resultWithUsage(), fmt.Errorf("upstream response failed: passthrough rule matched message=%s", errMsg) + } + if openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { + return resultWithUsage(), + s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage) + } } forceFlushFailedEvent = true sawFailedEvent = true @@ -839,7 +951,11 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( responseID = extractOpenAIResponseIDFromJSONBytes(dataBytes) } imageCounter.AddSSEData(dataBytes) - if sanitizedData, sanitized := sanitizeOpenAIResponseFailedEventForClient(dataBytes, eventType); sanitized { + if sanitizedData, sanitized := sanitizeOpenAIResponseFailedEventForClient( + dataBytes, + eventType, + openAIStreamClientOutputStarted(c, clientOutputStarted), + ); sanitized { dataBytes = sanitizedData trimmedData = strings.TrimSpace(string(sanitizedData)) line = "data: " + string(sanitizedData) diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index d17fe410e0..51fc3a621b 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -124,7 +124,7 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp failedMessage := "" clientOutputStarted := false upstreamRequestID := strings.TrimSpace(resp.Header.Get("x-request-id")) - var streamFailoverErr error + var streamEarlyErr error sendErrorEvent := func(reason string) { if errorEventSent || clientDisconnected { return @@ -225,7 +225,7 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp return resultWithUsage(), fmt.Errorf("stream read error: %w", scanErr), true } processSSELine := func(line string, queueDrained bool) { - if streamFailoverErr != nil { + if streamEarlyErr != nil { return } // Extract data from SSE line (supports both "data: " and "data:" formats) @@ -254,10 +254,26 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp UpstreamInTok: usage.InputTokens, UpstreamOutTok: usage.OutputTokens, }) - } else if !openAIStreamClientOutputStarted(c, clientOutputStarted) && openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { - sawFailedEvent = true - streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage) - return + } + if !openAIStreamClientOutputStarted(c, clientOutputStarted) { + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, dataBytes, failedMessage); matched { + sawFailedEvent = true + MarkResponseCommitted(c) + c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8") + c.JSON(status, gin.H{ + "error": gin.H{ + "type": errType, + "message": errMsg, + }, + }) + streamEarlyErr = fmt.Errorf("upstream response failed: passthrough rule matched message=%s", errMsg) + return + } + if openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { + sawFailedEvent = true + streamEarlyErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage) + return + } } forceFlushFailedEvent = true sawFailedEvent = true @@ -286,7 +302,11 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp line = "data: " + data eventType = strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String()) } - if sanitizedData, sanitized := sanitizeOpenAIResponseFailedEventForClient(dataBytes, eventType); sanitized { + if sanitizedData, sanitized := sanitizeOpenAIResponseFailedEventForClient( + dataBytes, + eventType, + openAIStreamClientOutputStarted(c, clientOutputStarted), + ); sanitized { dataBytes = sanitizedData data = string(sanitizedData) line = "data: " + data @@ -356,8 +376,8 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp defer putSSEScannerBuf64K(scanBuf) for scanner.Scan() { processSSELine(scanner.Text(), true) - if streamFailoverErr != nil { - return resultWithUsage(), streamFailoverErr + if streamEarlyErr != nil { + return resultWithUsage(), streamEarlyErr } } if result, err, done := handleScanErr(scanner.Err()); done { @@ -408,8 +428,8 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp return result, err } processSSELine(ev.line, len(events) == 0) - if streamFailoverErr != nil { - return resultWithUsage(), streamFailoverErr + if streamEarlyErr != nil { + return resultWithUsage(), streamEarlyErr } case <-intervalCh: @@ -927,14 +947,35 @@ func extractOpenAISSEErrorMessage(payload []byte) string { return sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(payload))) } -func sanitizeOpenAIResponseFailedEventForClient(payload []byte, eventType string) ([]byte, bool) { +func sanitizeOpenAIResponseFailedEventForClient(payload []byte, eventType string, clientOutputStarted bool) ([]byte, bool) { if eventType != "response.failed" || len(payload) == 0 || !gjson.ValidBytes(payload) { return payload, false } - if !gjson.GetBytes(payload, "response").Exists() { - return payload, false - } updated := payload + if clientOutputStarted && isOpenAIContextWindowError(extractOpenAISSEErrorMessage(payload), payload) { + errorPath := "" + switch { + case gjson.GetBytes(updated, "response.error").Exists(): + errorPath = "response.error" + case gjson.GetBytes(updated, "error").Exists(): + errorPath = "error" + } + if errorPath != "" { + next, err := sjson.SetBytes(updated, errorPath+".type", "invalid_request_error") + if err != nil { + return payload, false + } + updated = next + next, err = sjson.SetBytes(updated, errorPath+".code", "context_length_exceeded") + if err != nil { + return payload, false + } + updated = next + } + } + if !gjson.GetBytes(updated, "response").Exists() { + return updated, !bytes.Equal(updated, payload) + } for _, path := range []string{ "response.instructions", "response.output", diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index b3e9889a7d..11d45d9dc5 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -14,6 +14,7 @@ import ( "time" "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/model" "github.com/Wei-Shaw/sub2api/internal/pkg/openai" "github.com/cespare/xxhash/v2" "github.com/gin-gonic/gin" @@ -1410,6 +1411,7 @@ func TestOpenAIStreamingResponseFailedAfterOutputSanitizesVerboseResponseForClie body := rec.Body.String() require.Contains(t, body, "event: response.failed") require.Contains(t, body, "context_length_exceeded") + require.Contains(t, body, `"type":"invalid_request_error"`) require.Contains(t, body, "Your input exceeds the context window") require.NotContains(t, body, "You are GPT-5.1 running in the Codex CLI") require.NotContains(t, body, `"instructions"`) @@ -1451,9 +1453,59 @@ func TestOpenAIStreamingContextWindowResponseFailedBeforeOutputPassesThrough(t * require.False(t, errors.As(err, &failoverErr)) require.True(t, c.Writer.Written()) require.Contains(t, rec.Body.String(), "response.failed") + require.Contains(t, rec.Body.String(), `"type":"upstream_error"`) require.Contains(t, rec.Body.String(), "Your input exceeds the context window") } +func TestOpenAIStreamingContextWindowResponseFailedBeforeOutputAppliesPassthroughRule(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + StreamDataIntervalTimeout: 0, + StreamKeepaliveInterval: 0, + MaxLineSize: defaultMaxLineSize, + }, + } + svc := &OpenAIGatewayService{cfg: cfg} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + rule := newNonFailoverPassthroughRule(http.StatusBadRequest, "context_length_exceeded", http.StatusBadRequest, "") + rule.Platforms = []string{PlatformOpenAI} + rule.PassthroughBody = true + rule.CustomMessage = nil + ruleSvc := &ErrorPassthroughService{} + ruleSvc.setLocalCache([]*model.ErrorPassthroughRule{rule}) + BindErrorPassthroughService(c, ruleSvc) + + upstreamMessage := "Your input exceeds the context window of this model. Please adjust your input and try again." + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_1","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"` + upstreamMessage + `"}}}`, + "", + }, "\n"))), + Header: http.Header{"X-Request-Id": []string{"rid-context-window-passthrough-rule"}}, + } + + _, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, &Account{ID: 1, Platform: PlatformOpenAI, Name: "acc"}, time.Now(), "model", "model") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + require.True(t, IsResponseCommitted(c)) + require.Equal(t, http.StatusBadRequest, rec.Code) + body := rec.Body.String() + require.Equal(t, "upstream_error", gjson.Get(body, "error.type").String()) + require.Equal(t, upstreamMessage, gjson.Get(body, "error.message").String()) + require.NotContains(t, body, "response.failed") + require.NotContains(t, body, "Upstream request failed") +} + func TestOpenAIStreamingPreambleOnlyMissingTerminalReturnsFailover(t *testing.T) { gin.SetMode(gin.TestMode) cfg := &config.Config{ @@ -1793,6 +1845,89 @@ func TestOpenAIStreamingPassthroughResponseFailedBeforeOutputReturnsFailover(t * require.Empty(t, rec.Body.String()) } +func TestOpenAIStreamingPassthroughContextWindowResponseFailedBeforeOutputAppliesPassthroughRule(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + MaxLineSize: defaultMaxLineSize, + }, + } + svc := &OpenAIGatewayService{cfg: cfg} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + rule := newNonFailoverPassthroughRule(http.StatusBadRequest, "input exceeds the context window", http.StatusBadRequest, "") + rule.Platforms = []string{PlatformOpenAI} + rule.PassthroughBody = true + rule.CustomMessage = nil + ruleSvc := &ErrorPassthroughService{} + ruleSvc.setLocalCache([]*model.ErrorPassthroughRule{rule}) + BindErrorPassthroughService(c, ruleSvc) + + upstreamMessage := "Your input exceeds the context window of this model. Please adjust your input and try again." + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_1","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"` + upstreamMessage + `"}}}`, + "", + }, "\n"))), + Header: http.Header{"X-Request-Id": []string{"rid-pass-context-window-passthrough-rule"}}, + } + + _, err := svc.handleStreamingResponsePassthrough(c.Request.Context(), resp, c, &Account{ID: 1, Platform: PlatformOpenAI, Name: "acc"}, time.Now(), "", "") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + require.True(t, IsResponseCommitted(c)) + require.Equal(t, http.StatusBadRequest, rec.Code) + body := rec.Body.String() + require.Equal(t, "upstream_error", gjson.Get(body, "error.type").String()) + require.Equal(t, upstreamMessage, gjson.Get(body, "error.message").String()) + require.NotContains(t, body, "response.failed") + require.NotContains(t, body, "Upstream request failed") +} + +func TestOpenAIStreamingPassthroughContextWindowResponseFailedBeforeOutputWithoutRulePassesThrough(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + MaxLineSize: defaultMaxLineSize, + }, + } + svc := &OpenAIGatewayService{cfg: cfg} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_1","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."}}}`, + "", + }, "\n"))), + Header: http.Header{"X-Request-Id": []string{"rid-pass-context-window-no-rule"}}, + } + + _, err := svc.handleStreamingResponsePassthrough(c.Request.Context(), resp, c, &Account{ID: 1, Platform: PlatformOpenAI, Name: "acc"}, time.Now(), "", "") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + body := rec.Body.String() + require.Contains(t, body, "event: response.failed") + require.Contains(t, body, "context_length_exceeded") + require.Contains(t, body, "Your input exceeds the context window") +} + func TestOpenAIStreamingPassthroughResponseFailedAfterOutputSanitizesVerboseResponseForClient(t *testing.T) { gin.SetMode(gin.TestMode) cfg := &config.Config{ @@ -1833,6 +1968,7 @@ func TestOpenAIStreamingPassthroughResponseFailedAfterOutputSanitizesVerboseResp body := rec.Body.String() require.Contains(t, body, "event: response.failed") require.Contains(t, body, "context_length_exceeded") + require.Contains(t, body, `"type":"invalid_request_error"`) require.Contains(t, body, "Your input exceeds the context window") require.NotContains(t, body, "You are GPT-5.1 running in the Codex CLI") require.NotContains(t, body, `"instructions"`)