diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index bf52f4adc0..d0e2c6b730 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -477,12 +477,17 @@ func (h *GatewayHandler) Messages(c *gin.Context) { return } } - wroteFallback := h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := gatewayForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } forwardFailedFields := []zap.Field{ zap.Int64("account_id", account.ID), zap.String("account_name", account.Name), zap.String("account_platform", account.Platform), zap.Bool("fallback_error_response_written", wroteFallback), + zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated), zap.Error(err), } if account.Proxy != nil { @@ -874,12 +879,17 @@ func (h *GatewayHandler) Messages(c *gin.Context) { return } } - wroteFallback := h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := gatewayForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } forwardFailedFields := []zap.Field{ zap.Int64("account_id", account.ID), zap.String("account_name", account.Name), zap.String("account_platform", account.Platform), zap.Bool("fallback_error_response_written", wroteFallback), + zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated), zap.Error(err), } if account.Proxy != nil { @@ -1615,6 +1625,31 @@ func (h *GatewayHandler) ensureForwardErrorResponse(c *gin.Context, streamStarte return true } +// gatewayForwardErrorAlreadyCommunicated reports whether a Forward implementation +// has already written a complete error response to the client before returning +// an error to the handler. +// +// This is intentionally narrower than "writer size changed": a stream may have +// only emitted keepalive pings or partial data, in which case the handler still +// needs to append a protocol-level terminal error. Non-SSE output from Forward +// is different: service-level helpers such as handleErrorResponse/writeClaudeError +// already wrote the client-visible JSON body, so adding the generic streaming +// fallback would corrupt the response by appending a second `data: ...` frame. +func gatewayForwardErrorAlreadyCommunicated(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 + } + + contentType := strings.ToLower(strings.TrimSpace(c.Writer.Header().Get("Content-Type"))) + if contentType == "" { + return false + } + return !strings.Contains(contentType, "text/event-stream") +} + // checkClaudeCodeVersion 检查 Claude Code 客户端版本是否满足版本要求 // 仅对已识别的 Claude Code 客户端执行，count_tokens 路径除外 func (h *GatewayHandler) checkClaudeCodeVersion(c *gin.Context) bool { diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index fc034029c4..49347d02f0 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -281,9 +281,15 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { return } } - h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := gatewayForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } reqLog.Error("gateway.cc.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/gateway_handler_error_fallback_test.go b/backend/internal/handler/gateway_handler_error_fallback_test.go index fe9e2ebf44..40ba79f7f1 100644 --- a/backend/internal/handler/gateway_handler_error_fallback_test.go +++ b/backend/internal/handler/gateway_handler_error_fallback_test.go @@ -2,8 +2,10 @@ package handler import ( "encoding/json" + "errors" "net/http" "net/http/httptest" + "strings" "testing" "github.com/gin-gonic/gin" @@ -69,3 +71,97 @@ func TestGatewayEnsureForwardErrorResponse_ResponsesRouteAfterWrittenEmitsRespon assert.Contains(t, body, "event: response.failed\n") assert.Contains(t, body, `"type":"response.failed"`) } + +func TestGatewayForwardErrorAlreadyCommunicated(t *testing.T) { + gin.SetMode(gin.TestMode) + + t.Run("json error already written", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil) + before := c.Writer.Size() + c.JSON(http.StatusBadGateway, gin.H{ + "type": "error", + "error": gin.H{ + "type": "upstream_error", + "message": "Your Claude Code version (2.1.39) is below the minimum required version (2.1.81). Please update: npm update -g @anthropic-ai/claude-code", + }, + }) + + reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low")) + + require.True(t, reported) + body := w.Body.String() + assert.NotContains(t, body, `data: {"type":"error"`) + }) + + t.Run("sse ping still needs fallback", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil) + c.Header("Content-Type", "text/event-stream") + before := c.Writer.Size() + _, _ = c.Writer.WriteString(":\n\n") + + reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("stream read error: unexpected EOF")) + + require.False(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, EndpointMessages, nil) + + reported := gatewayForwardErrorAlreadyCommunicated(c, c.Writer.Size(), errors.New("upstream request failed")) + + require.False(t, reported) + }) + + // apikey 场景核心回归：复刻 GatewayService.handleErrorResponse 的 case 400 —— + // 原样透传上游 JSON body 后返回 err。此时错误已经完整告知客户端， + // handler 不得再追加 data:{"type":"error"} 帧，否则响应被污染成「JSON + 一行 data:」。 + t.Run("upstream 400 json passthrough via c.Data", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil) + before := c.Writer.Size() + upstreamBody := []byte(`{"type":"error","error":{"type":"upstream_error","message":"Your Claude Code version (2.1.39) is below the minimum required version (2.1.81). Please update: npm update -g @anthropic-ai/claude-code"}}`) + c.Data(http.StatusBadRequest, "application/json", upstreamBody) + + reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low")) + + require.True(t, reported) + body := w.Body.String() + assert.NotContains(t, body, `data: {"type":"error"`) + // 客户端只应收到上游那一份错误，没有被追加第二份。 + assert.Equal(t, 1, strings.Count(body, `"type":"error"`)) + }) + + // 流式已开始（已 flush 真实 SSE 事件，不只是 ping）+ 上游中途 400： + // HTTP 200 已固化，仍需 handler 补协议级终止帧，故不算「已完整告知」。 + t.Run("streaming 400 mid-stream still needs fallback", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil) + c.Header("Content-Type", "text/event-stream") + before := c.Writer.Size() + _, _ = c.Writer.WriteString("event: message_start\ndata: {\"type\":\"message_start\"}\n\n") + + reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low")) + + require.False(t, reported) + }) + + // 防御边界：err 为 nil 时永远不算「已告知」，避免在成功路径误吞兜底逻辑。 + t.Run("nil error never reports communicated", func(t *testing.T) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil) + c.JSON(http.StatusOK, gin.H{"ok": true}) + + reported := gatewayForwardErrorAlreadyCommunicated(c, 0, nil) + + require.False(t, reported) + }) +} diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index 3edecd7727..141c85ae61 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -260,9 +260,15 @@ func (h *GatewayHandler) Responses(c *gin.Context) { return } } - h.ensureForwardErrorResponse(c, streamStarted) + upstreamErrorAlreadyCommunicated := gatewayForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) + wroteFallback := false + if !upstreamErrorAlreadyCommunicated { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } reqLog.Error("gateway.responses.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