diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index d58656205d..bbccddf8d5 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -254,10 +254,15 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { continue } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) - wroteFallback := h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := openAIForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } reqLog.Warn("openai_chat_completions.forward_failed", zap.Int64("account_id", account.ID), zap.Bool("fallback_error_response_written", wroteFallback), + zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated), zap.Error(err), ) return diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 2a147f2b05..edf503847a 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -433,10 +433,15 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { continue } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) - wroteFallback := h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := openAIForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } fields := []zap.Field{ zap.Int64("account_id", account.ID), zap.Bool("fallback_error_response_written", wroteFallback), + zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated), zap.Error(err), } if shouldLogOpenAIForwardFailureAsWarn(c, wroteFallback) { @@ -1853,6 +1858,37 @@ func shouldLogOpenAIForwardFailureAsWarn(c *gin.Context, wroteFallback bool) boo return c.Writer.Written() } +// openAIForwardErrorAlreadyCommunicated reports whether Forward returned an +// error after it had already written the upstream terminal error response to +// the client. +// +// This matters for Responses streams: upstream may return HTTP 200 with a +// non-retryable `response.failed` event (for example a policy/safety rejection). +// The service layer forwards that terminal event verbatim, then returns an +// error so the caller can log/account for the failed upstream response. The +// handler must not append its generic fallback `response.failed`, otherwise +// strict clients may see the useful upstream message replaced by "Upstream +// request failed" or receive duplicate terminal events. +func openAIForwardErrorAlreadyCommunicated(c *gin.Context, writerSizeBeforeForward int, err error) bool { + if err == nil || c == nil || c.Writer == nil { + return false + } + if c.Writer.Size() == writerSizeBeforeForward { + return false + } + + msg := strings.TrimSpace(err.Error()) + for _, prefix := range []string{ + "upstream response failed:", + "non-streaming openai protocol error:", + } { + if strings.HasPrefix(msg, prefix) { + return true + } + } + return false +} + // errorResponse returns OpenAI API format error response func (h *OpenAIGatewayHandler) errorResponse(c *gin.Context, status int, errType, message string) { c.JSON(status, gin.H{ diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index b3fb35eee9..e7605e35e8 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -1579,3 +1579,44 @@ func runOpenAIResponsesWebSocketUsageLogCase(t *testing.T, tc openAIResponsesWSU func testStringPtr(v string) *string { return &v } + +func TestOpenAIForwardErrorAlreadyCommunicated(t *testing.T) { + gin.SetMode(gin.TestMode) + + t.Run("upstream response failed after write", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + before := c.Writer.Size() + _, _ = c.Writer.WriteString(`event: response.failed +data: {"type":"response.failed","error":{"message":"This content was flagged"}} + +`) + + reported := openAIForwardErrorAlreadyCommunicated(c, before, errors.New("upstream response failed: This content was flagged")) + + require.True(t, reported) + }) + + t.Run("no write still needs fallback", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + + reported := openAIForwardErrorAlreadyCommunicated(c, c.Writer.Size(), errors.New("upstream response failed: This content was flagged")) + + require.False(t, reported) + }) + + t.Run("generic error after write still needs fallback", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + before := c.Writer.Size() + _, _ = c.Writer.WriteString(":\n\n") + + reported := openAIForwardErrorAlreadyCommunicated(c, before, errors.New("stream read error: unexpected EOF")) + + require.False(t, reported) + }) +} diff --git a/backend/internal/handler/openai_images.go b/backend/internal/handler/openai_images.go index 580ecab992..bfa93de977 100644 --- a/backend/internal/handler/openai_images.go +++ b/backend/internal/handler/openai_images.go @@ -197,6 +197,7 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { service.SetOpsLatencyMs(c, service.OpsRoutingLatencyMsKey, time.Since(routingStart).Milliseconds()) forwardStart := time.Now() + writerSizeBeforeForward := c.Writer.Size() result, err := func() (*service.OpenAIForwardResult, error) { defer func() { if accountReleaseFunc != nil { @@ -277,10 +278,15 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { continue } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) - wroteFallback := h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := openAIForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } fields := []zap.Field{ zap.Int64("account_id", account.ID), zap.Bool("fallback_error_response_written", wroteFallback), + zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated), zap.Error(err), } if shouldLogOpenAIForwardFailureAsWarn(c, wroteFallback) {